"""
Pattern 056: Blackboard System
Description:
The Blackboard System pattern implements a shared workspace (blackboard) where
multiple independent knowledge sources (agents) collaboratively solve complex
problems. Each knowledge source monitors the blackboard and contributes when
it has relevant expertise, enabling opportunistic problem-solving through
emergent collaboration.
Components:
1. Blackboard: Shared data structure storing problem state
2. Knowledge Sources (KS): Independent expert agents
3. Control Module: Coordinates knowledge source activation
4. Problem Space: Hierarchical representation of problem
5. Solution Assembly: Combines partial solutions
Use Cases:
- Complex multi-disciplinary problems
- Speech and image recognition systems
- Medical diagnosis with multiple specialists
- Software architecture design
- Scientific research synthesis
- Strategic planning and decision-making
LangChain Implementation:
Uses a shared state dictionary as blackboard, with multiple LLM-based
knowledge sources that opportunistically contribute based on current state.
"""
import os
import time
from typing import Dict, Any, List, Optional, Callable, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import json
from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
load_dotenv()
class KSPriority(Enum):
"""Priority levels for knowledge sources"""
CRITICAL = 5
HIGH = 4
MEDIUM = 3
LOW = 2
MINIMAL = 1
class BlackboardState(Enum):
"""States of blackboard processing"""
INITIALIZED = "initialized"
GATHERING_INFO = "gathering_info"
ANALYZING = "analyzing"
GENERATING_SOLUTIONS = "generating_solutions"
REFINING = "refining"
COMPLETE = "complete"
@dataclass
class BlackboardEntry:
"""Entry on the blackboard"""
entry_id: str
layer: str # hypothesis, evidence, solution, etc.
content: str
contributor: str # which KS added this
confidence: float # 0.0-1.0
timestamp: datetime
dependencies: List[str] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
return {
"id": self.entry_id,
"layer": self.layer,
"content": self.content[:100] + "..." if len(self.content) > 100 else self.content,
"contributor": self.contributor,
"confidence": f"{self.confidence:.2f}",
"timestamp": self.timestamp.isoformat()
}
@dataclass
class KSActivation:
"""Knowledge source activation record"""
ks_name: str
trigger_condition: str
contribution: str
confidence: float
timestamp: datetime
@dataclass
class Blackboard:
"""Shared workspace for collaborative problem-solving"""
problem: str
state: BlackboardState
entries: List[BlackboardEntry] = field(default_factory=list)
activations: List[KSActivation] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
def add_entry(self, entry: BlackboardEntry):
"""Add entry to blackboard"""
self.entries.append(entry)
def get_entries_by_layer(self, layer: str) -> List[BlackboardEntry]:
"""Get all entries in a specific layer"""
return [e for e in self.entries if e.layer == layer]
def get_latest_entries(self, n: int = 5) -> List[BlackboardEntry]:
"""Get most recent entries"""
return sorted(self.entries, key=lambda x: x.timestamp, reverse=True)[:n]
def record_activation(self, activation: KSActivation):
"""Record knowledge source activation"""
self.activations.append(activation)
def get_state_summary(self) -> str:
"""Get summary of current blackboard state"""
layer_counts = {}
for entry in self.entries:
layer_counts[entry.layer] = layer_counts.get(entry.layer, 0) + 1
summary = f"Problem: {self.problem}\n"
summary += f"State: {self.state.value}\n"
summary += f"Total Entries: {len(self.entries)}\n"
summary += "Entries by Layer:\n"
for layer, count in layer_counts.items():
summary += f" - {layer}: {count}\n"
return summary
class KnowledgeSource:
"""Independent expert agent that contributes to blackboard"""
def __init__(
self,
name: str,
expertise: str,
trigger_conditions: List[str],
priority: KSPriority,
llm: ChatOpenAI,
system_prompt: str
):
self.name = name
self.expertise = expertise
self.trigger_conditions = trigger_conditions
self.priority = priority
self.llm = llm
self.system_prompt = system_prompt
self.activation_count = 0
def should_activate(self, blackboard: Blackboard) -> tuple[bool, str]:
"""Determine if this KS should contribute"""
# Check if any trigger conditions are met
bb_summary = blackboard.get_state_summary()
latest_entries = blackboard.get_latest_entries(3)
for condition in self.trigger_conditions:
# Simple keyword matching (could be more sophisticated)
if condition.lower() in bb_summary.lower():
return True, f"Triggered by: {condition}"
# Check recent entries
for entry in latest_entries:
if condition.lower() in entry.content.lower():
return True, f"Triggered by entry: {entry.entry_id}"
return False, ""
def contribute(self, blackboard: Blackboard) -> Optional[BlackboardEntry]:
"""Generate contribution to blackboard"""
# Get context from blackboard
recent_entries = blackboard.get_latest_entries(5)
context = "\n".join([
f"[{e.layer}] {e.content[:100]}"
for e in recent_entries
])
# Generate contribution
prompt = ChatPromptTemplate.from_messages([
("system", self.system_prompt),
("user", """Problem: {problem}
Current Blackboard State:
{context}
Based on your expertise in {expertise}, provide your contribution.
Be specific and build on existing information.
Your contribution:""")
])
chain = prompt | self.llm | StrOutputParser()
try:
contribution = chain.invoke({
"problem": blackboard.problem,
"context": context or "Empty blackboard",
"expertise": self.expertise
})
# Determine layer for contribution
layer = "analysis" # Default
if "hypothesis" in contribution.lower() or "theory" in contribution.lower():
layer = "hypothesis"
elif "evidence" in contribution.lower() or "data" in contribution.lower():
layer = "evidence"
elif "solution" in contribution.lower() or "approach" in contribution.lower():
layer = "solution"
elif "recommend" in contribution.lower():
layer = "recommendation"
# Create entry
entry = BlackboardEntry(
entry_id=f"{self.name}_{len(blackboard.entries)}",
layer=layer,
content=contribution,
contributor=self.name,
confidence=0.7 + (self.priority.value * 0.05),
timestamp=datetime.now()
)
self.activation_count += 1
return entry
except Exception as e:
print(f"Error in {self.name} contribution: {str(e)}")
return None
class BlackboardControl:
"""Controls knowledge source activation and coordination"""
def __init__(self, problem: str):
self.blackboard = Blackboard(
problem=problem,
state=BlackboardState.INITIALIZED
)
self.knowledge_sources: List[KnowledgeSource] = []
self.max_iterations = 10
self.current_iteration = 0
def register_knowledge_source(self, ks: KnowledgeSource):
"""Register a knowledge source"""
self.knowledge_sources.append(ks)
def _select_next_ks(self) -> Optional[KnowledgeSource]:
"""Select next knowledge source to activate"""
# Find all KS that can contribute
candidates = []
for ks in self.knowledge_sources:
should_activate, trigger = ks.should_activate(self.blackboard)
if should_activate:
candidates.append((ks, trigger))
if not candidates:
return None
# Sort by priority and activation count (prefer high priority, low activation)
candidates.sort(
key=lambda x: (x[0].priority.value, -x[0].activation_count),
reverse=True
)
return candidates[0][0]
def _update_state(self):
"""Update blackboard state based on progress"""
entry_count = len(self.blackboard.entries)
if entry_count == 0:
self.blackboard.state = BlackboardState.INITIALIZED
elif entry_count < 3:
self.blackboard.state = BlackboardState.GATHERING_INFO
elif entry_count < 6:
self.blackboard.state = BlackboardState.ANALYZING
elif entry_count Blackboard:
"""Execute blackboard problem-solving cycle"""
if max_iterations:
self.max_iterations = max_iterations
print(f"\nšÆ Starting blackboard problem-solving for: {self.blackboard.problem}")
while self.current_iteration = 2 and len(recommendations) >= 1:
print("\nSufficient solutions generated")
self.blackboard.state = BlackboardState.COMPLETE
break
return self.blackboard
def create_default_knowledge_sources() -> List[KnowledgeSource]:
"""Create default set of knowledge sources"""
knowledge_sources = []
# Problem Analyzer
ks1 = KnowledgeSource(
name="ProblemAnalyzer",
expertise="problem decomposition and analysis",
trigger_conditions=["initialized", "problem"],
priority=KSPriority.CRITICAL,
llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.3),
system_prompt="""You are a problem analysis expert. Break down complex problems
into components, identify key challenges, and frame the problem clearly."""
)
knowledge_sources.append(ks1)
# Hypothesis Generator
ks2 = KnowledgeSource(
name="HypothesisGenerator",
expertise="generating hypotheses and theories",
trigger_conditions=["analysis", "decomp", "component"],
priority=KSPriority.HIGH,
llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7),
system_prompt="""You are a creative thinker who generates hypotheses and theories.
Propose possible explanations and approaches based on the analysis."""
)
knowledge_sources.append(ks2)
# Evidence Collector
ks3 = KnowledgeSource(
name="EvidenceCollector",
expertise="gathering and evaluating evidence",
trigger_conditions=["hypothesis", "theory", "claim"],
priority=KSPriority.MEDIUM,
llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.2),
system_prompt="""You are an evidence expert who evaluates claims and hypotheses.
Identify what evidence would support or refute the hypotheses."""
)
knowledge_sources.append(ks3)
# Solution Designer
ks4 = KnowledgeSource(
name="SolutionDesigner",
expertise="designing practical solutions",
trigger_conditions=["evidence", "hypothesis", "analyzing"],
priority=KSPriority.HIGH,
llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.5),
system_prompt="""You are a solution architect who designs practical approaches.
Create concrete, actionable solutions based on the analysis and evidence."""
)
knowledge_sources.append(ks4)
# Critic
ks5 = KnowledgeSource(
name="Critic",
expertise="critical evaluation and refinement",
trigger_conditions=["solution", "approach", "recommend"],
priority=KSPriority.MEDIUM,
llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.4),
system_prompt="""You are a critical evaluator who identifies weaknesses and risks.
Critique proposed solutions and suggest improvements."""
)
knowledge_sources.append(ks5)
# Synthesizer
ks6 = KnowledgeSource(
name="Synthesizer",
expertise="synthesis and final recommendations",
trigger_conditions=["solution", "refining"],
priority=KSPriority.CRITICAL,
llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.3),
system_prompt="""You are a synthesizer who combines multiple perspectives.
Provide final recommendations that integrate all contributions."""
)
knowledge_sources.append(ks6)
return knowledge_sources
def demonstrate_blackboard_system():
"""Demonstrate Blackboard System pattern"""
print("=" * 80)
print("PATTERN 056: BLACKBOARD SYSTEM DEMONSTRATION")
print("=" * 80)
print("\nCollaborative problem-solving through shared workspace\n")
# Test 1: Technical problem
print("\n" + "=" * 80)
print("TEST 1: Technical Problem Solving")
print("=" * 80)
problem1 = "How can we improve the performance of a slow web application?"
control1 = BlackboardControl(problem1)
# Register knowledge sources
for ks in create_default_knowledge_sources():
control1.register_knowledge_source(ks)
print(f"\nRegistered {len(control1.knowledge_sources)} knowledge sources:")
for ks in control1.knowledge_sources:
print(f" - {ks.name}: {ks.expertise}")
# Solve problem
result1 = control1.solve(max_iterations=8)
print(f"\nš Final Blackboard State:")
print(f" Total Entries: {len(result1.entries)}")
print(f" Total Activations: {len(result1.activations)}")
print(f" Final State: {result1.state.value}")
# Show contributions by layer
print(f"\nš Entries by Layer:")
layers = {}
for entry in result1.entries:
if entry.layer not in layers:
layers[entry.layer] = []
layers[entry.layer].append(entry)
for layer, entries in layers.items():
print(f"\n {layer.upper()} ({len(entries)} entries):")
for entry in entries[:2]: # Show first 2 per layer
print(f" [{entry.contributor}] {entry.content[:100]}...")
# Test 2: Strategic planning problem
print("\n" + "=" * 80)
print("TEST 2: Strategic Planning Problem")
print("=" * 80)
problem2 = "What strategy should a startup use to compete with established companies?"
control2 = BlackboardControl(problem2)
for ks in create_default_knowledge_sources():
control2.register_knowledge_source(ks)
result2 = control2.solve(max_iterations=6)
print(f"\nšÆ Problem Solving Summary:")
print(f" Problem: {problem2}")
print(f" Iterations: {control2.current_iteration}")
print(f" Knowledge Sources Activated:")
# Count activations per KS
ks_counts = {}
for activation in result2.activations:
ks_counts[activation.ks_name] = ks_counts.get(activation.ks_name, 0) + 1
for ks_name, count in sorted(ks_counts.items(), key=lambda x: x[1], reverse=True):
print(f" {ks_name}: {count} times")
# Show final recommendation
recommendations = result2.get_entries_by_layer("recommendation")
if recommendations:
print(f"\nš” Final Recommendation:")
print(f" {recommendations[-1].content[:300]}...")
# Test 3: Show collaboration pattern
print("\n" + "=" * 80)
print("TEST 3: Collaboration Pattern Analysis")
print("=" * 80)
print(f"\nš Knowledge Source Activation Sequence:")
for i, activation in enumerate(result2.activations, 1):
print(f" {i}. {activation.ks_name} (confidence: {activation.confidence:.2f})")
print(f"\nšļø Blackboard Evolution:")
for i in range(0, len(result2.entries), 2):
entry = result2.entries[i]
print(f" Entry {i+1}: [{entry.layer}] by {entry.contributor}")
# Summary
print("\n" + "=" * 80)
print("BLACKBOARD SYSTEM PATTERN SUMMARY")
print("=" * 80)
print("""
Key Benefits:
1. Opportunistic Problem-Solving: KS contribute when relevant
2. Emergent Collaboration: No predefined workflow
3. Modularity: Independent knowledge sources
4. Incremental Progress: Builds solution step-by-step
5. Transparency: All contributions visible on blackboard
System Components:
1. Blackboard: Shared workspace with layered structure
2. Knowledge Sources: Independent expert agents
3. Control Module: Coordinates KS activation
4. Activation Conditions: Triggers for KS contributions
5. Priority System: Manages contribution order
Blackboard Layers:
- Hypothesis: Theories and possible explanations
- Evidence: Supporting or refuting data
- Analysis: Problem decomposition and insights
- Solution: Concrete approaches and methods
- Recommendation: Final synthesized advice
Knowledge Source Types:
1. Problem Analyzer: Breaks down problems
2. Hypothesis Generator: Creates theories
3. Evidence Collector: Gathers supporting data
4. Solution Designer: Designs approaches
5. Critic: Evaluates and refines
6. Synthesizer: Combines perspectives
Control Strategies:
1. Priority-Based: High-priority KS activate first
2. Trigger-Based: KS activate when conditions met
3. Load-Balanced: Distribute activations evenly
4. State-Driven: Blackboard state guides activation
Activation Patterns:
- Initial Analysis: Problem decomposition
- Hypothesis Generation: Multiple theories
- Evidence Collection: Supporting data
- Solution Design: Practical approaches
- Critical Evaluation: Refinement
- Synthesis: Final recommendations
Use Cases:
- Complex multi-disciplinary problems
- Medical diagnosis systems
- Scientific research synthesis
- Software architecture design
- Strategic planning
- Signal processing and recognition
Best Practices:
1. Clear KS expertise boundaries
2. Well-defined trigger conditions
3. Appropriate priority levels
4. Layered blackboard structure
5. Limit iteration count
6. Record all contributions
7. Final synthesis step
Production Considerations:
- Blackboard persistence
- Concurrent KS activation
- Conflict resolution
- Termination criteria
- Performance optimization
- Scalability with many KS
- Debugging and tracing
Comparison with Related Patterns:
- vs. Multi-Agent: Shared state vs communication
- vs. Pipeline: Opportunistic vs sequential
- vs. Mixture of Agents: Collaboration vs aggregation
- vs. Ensemble: Emergent vs predetermined combination
The Blackboard System pattern excels at complex problems requiring
multiple perspectives and emergent, opportunistic collaboration where
the solution path is not predetermined.
""")
if __name__ == "__main__":
demonstrate_blackboard_system()
Agentic Design Pattern: Blackboard System
Pattern 056: Blackboard System – Line-by-Line Explanation
Overview
This document provides a detailed line-by-line explanation of the Blackboard System pattern implementation.
Pattern Description
Pattern 056: Blackboard System
Description: The Blackboard System pattern implements a shared workspace (blackboard) where multiple independent knowledge sources (agents) collaboratively solve complex problems. Each knowledge source monitors the blackboard and contributes when it has relevant expertise, enabling opportunistic problem-solving through emergent collaboration.
Components: 1. Blackboard: Shared data structure storing problem state 2. Knowledge Sources (KS): Independent expert agents 3. Control Module: Coordinates knowledge source activation 4. Problem Space: Hierarchical representation of problem 5. Solution Assembly: Combines partial solutions
Use Cases: – Complex multi-disciplinary problems – Speech and image recognition systems – Medical diagnosis with multiple specialists – Software architecture design – Scientific research synthesis – Strategic planning and decision-making
LangChain Implementation: Uses a shared state dictionary as blackboard, with multiple LLM-based knowledge sources that opportunistically contribute based on current state.
File Statistics
- Total Lines: 611
- Classes: 7
- Functions: 2
- Constants: 0
Module Documentation
"""
Pattern 056: Blackboard System
Description:
The Blackboard System pattern implements a shared workspace (blackboard) where
multiple independent knowledge sources (agents) collaboratively solve complex
problems. Each knowledge source monitors the blackboard and contributes when
it has relevant expertise, enabling opportunistic problem-solving through
emergent collaboration.
Components:
1. Blackboard: Shared data structure storing problem state
2. Knowledge Sources (KS): Independent expert agents
3. Control Module: Coordinates knowledge source activation
4. Problem Space: Hierarchical representation of problem
5. Solution Assembly: Combines partial solutions
Use Cases:
- Complex multi-disciplinary problems
- Speech and image recognition systems
- Medical diagnosis with multiple specialists
- Software architecture design
- Scientific research synthesis
- Strategic planning and decision-making
LangChain Implementation:
Uses a shared state dictionary as blackboard, with multiple LLM-based
knowledge sources that opportunistically contribute based on current state.
"""
Purpose: This module implements the Blackboard System pattern, providing a structured approach to agent design and implementation.
Import Statements
Standard Library Imports
import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import enum.Enum
import json
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import time
import typing.Any
import typing.Callable
import typing.Dict
import typing.List
import typing.Optional
import typing.Set
Purpose: Core Python functionality:
– dataclasses.dataclass: Simplified class definitions with @dataclass
– dataclasses.field: Simplified class definitions with @dataclass
– datetime.datetime: Date and time handling
– enum.Enum: Enumeration support
– json: JSON serialization/deserialization
– langchain_core.output_parsers.StrOutputParser: Regular expression operations
– langchain_core.prompts.ChatPromptTemplate: Regular expression operations
– os: Environment variable access and system operations
– time: Time-related functions
– typing.Any: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.Callable: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.List: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional)
– typing.Set: Type hints for better code documentation (List, Dict, Any, Optional)
LangChain Imports
from langchain_openai.ChatOpenAI
Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows
Other Imports
from dotenv.load_dotenv
Purpose: Additional dependencies for specialized functionality.
Classes
Class: KSPriority
Lines 47-53
Inherits from: Enum
Purpose:
Priority levels for knowledge sources
Class: BlackboardState
Lines 56-63
Inherits from: Enum
Purpose:
States of blackboard processing
Class: BlackboardEntry
Lines 67-86
Purpose:
Entry on the blackboard
Methods:
to_dict(self)
Lines 78-86
Purpose: To Dict operation.
Parameters:
Returns: <ast.Subscript object at 0x000001A06D7664D0>
Class: KSActivation
Lines 90-96
Purpose:
Knowledge source activation record
Class: Blackboard
Lines 100-137
Purpose:
Shared workspace for collaborative problem-solving
Methods:
add_entry(self, entry)
Lines 108-110
Purpose:
Add entry to blackboard
Parameters:
– entry: Entry
get_entries_by_layer(self, layer)
Lines 112-114
Purpose:
Get all entries in a specific layer
Parameters:
– layer: Layer
Returns: <ast.Subscript object at 0x000001A06D6F48D0>
get_latest_entries(self, n)
Lines 116-118
Purpose:
Get most recent entries
Parameters:
– n: N
Returns: <ast.Subscript object at 0x000001A06D6F5F90>
record_activation(self, activation)
Lines 120-122
Purpose:
Record knowledge source activation
Parameters:
– activation: Activation
get_state_summary(self)
Lines 124-137
Purpose:
Get summary of current blackboard state
Parameters:
Returns: str
Class: KnowledgeSource
Lines 140-239
Purpose:
Independent expert agent that contributes to blackboard
Methods:
__init__(self, name, expertise, trigger_conditions, priority, llm, system_prompt)
Lines 143-158
Purpose: Initialize the class instance with required parameters.
Parameters:
– name: Name
– expertise: Expertise
– trigger_conditions: Trigger conditions
– priority: Priority
– llm: Language model
– system_prompt: Prompt template or text
should_activate(self, blackboard)
Lines 160-177
Purpose:
Determine if this KS should contribute
Parameters:
– blackboard: Blackboard
Returns: <ast.Subscript object at 0x000001A06D16D310>
contribute(self, blackboard)
Lines 179-239
Purpose:
Generate contribution to blackboard
Parameters:
– blackboard: Blackboard
Returns: <ast.Subscript object at 0x000001A06D626810>
Class: BlackboardControl
Lines 242-348
Purpose:
Controls knowledge source activation and coordination
Methods:
__init__(self, problem)
Lines 245-252
Purpose: Initialize the class instance with required parameters.
Parameters:
– problem: Problem
register_knowledge_source(self, ks)
Lines 254-256
Purpose:
Register a knowledge source
Parameters:
– ks: Ks
_select_next_ks(self)
Lines 258-278
Purpose:
Select next knowledge source to activate
Parameters:
Returns: <ast.Subscript object at 0x000001A06D60B210>
_update_state(self)
Lines 280-294
Purpose:
Update blackboard state based on progress
Parameters:
solve(self, max_iterations)
Lines 296-348
Purpose:
Execute blackboard problem-solving cycle
Parameters:
– max_iterations: Max iterations
Returns: Blackboard
Functions
create_default_knowledge_sources()
Lines 351-428
Purpose:
Create default set of knowledge sources
Returns: <ast.Subscript object at 0x000001A06D60CC90>
demonstrate_blackboard_system()
Lines 431-606
Purpose:
Demonstrate Blackboard System pattern
Key Concepts Summary
Pattern Overview
The Blackboard System pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.
Core Components
- Agent Class: Main implementation of the pattern logic
- Prompt Engineering: Carefully crafted prompts to guide LLM behavior
- Tool Integration: Optional tools for extended functionality
- Memory Management: State handling across interactions
- Output Parsing: Structured response processing
When to Use
ā Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability
ā Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)
Best Practices Demonstrated
1. Environment Configuration
- ā
Use
.envfiles for API keys - ā Never hardcode credentials
- ā Validate environment variables on startup
2. Error Handling
- ā Graceful degradation on failures
- ā Informative error messages
- ā Logging for debugging
3. Prompt Engineering
- ā Clear, specific instructions
- ā Structured output formats
- ā Few-shot examples when beneficial
4. Code Organization
- ā Separation of concerns (class-based design)
- ā Type hints for clarity
- ā Docstrings for documentation
5. Performance Optimization
- ā Appropriate temperature settings
- ā Token budget management
- ā Caching where applicable
LangChain Components Used
- ChatOpenAI: OpenAI model interface
- Chains: Sequential component composition
Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development
Execution Flow
1. Initialize agent with configuration
ā
2. Load environment and dependencies
ā
3. Prepare prompt with input data
ā
4. Invoke LLM through LangChain
ā
5. Parse and validate response
ā
6. Return formatted result
Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation
Performance Considerations
Computational Cost
- LLM Calls: Primary cost factor (based on tokens)
- Latency: Depends on model and input size
- Memory: Minimal for most patterns
Optimization Strategies
- Prompt Optimization: Minimize tokens while maintaining clarity
- Caching: Cache repeated queries/responses
- Model Selection: Use smaller models when appropriate
- Batch Processing: Group similar requests
- Streaming: For long responses, stream tokens
Cost Estimates (Approximate)
- GPT-3.5-turbo: ~$0.002 per request
- GPT-4: ~$0.03 per request
- GPT-4-turbo: ~$0.01 per request
Actual costs vary based on input/output token counts
Common Pitfalls to Avoid
1. Prompt Design Issues
ā Vague or ambiguous instructions
ā
Clear, specific, testable prompts
2. Error Handling
ā Ignoring API errors or rate limits
ā
Implement retry logic and graceful degradation
3. Token Management
ā Exceeding model token limits
ā
Truncate inputs, chunk processing
4. Security
ā Exposing API keys in code
ā
Use environment variables
5. Testing
ā Only testing happy paths
ā
Test edge cases, failures, malformed inputs
6. Cost Management
ā Unmonitored API usage
ā
Implement budgets, alerts, and monitoring
Further Improvements
Production Enhancements
- Monitoring: Add logging, metrics, tracing
- Validation: Input/output schema validation
- Rate Limiting: Implement request throttling
- Caching: Redis or in-memory cache for responses
- Async Support: Non-blocking operations for scalability
Feature Extensions
- Multi-Model Support: Switch between providers
- Fallback Models: Graceful degradation strategy
- Custom Tools: Domain-specific function calling
- Advanced Memory: Vector stores for long-term memory
- Evaluation: Automated testing and benchmarking
Code Quality
- Testing: Unit tests, integration tests
- Documentation: API docs, usage examples
- Type Safety: Stricter type hints
- Configuration: Externalized config files
Related Patterns
- Pattern 055
- Pattern 057
Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…
Summary
The Blackboard System pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:
ā Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns
šÆ Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability
š Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/
File: 056_blackboard_system.py
Total Lines: 611
Pattern Category: AI Agent Design Pattern #056
Generated from actual implementation code
For educational and reference purposes
Leave a Reply