"""
Pattern 124: Ontology-Based Reasoning

Description:
    Uses formal ontologies (OWL, RDF, SKOS) for domain knowledge
    representation and semantic reasoning.

Components:
    - Ontology definition
    - Class hierarchies
    - Property definitions
    - Inference rules

Use Cases:
    - Semantic reasoning
    - Domain expertise
    - Formal logic

LangChain Implementation:
    Implements ontology concepts with LLM-based reasoning over formal structures.
"""

import os
from typing import List, Dict, Any, Set
from dataclasses import dataclass
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()


@dataclass
class OntologyClass:
    """Represents a class in the ontology."""
    name: str
    parent: str = None
    properties: List[str] = None
    description: str = ""
    
    def __post_init__(self):
        if self.properties is None:
            self.properties = []


@dataclass
class OntologyProperty:
    """Represents a property in the ontology."""
    name: str
    domain: str  # Class it belongs to
    range: str   # Type of value
    description: str = ""


@dataclass
class OntologyInstance:
    """Represents an instance of a class."""
    id: str
    class_name: str
    properties: Dict[str, Any]


class Ontology:
    """Formal ontology for domain knowledge."""
    
    def __init__(self, name: str):
        self.name = name
        self.classes: Dict[str, OntologyClass] = {}
        self.properties: Dict[str, OntologyProperty] = {}
        self.instances: Dict[str, OntologyInstance] = {}
        
    def add_class(self, ontology_class: OntologyClass):
        """Add a class to the ontology."""
        self.classes[ontology_class.name] = ontology_class
        
    def add_property(self, ontology_property: OntologyProperty):
        """Add a property to the ontology."""
        self.properties[ontology_property.name] = ontology_property
        
    def add_instance(self, instance: OntologyInstance):
        """Add an instance to the ontology."""
        self.instances[instance.id] = instance
        
    def get_subclasses(self, class_name: str) -> List[str]:
        """Get all subclasses of a class."""
        subclasses = []
        for cls_name, cls in self.classes.items():
            if cls.parent == class_name:
                subclasses.append(cls_name)
                # Recursively get subclasses
                subclasses.extend(self.get_subclasses(cls_name))
        return subclasses
    
    def get_superclasses(self, class_name: str) -> List[str]:
        """Get all superclasses of a class."""
        superclasses = []
        current = self.classes.get(class_name)
        while current and current.parent:
            superclasses.append(current.parent)
            current = self.classes.get(current.parent)
        return superclasses
    
    def is_subclass_of(self, class1: str, class2: str) -> bool:
        """Check if class1 is a subclass of class2."""
        return class2 in self.get_superclasses(class1)
    
    def get_instances_of_class(self, class_name: str, include_subclasses: bool = True) -> List[OntologyInstance]:
        """Get all instances of a class."""
        instances = [inst for inst in self.instances.values() if inst.class_name == class_name]
        
        if include_subclasses:
            subclasses = self.get_subclasses(class_name)
            for subclass in subclasses:
                instances.extend([inst for inst in self.instances.values() if inst.class_name == subclass])
        
        return instances


class OntologyReasoningAgent:
    """Agent that performs reasoning using ontologies."""
    
    def __init__(self, ontology: Ontology, model_name: str = "gpt-4"):
        self.ontology = ontology
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        
    def classify_instance(self, instance_description: str) -> str:
        """Classify an instance into ontology classes."""
        # Get ontology structure
        ontology_structure = "Ontology Classes:\n"
        for cls_name, cls in self.ontology.classes.items():
            ontology_structure += f"- {cls_name}"
            if cls.parent:
                ontology_structure += f" (subclass of {cls.parent})"
            ontology_structure += f": {cls.description}\n"
        
        classification_prompt = ChatPromptTemplate.from_messages([
            ("system", "Classify this instance into the most appropriate ontology class."),
            ("user", """{ontology_structure}

Instance to classify: {instance}

Most appropriate class:""")
        ])
        
        chain = classification_prompt | self.llm | StrOutputParser()
        return chain.invoke({
            "ontology_structure": ontology_structure,
            "instance": instance_description
        })
    
    def infer_properties(self, instance_id: str) -> Dict[str, Any]:
        """Infer properties of an instance based on ontology."""
        instance = self.ontology.instances.get(instance_id)
        if not instance:
            return {}
        
        # Get class hierarchy
        superclasses = self.ontology.get_superclasses(instance.class_name)
        
        # Collect inherited properties
        inherited_properties = {}
        for superclass in superclasses:
            cls = self.ontology.classes.get(superclass)
            if cls:
                for prop in cls.properties:
                    if prop not in instance.properties:
                        inherited_properties[prop] = f"inherited from {superclass}"
        
        inference_prompt = ChatPromptTemplate.from_messages([
            ("system", "Infer additional properties for this instance based on ontology."),
            ("user", """Instance: {instance_id}
Class: {class_name}
Superclasses: {superclasses}
Current Properties: {current_props}

Infer likely values for these properties:
{inherited_props}""")
        ])
        
        chain = inference_prompt | self.llm | StrOutputParser()
        inferences = chain.invoke({
            "instance_id": instance_id,
            "class_name": instance.class_name,
            "superclasses": ", ".join(superclasses),
            "current_props": str(instance.properties),
            "inherited_props": ", ".join(inherited_properties.keys())
        })
        
        return {"inferences": inferences, "inherited": inherited_properties}
    
    def answer_semantic_query(self, query: str) -> str:
        """Answer semantic queries using ontology."""
        # Serialize ontology info
        ontology_info = f"Ontology: {self.ontology.name}\n\n"
        ontology_info += "Classes:\n"
        for cls_name, cls in self.ontology.classes.items():
            ontology_info += f"- {cls_name}: {cls.description}\n"
        
        ontology_info += "\nInstances:\n"
        for inst_id, inst in list(self.ontology.instances.items())[:10]:  # Limit to 10
            ontology_info += f"- {inst_id} ({inst.class_name}): {inst.properties}\n"
        
        query_prompt = ChatPromptTemplate.from_messages([
            ("system", "Answer this semantic query using the ontology knowledge."),
            ("user", """{ontology_info}

Query: {query}

Answer:""")
        ])
        
        chain = query_prompt | self.llm | StrOutputParser()
        return chain.invoke({
            "ontology_info": ontology_info,
            "query": query
        })
    
    def check_consistency(self) -> str:
        """Check ontology consistency."""
        consistency_prompt = ChatPromptTemplate.from_messages([
            ("system", "Check this ontology for logical consistency and conflicts."),
            ("user", """Ontology Structure:
{structure}

Check for:
1. Circular class hierarchies
2. Property domain/range conflicts
3. Instance classification errors
4. Missing definitions""")
        ])
        
        structure = ""
        for cls_name, cls in self.ontology.classes.items():
            structure += f"{cls_name} -> {cls.parent}\n"
        
        chain = consistency_prompt | self.llm | StrOutputParser()
        return chain.invoke({"structure": structure})


def demonstrate_ontology_based_reasoning():
    """Demonstrate ontology-based reasoning pattern."""
    print("=== Ontology-Based Reasoning Pattern ===\n")
    
    # Create ontology
    print("1. Building Domain Ontology")
    print("-" * 50)
    
    ontology = Ontology("VehicleOntology")
    
    # Define class hierarchy
    ontology.add_class(OntologyClass(
        name="Vehicle",
        description="Any mode of transportation"
    ))
    
    ontology.add_class(OntologyClass(
        name="LandVehicle",
        parent="Vehicle",
        description="Vehicles that travel on land"
    ))
    
    ontology.add_class(OntologyClass(
        name="WaterVehicle",
        parent="Vehicle",
        description="Vehicles that travel on water"
    ))
    
    ontology.add_class(OntologyClass(
        name="AirVehicle",
        parent="Vehicle",
        description="Vehicles that travel in air"
    ))
    
    ontology.add_class(OntologyClass(
        name="Car",
        parent="LandVehicle",
        properties=["num_wheels", "fuel_type", "num_doors"],
        description="Four-wheeled motor vehicle"
    ))
    
    ontology.add_class(OntologyClass(
        name="Bicycle",
        parent="LandVehicle",
        properties=["num_wheels", "has_motor"],
        description="Two-wheeled human-powered vehicle"
    ))
    
    ontology.add_class(OntologyClass(
        name="Boat",
        parent="WaterVehicle",
        properties=["propulsion_type", "max_speed"],
        description="Watercraft of any size"
    ))
    
    # Add instances
    ontology.add_instance(OntologyInstance(
        id="vehicle1",
        class_name="Car",
        properties={"num_wheels": 4, "fuel_type": "electric", "num_doors": 4}
    ))
    
    ontology.add_instance(OntologyInstance(
        id="vehicle2",
        class_name="Bicycle",
        properties={"num_wheels": 2, "has_motor": False}
    ))
    
    print(f"✓ Created ontology with {len(ontology.classes)} classes")
    print(f"✓ Added {len(ontology.instances)} instances")
    print()
    
    # Create reasoning agent
    agent = OntologyReasoningAgent(ontology)
    
    # Classification
    print("2. Instance Classification")
    print("-" * 50)
    
    test_descriptions = [
        "A motorized two-wheeler used for transportation",
        "A watercraft with sails",
        "A four-wheeled vehicle with an engine"
    ]
    
    for desc in test_descriptions:
        print(f"\nDescription: {desc}")
        classification = agent.classify_instance(desc)
        print(f"Classification: {classification}")
    
    # Property inference
    print("\n\n3. Property Inference")
    print("-" * 50)
    
    inferred = agent.infer_properties("vehicle1")
    print(f"\nInferred properties for vehicle1:")
    print(inferred)
    
    # Semantic queries
    print("\n\n4. Semantic Queries")
    print("-" * 50)
    
    queries = [
        "What are all the types of land vehicles?",
        "What properties do cars have?",
        "How are bicycles and cars related?"
    ]
    
    for query in queries:
        print(f"\nQuery: {query}")
        answer = agent.answer_semantic_query(query)
        print(f"Answer: {answer}")
    
    # Consistency check
    print("\n\n5. Ontology Consistency Check")
    print("-" * 50)
    consistency = agent.check_consistency()
    print(consistency)
    
    print("\n=== Summary ===")
    print("Ontology-based reasoning demonstrated with:")
    print("- Formal class hierarchies")
    print("- Property definitions")
    print("- Instance classification")
    print("- Property inference")
    print("- Semantic queries")
    print("- Consistency checking")


if __name__ == "__main__":
    demonstrate_ontology_based_reasoning()

    

Agentic Design Pattern: Ontology Based Reasoning

Pattern 124: Ontology Based Reasoning – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Ontology Based Reasoning pattern implementation.

---

Pattern Description

Pattern 124: Ontology-Based Reasoning

Description: Uses formal ontologies (OWL, RDF, SKOS) for domain knowledge representation and semantic reasoning.

Components: – Ontology definition – Class hierarchies – Property definitions – Inference rules

Use Cases: – Semantic reasoning – Domain expertise – Formal logic

LangChain Implementation: Implements ontology concepts with LLM-based reasoning over formal structures.

---

File Statistics

• Total Lines: 373
• Classes: 5
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 124: Ontology-Based Reasoning

Description:
    Uses formal ontologies (OWL, RDF, SKOS) for domain knowledge
    representation and semantic reasoning.

Components:
    - Ontology definition
    - Class hierarchies
    - Property definitions
    - Inference rules

Use Cases:
    - Semantic reasoning
    - Domain expertise
    - Formal logic

LangChain Implementation:
    Implements ontology concepts with LLM-based reasoning over formal structures.
"""

Purpose: This module implements the Ontology Based Reasoning pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Set

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: 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.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: OntologyClass

Lines 36-45

Purpose:

Represents a class in the ontology.

Methods:

__post_init__(self)

Lines 43-45

Purpose: Post Init operation.

Parameters:

---

Class: OntologyProperty

Lines 49-54

Purpose:

Represents a property in the ontology.

---

Class: OntologyInstance

Lines 58-62

Purpose:

Represents an instance of a class.

---

Class: Ontology

Lines 65-118

Purpose:

Formal ontology for domain knowledge.

Methods:

__init__(self, name)

Lines 68-72

Purpose: Initialize the class instance with required parameters.

Parameters: – name: Name

add_class(self, ontology_class)

Lines 74-76

Purpose:

Add a class to the ontology.

Parameters: – ontology_class: Ontology class

add_property(self, ontology_property)

Lines 78-80

Purpose:

Add a property to the ontology.

Parameters: – ontology_property: Ontology property

add_instance(self, instance)

Lines 82-84

Purpose:

Add an instance to the ontology.

Parameters: – instance: Instance

get_subclasses(self, class_name)

Lines 86-94

Purpose:

Get all subclasses of a class.

Parameters: – class_name: Class name

Returns: <ast.Subscript object at 0x000001A06D749290>

get_superclasses(self, class_name)

Lines 96-103

Purpose:

Get all superclasses of a class.

Parameters: – class_name: Class name

Returns: <ast.Subscript object at 0x000001A06D64BF50>

is_subclass_of(self, class1, class2)

Lines 105-107

Purpose:

Check if class1 is a subclass of class2.

Parameters: – class1: Class1 – class2: Class2

Returns: bool

get_instances_of_class(self, class_name, include_subclasses)

Lines 109-118

Purpose:

Get all instances of a class.

Parameters: – class_name: Class name – include_subclasses: Include subclasses

Returns: <ast.Subscript object at 0x000001A06D60FF90>

---

Class: OntologyReasoningAgent

Lines 121-239

Purpose:

Agent that performs reasoning using ontologies.

Methods:

__init__(self, ontology, model_name)

Lines 124-126

Purpose: Initialize the class instance with required parameters.

Parameters: – ontology: Ontology – model_name: Language model instance

classify_instance(self, instance_description)

Lines 128-151

Purpose:

Classify an instance into ontology classes.

Parameters: – instance_description: Instance description

Returns: str

infer_properties(self, instance_id)

Lines 153-191

Purpose:

Infer properties of an instance based on ontology.

Parameters: – instance_id: Instance id

Returns: <ast.Subscript object at 0x000001A06D825C90>

answer_semantic_query(self, query)

Lines 193-218

Purpose:

Answer semantic queries using ontology.

Parameters: – query: Input query or question

Returns: str

check_consistency(self)

Lines 220-239

Purpose:

Check ontology consistency.

Parameters:

Returns: str

---

Functions

demonstrate_ontology_based_reasoning()

Lines 242-368

Purpose:

Demonstrate ontology-based reasoning pattern.

---

Key Concepts Summary

Pattern Overview

The Ontology Based Reasoning 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

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. 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 .env files 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

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. 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

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 123
• Pattern 125

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 Ontology Based Reasoning 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: 124_ontology_based_reasoning.py
Total Lines: 373
Pattern Category: AI Agent Design Pattern #124

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 123: Knowledge Graph Integration

Description:
    Uses structured knowledge graphs for reasoning with graph traversal,
    pattern matching, and inference capabilities.

Components:
    - Knowledge graph
    - Graph traversal
    - Pattern matching
    - Inference engine

Use Cases:
    - Complex reasoning
    - Knowledge-intensive tasks
    - Relationship discovery

LangChain Implementation:
    Integrates knowledge graphs with LLM reasoning for enhanced understanding.
"""

import os
from typing import List, Dict, Any, Set, Tuple
from dataclasses import dataclass
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()


@dataclass
class Node:
    """Represents a node in the knowledge graph."""
    id: str
    type: str
    properties: Dict[str, Any]


@dataclass
class Edge:
    """Represents an edge in the knowledge graph."""
    source: str
    target: str
    relationship: str
    properties: Dict[str, Any] = None
    
    def __post_init__(self):
        if self.properties is None:
            self.properties = {}


class KnowledgeGraph:
    """Knowledge graph for structured knowledge representation."""
    
    def __init__(self):
        self.nodes: Dict[str, Node] = {}
        self.edges: List[Edge] = []
        
    def add_node(self, node: Node):
        """Add a node to the graph."""
        self.nodes[node.id] = node
        
    def add_edge(self, edge: Edge):
        """Add an edge to the graph."""
        self.edges.append(edge)
        
    def get_neighbors(self, node_id: str, relationship: str = None) -> List[Tuple[Node, str]]:
        """Get neighboring nodes."""
        neighbors = []
        for edge in self.edges:
            if edge.source == node_id:
                if relationship is None or edge.relationship == relationship:
                    if edge.target in self.nodes:
                        neighbors.append((self.nodes[edge.target], edge.relationship))
        return neighbors
    
    def find_path(self, start_id: str, end_id: str, max_depth: int = 5) -> List[str]:
        """Find path between two nodes."""
        visited = set()
        queue = [(start_id, [start_id])]
        
        while queue:
            current, path = queue.pop(0)
            
            if current == end_id:
                return path
            
            if len(path) >= max_depth:
                continue
            
            if current in visited:
                continue
            
            visited.add(current)
            
            for neighbor, _ in self.get_neighbors(current):
                if neighbor.id not in visited:
                    queue.append((neighbor.id, path + [neighbor.id]))
        
        return []
    
    def query_subgraph(self, node_id: str, depth: int = 2) -> Dict[str, Any]:
        """Extract subgraph around a node."""
        visited = set()
        subgraph_nodes = {}
        subgraph_edges = []
        
        def explore(nid, d):
            if d > depth or nid in visited:
                return
            visited.add(nid)
            
            if nid in self.nodes:
                subgraph_nodes[nid] = self.nodes[nid]
                
                for neighbor, relationship in self.get_neighbors(nid):
                    edge = Edge(nid, neighbor.id, relationship)
                    subgraph_edges.append(edge)
                    explore(neighbor.id, d + 1)
        
        explore(node_id, 0)
        
        return {
            "nodes": subgraph_nodes,
            "edges": subgraph_edges
        }


class KnowledgeGraphAgent:
    """Agent that uses knowledge graph for reasoning."""
    
    def __init__(self, kg: KnowledgeGraph, model_name: str = "gpt-4"):
        self.kg = kg
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        
    def answer_question(self, question: str) -> str:
        """Answer question using knowledge graph."""
        # Extract relevant entities from question
        entity_extraction_prompt = ChatPromptTemplate.from_messages([
            ("system", "Extract key entities from this question."),
            ("user", "Question: {question}\n\nEntities (comma-separated):")
        ])
        
        chain = entity_extraction_prompt | self.llm | StrOutputParser()
        entities_str = chain.invoke({"question": question})
        entities = [e.strip() for e in entities_str.split(",")]
        
        # Find relevant nodes
        relevant_nodes = []
        for entity in entities:
            for node_id, node in self.kg.nodes.items():
                if entity.lower() in node.properties.get("name", "").lower():
                    relevant_nodes.append(node)
        
        # Extract relevant subgraph
        kg_context = ""
        for node in relevant_nodes[:3]:  # Limit to top 3
            subgraph = self.kg.query_subgraph(node.id, depth=2)
            
            # Format subgraph info
            node_info = f"\nEntity: {node.properties.get('name', node.id)}\n"
            for edge in subgraph['edges']:
                if edge.source in subgraph['nodes'] and edge.target in subgraph['nodes']:
                    source_name = subgraph['nodes'][edge.source].properties.get('name', edge.source)
                    target_name = subgraph['nodes'][edge.target].properties.get('name', edge.target)
                    node_info += f"  {source_name} --[{edge.relationship}]--> {target_name}\n"
            
            kg_context += node_info
        
        # Answer using knowledge graph context
        qa_prompt = ChatPromptTemplate.from_messages([
            ("system", "Answer the question using the knowledge graph information."),
            ("user", """Knowledge Graph Context:
{kg_context}

Question: {question}

Answer:""")
        ])
        
        chain = qa_prompt | self.llm | StrOutputParser()
        return chain.invoke({
            "kg_context": kg_context or "No relevant information found in knowledge graph",
            "question": question
        })
    
    def infer_relationships(self, entity1: str, entity2: str) -> str:
        """Infer relationships between entities."""
        # Find nodes
        node1 = None
        node2 = None
        
        for node_id, node in self.kg.nodes.items():
            if entity1.lower() in node.properties.get("name", "").lower():
                node1 = node
            if entity2.lower() in node.properties.get("name", "").lower():
                node2 = node
        
        if not node1 or not node2:
            return f"Could not find entities in knowledge graph"
        
        # Find path
        path = self.kg.find_path(node1.id, node2.id)
        
        if not path:
            return f"No direct relationship found between {entity1} and {entity2}"
        
        # Construct relationship chain
        relationships = []
        for i in range(len(path) - 1):
            for edge in self.kg.edges:
                if edge.source == path[i] and edge.target == path[i+1]:
                    source_name = self.kg.nodes[edge.source].properties.get('name', edge.source)
                    target_name = self.kg.nodes[edge.target].properties.get('name', edge.target)
                    relationships.append(f"{source_name} --[{edge.relationship}]--> {target_name}")
        
        # Generate inference
        inference_prompt = ChatPromptTemplate.from_messages([
            ("system", "Explain the relationship between entities based on this knowledge graph path."),
            ("user", """Entity 1: {entity1}
Entity 2: {entity2}

Relationship Chain:
{chain}

Explain the relationship:""")
        ])
        
        chain_str = "\n".join(relationships)
        chain = inference_prompt | self.llm | StrOutputParser()
        
        return chain.invoke({
            "entity1": entity1,
            "entity2": entity2,
            "chain": chain_str
        })
    
    def discover_patterns(self) -> str:
        """Discover patterns in the knowledge graph."""
        # Analyze graph structure
        structure_info = f"Total nodes: {len(self.kg.nodes)}\n"
        structure_info += f"Total edges: {len(self.kg.edges)}\n\n"
        
        # Common relationships
        relationship_counts = {}
        for edge in self.kg.edges:
            relationship_counts[edge.relationship] = relationship_counts.get(edge.relationship, 0) + 1
        
        structure_info += "Relationship types:\n"
        for rel, count in sorted(relationship_counts.items(), key=lambda x: x[1], reverse=True):
            structure_info += f"  {rel}: {count}\n"
        
        pattern_prompt = ChatPromptTemplate.from_messages([
            ("system", "Analyze this knowledge graph structure and identify interesting patterns."),
            ("user", """Graph Structure:
{structure}

Identify:
1. Central entities (hubs)
2. Common patterns
3. Interesting insights
4. Potential gaps""")
        ])
        
        chain = pattern_prompt | self.llm | StrOutputParser()
        return chain.invoke({"structure": structure_info})


def demonstrate_knowledge_graph_integration():
    """Demonstrate knowledge graph integration pattern."""
    print("=== Knowledge Graph Integration Pattern ===\n")
    
    # Create knowledge graph
    print("1. Building Knowledge Graph")
    print("-" * 50)
    
    kg = KnowledgeGraph()
    
    # Add nodes (people, companies, technologies)
    kg.add_node(Node("person1", "Person", {"name": "Alice", "role": "CEO"}))
    kg.add_node(Node("person2", "Person", {"name": "Bob", "role": "CTO"}))
    kg.add_node(Node("person3", "Person", {"name": "Charlie", "role": "Engineer"}))
    
    kg.add_node(Node("company1", "Company", {"name": "TechCorp", "industry": "Software"}))
    kg.add_node(Node("company2", "Company", {"name": "DataInc", "industry": "Analytics"}))
    
    kg.add_node(Node("tech1", "Technology", {"name": "Python", "type": "Language"}))
    kg.add_node(Node("tech2", "Technology", {"name": "Machine Learning", "type": "Field"}))
    kg.add_node(Node("tech3", "Technology", {"name": "TensorFlow", "type": "Framework"}))
    
    # Add edges (relationships)
    kg.add_edge(Edge("person1", "company1", "WORKS_AT"))
    kg.add_edge(Edge("person2", "company1", "WORKS_AT"))
    kg.add_edge(Edge("person3", "company2", "WORKS_AT"))
    
    kg.add_edge(Edge("person1", "person2", "MANAGES"))
    kg.add_edge(Edge("person2", "person3", "MENTORS"))
    
    kg.add_edge(Edge("person2", "tech1", "EXPERT_IN"))
    kg.add_edge(Edge("person2", "tech2", "EXPERT_IN"))
    kg.add_edge(Edge("person3", "tech1", "USES"))
    kg.add_edge(Edge("person3", "tech3", "USES"))
    
    kg.add_edge(Edge("tech3", "tech1", "IMPLEMENTED_IN"))
    kg.add_edge(Edge("tech3", "tech2", "USED_FOR"))
    
    kg.add_edge(Edge("company1", "tech2", "SPECIALIZES_IN"))
    kg.add_edge(Edge("company2", "tech2", "USES"))
    
    print(f"✓ Created knowledge graph with {len(kg.nodes)} nodes and {len(kg.edges)} edges")
    print()
    
    # Create agent
    agent = KnowledgeGraphAgent(kg)
    
    # Question answering
    print("2. Question Answering with Knowledge Graph")
    print("-" * 50)
    
    questions = [
        "Who works at TechCorp?",
        "What technologies does Bob know?",
        "Which companies use Machine Learning?"
    ]
    
    for question in questions:
        print(f"\nQ: {question}")
        answer = agent.answer_question(question)
        print(f"A: {answer}")
    
    # Relationship inference
    print("\n\n3. Inferring Relationships")
    print("-" * 50)
    
    print("\nInferring relationship between Alice and Charlie:")
    inference = agent.infer_relationships("Alice", "Charlie")
    print(inference)
    
    print("\nInferring relationship between Bob and TensorFlow:")
    inference = agent.infer_relationships("Bob", "TensorFlow")
    print(inference)
    
    # Pattern discovery
    print("\n\n4. Discovering Patterns in Knowledge Graph")
    print("-" * 50)
    patterns = agent.discover_patterns()
    print(patterns)
    
    print("\n=== Summary ===")
    print("Knowledge graph integration demonstrated with:")
    print("- Graph construction (nodes and edges)")
    print("- Graph traversal and querying")
    print("- Question answering with graph context")
    print("- Relationship inference")
    print("- Pattern discovery")
    print("- LLM-enhanced graph reasoning")


if __name__ == "__main__":
    demonstrate_knowledge_graph_integration()

    

Agentic Design Pattern: Knowledge Graph Integration

Pattern 123: Knowledge Graph Integration – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Knowledge Graph Integration pattern implementation.

---

Pattern Description

Pattern 123: Knowledge Graph Integration

Description: Uses structured knowledge graphs for reasoning with graph traversal, pattern matching, and inference capabilities.

Components: – Knowledge graph – Graph traversal – Pattern matching – Inference engine

Use Cases: – Complex reasoning – Knowledge-intensive tasks – Relationship discovery

LangChain Implementation: Integrates knowledge graphs with LLM reasoning for enhanced understanding.

---

File Statistics

• Total Lines: 365
• Classes: 4
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 123: Knowledge Graph Integration

Description:
    Uses structured knowledge graphs for reasoning with graph traversal,
    pattern matching, and inference capabilities.

Components:
    - Knowledge graph
    - Graph traversal
    - Pattern matching
    - Inference engine

Use Cases:
    - Complex reasoning
    - Knowledge-intensive tasks
    - Relationship discovery

LangChain Implementation:
    Integrates knowledge graphs with LLM reasoning for enhanced understanding.
"""

Purpose: This module implements the Knowledge Graph Integration pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Set
import typing.Tuple

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: 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.Set: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: 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: Node

Lines 36-40

Purpose:

Represents a node in the knowledge graph.

---

Class: Edge

Lines 44-53

Purpose:

Represents an edge in the knowledge graph.

Methods:

__post_init__(self)

Lines 51-53

Purpose: Post Init operation.

Parameters:

---

Class: KnowledgeGraph

Lines 56-130

Purpose:

Knowledge graph for structured knowledge representation.

Methods:

__init__(self)

Lines 59-61

Purpose: Initialize the class instance with required parameters.

Parameters:

add_node(self, node)

Lines 63-65

Purpose:

Add a node to the graph.

Parameters: – node: Node

add_edge(self, edge)

Lines 67-69

Purpose:

Add an edge to the graph.

Parameters: – edge: Edge

get_neighbors(self, node_id, relationship)

Lines 71-79

Purpose:

Get neighboring nodes.

Parameters: – node_id: Node id – relationship: Relationship

Returns: <ast.Subscript object at 0x000001A06D6EB9D0>

find_path(self, start_id, end_id, max_depth)

Lines 81-104

Purpose:

Find path between two nodes.

Parameters: – start_id: Start id – end_id: End id – max_depth: Max depth

Returns: <ast.Subscript object at 0x000001A06D63BD90>

query_subgraph(self, node_id, depth)

Lines 106-130

Purpose:

Extract subgraph around a node.

Parameters: – node_id: Node id – depth: Depth

Returns: <ast.Subscript object at 0x000001A06D646E50>

---

Class: KnowledgeGraphAgent

Lines 133-270

Purpose:

Agent that uses knowledge graph for reasoning.

Methods:

__init__(self, kg, model_name)

Lines 136-138

Purpose: Initialize the class instance with required parameters.

Parameters: – kg: Kg – model_name: Language model instance

answer_question(self, question)

Lines 140-189

Purpose:

Answer question using knowledge graph.

Parameters: – question: Question to answer

Returns: str

infer_relationships(self, entity1, entity2)

Lines 191-240

Purpose:

Infer relationships between entities.

Parameters: – entity1: Entity1 – entity2: Entity2

Returns: str

discover_patterns(self)

Lines 242-270

Purpose:

Discover patterns in the knowledge graph.

Parameters:

Returns: str

---

Functions

demonstrate_knowledge_graph_integration()

Lines 273-360

Purpose:

Demonstrate knowledge graph integration pattern.

---

Key Concepts Summary

Pattern Overview

The Knowledge Graph Integration 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

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. 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 .env files 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

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. 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

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 122
• Pattern 124

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 Knowledge Graph Integration 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: 123_knowledge_graph_integration.py
Total Lines: 365
Pattern Category: AI Agent Design Pattern #123

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 122: Service Mesh Pattern

Description:
    Infrastructure layer for agent-to-agent communication with service
    discovery, load balancing, and observability.

Components:
    - Service registry
    - Load balancer
    - Circuit breaker
    - Observability layer

Use Cases:
    - Large-scale distributed agents
    - Microservices architecture
    - Reliable communication

LangChain Implementation:
    Implements service mesh concepts for agent communication management.
"""

import os
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
import random
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 ServiceStatus(Enum):
    """Service health status."""
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"


@dataclass
class ServiceInstance:
    """Represents a service instance."""
    id: str
    name: str
    host: str
    port: int
    status: ServiceStatus = ServiceStatus.HEALTHY
    load: int = 0
    max_capacity: int = 100
    response_times: List[float] = None
    
    def __post_init__(self):
        if self.response_times is None:
            self.response_times = []


class ServiceRegistry:
    """Service discovery and registration."""
    
    def __init__(self):
        self.services: Dict[str, List[ServiceInstance]] = {}
        
    def register(self, service_name: str, instance: ServiceInstance):
        """Register a service instance."""
        if service_name not in self.services:
            self.services[service_name] = []
        self.services[service_name].append(instance)
        print(f"✓ Registered {service_name} instance: {instance.id}")
    
    def deregister(self, service_name: str, instance_id: str):
        """Deregister a service instance."""
        if service_name in self.services:
            self.services[service_name] = [
                i for i in self.services[service_name] if i.id != instance_id
            ]
            print(f"✓ Deregistered {service_name} instance: {instance_id}")
    
    def discover(self, service_name: str) -> List[ServiceInstance]:
        """Discover healthy instances of a service."""
        instances = self.services.get(service_name, [])
        return [i for i in instances if i.status == ServiceStatus.HEALTHY]
    
    def get_all_services(self) -> List[str]:
        """Get list of all registered services."""
        return list(self.services.keys())


class LoadBalancer:
    """Load balancing for service requests."""
    
    def __init__(self, strategy: str = "least_loaded"):
        self.strategy = strategy
        
    def select_instance(self, instances: List[ServiceInstance]) -> Optional[ServiceInstance]:
        """Select an instance based on load balancing strategy."""
        if not instances:
            return None
        
        if self.strategy == "round_robin":
            return random.choice(instances)
        
        elif self.strategy == "least_loaded":
            return min(instances, key=lambda i: i.load)
        
        elif self.strategy == "best_performance":
            # Choose instance with best average response time
            instances_with_times = [
                i for i in instances if i.response_times
            ]
            if instances_with_times:
                return min(
                    instances_with_times,
                    key=lambda i: sum(i.response_times) / len(i.response_times)
                )
            return instances[0]
        
        return instances[0]


class CircuitBreaker:
    """Circuit breaker for fault tolerance."""
    
    def __init__(self, failure_threshold: int = 3, timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failures: Dict[str, int] = {}
        self.open_circuits: Dict[str, datetime] = {}
        
    def record_failure(self, service_id: str):
        """Record a service failure."""
        self.failures[service_id] = self.failures.get(service_id, 0) + 1
        
        if self.failures[service_id] >= self.failure_threshold:
            self.open_circuits[service_id] = datetime.now()
            print(f"⚠ Circuit opened for {service_id}")
    
    def record_success(self, service_id: str):
        """Record a successful call."""
        self.failures[service_id] = 0
        if service_id in self.open_circuits:
            del self.open_circuits[service_id]
            print(f"✓ Circuit closed for {service_id}")
    
    def is_open(self, service_id: str) -> bool:
        """Check if circuit is open."""
        if service_id in self.open_circuits:
            # Check if timeout has passed
            elapsed = (datetime.now() - self.open_circuits[service_id]).seconds
            if elapsed > self.timeout:
                del self.open_circuits[service_id]
                return False
            return True
        return False


class ServiceMesh:
    """Service mesh for agent communication."""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.registry = ServiceRegistry()
        self.load_balancer = LoadBalancer(strategy="least_loaded")
        self.circuit_breaker = CircuitBreaker()
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.request_log: List[Dict[str, Any]] = []
        
    def call_service(self, service_name: str, request_data: Dict[str, Any]) -> Dict[str, Any]:
        """Make a service call through the mesh."""
        print(f"\n🔄 Calling service: {service_name}")
        
        # Discover instances
        instances = self.registry.discover(service_name)
        
        if not instances:
            print(f"  ✗ No healthy instances found")
            return {"error": "ServiceUnavailable"}
        
        # Select instance via load balancer
        instance = self.load_balancer.select_instance(instances)
        
        if not instance:
            return {"error": "NoInstanceAvailable"}
        
        # Check circuit breaker
        if self.circuit_breaker.is_open(instance.id):
            print(f"  ⚠ Circuit open for {instance.id}, trying another instance")
            # Try another instance
            instances.remove(instance)
            if instances:
                instance = self.load_balancer.select_instance(instances)
            else:
                return {"error": "AllCircuitsOpen"}
        
        print(f"  → Routing to instance: {instance.id} (load: {instance.load})")
        
        # Simulate service call
        try:
            start_time = datetime.now()
            
            # Simulate processing
            instance.load += 10
            response_time = random.uniform(0.1, 0.5)
            instance.response_times.append(response_time)
            
            # Simulate occasional failures
            if random.random()  str:
        """Generate observability report."""
        report_prompt = ChatPromptTemplate.from_messages([
            ("system", "Analyze service mesh metrics and provide insights."),
            ("user", """Request Log:
{log}

Service Registry:
{registry}

Provide:
1. Service health summary
2. Performance metrics
3. Bottlenecks identified
4. Recommendations""")
        ])
        
        log_summary = "\n".join([
            f"{r['service']}: {r['status']} ({r.get('response_time', 'N/A')}s)"
            for r in self.request_log[-20:]
        ])
        
        registry_summary = "\n".join([
            f"{name}: {len(instances)} instances"
            for name, instances in self.registry.services.items()
        ])
        
        chain = report_prompt | self.llm | StrOutputParser()
        return chain.invoke({
            "log": log_summary,
            "registry": registry_summary
        })


def demonstrate_service_mesh():
    """Demonstrate service mesh pattern."""
    print("=== Service Mesh Pattern ===\n")
    
    mesh = ServiceMesh()
    
    # Register services
    print("1. Registering Service Instances")
    print("-" * 50)
    
    # Data processing service
    mesh.registry.register("data-processor", ServiceInstance(
        id="dp-1", name="data-processor", host="10.0.1.1", port=8001
    ))
    mesh.registry.register("data-processor", ServiceInstance(
        id="dp-2", name="data-processor", host="10.0.1.2", port=8001
    ))
    mesh.registry.register("data-processor", ServiceInstance(
        id="dp-3", name="data-processor", host="10.0.1.3", port=8001
    ))
    
    # Analytics service
    mesh.registry.register("analytics", ServiceInstance(
        id="an-1", name="analytics", host="10.0.2.1", port=8002
    ))
    mesh.registry.register("analytics", ServiceInstance(
        id="an-2", name="analytics", host="10.0.2.2", port=8002
    ))
    
    print()
    
    # Make service calls
    print("2. Making Service Calls Through Mesh")
    print("-" * 50)
    
    for i in range(10):
        result = mesh.call_service("data-processor", {"data": f"request_{i}"})
    
    for i in range(5):
        result = mesh.call_service("analytics", {"query": f"query_{i}"})
    
    # Observability
    print("\n3. Service Mesh Observability")
    print("-" * 50)
    
    print(f"Total requests: {len(mesh.request_log)}")
    successful = sum(1 for r in mesh.request_log if r['status'] == 'success')
    print(f"Successful: {successful}")
    print(f"Failed: {len(mesh.request_log) - successful}")
    print(f"Success rate: {successful/len(mesh.request_log)*100:.1f}%")
    
    # Generate report
    print("\n4. Detailed Observability Report")
    print("-" * 50)
    report = mesh.get_observability_report()
    print(report)
    
    print("\n=== Summary ===")
    print("Service mesh demonstrated with:")
    print("- Service discovery and registration")
    print("- Load balancing (least loaded strategy)")
    print("- Circuit breaker pattern")
    print("- Request routing")
    print("- Health monitoring")
    print("- Observability and metrics")


if __name__ == "__main__":
    demonstrate_service_mesh()

    

Agentic Design Pattern: Service Mesh Pattern

Pattern 122: Service Mesh Pattern – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Service Mesh Pattern pattern implementation.

---

Pattern Description

Pattern 122: Service Mesh Pattern

Description: Infrastructure layer for agent-to-agent communication with service discovery, load balancing, and observability.

Components: – Service registry – Load balancer – Circuit breaker – Observability layer

Use Cases: – Large-scale distributed agents – Microservices architecture – Reliable communication

LangChain Implementation: Implements service mesh concepts for agent communication management.

---

File Statistics

• Total Lines: 355
• Classes: 6
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 122: Service Mesh Pattern

Description:
    Infrastructure layer for agent-to-agent communication with service
    discovery, load balancing, and observability.

Components:
    - Service registry
    - Load balancer
    - Circuit breaker
    - Observability layer

Use Cases:
    - Large-scale distributed agents
    - Microservices architecture
    - Reliable communication

LangChain Implementation:
    Implements service mesh concepts for agent communication management.
"""

Purpose: This module implements the Service Mesh Pattern pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import datetime.datetime
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import random
import typing.Any
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – random: Random number generation – typing.Any: 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)

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: ServiceStatus

Lines 38-42

Inherits from: Enum

Purpose:

Service health status.

---

Class: ServiceInstance

Lines 46-59

Purpose:

Represents a service instance.

Methods:

__post_init__(self)

Lines 57-59

Purpose: Post Init operation.

Parameters:

---

Class: ServiceRegistry

Lines 62-90

Purpose:

Service discovery and registration.

Methods:

__init__(self)

Lines 65-66

Purpose: Initialize the class instance with required parameters.

Parameters:

register(self, service_name, instance)

Lines 68-73

Purpose:

Register a service instance.

Parameters: – service_name: Service name – instance: Instance

deregister(self, service_name, instance_id)

Lines 75-81

Purpose:

Deregister a service instance.

Parameters: – service_name: Service name – instance_id: Instance id

discover(self, service_name)

Lines 83-86

Purpose:

Discover healthy instances of a service.

Parameters: – service_name: Service name

Returns: <ast.Subscript object at 0x000001A06D718710>

get_all_services(self)

Lines 88-90

Purpose:

Get list of all registered services.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D71B2D0>

---

Class: LoadBalancer

Lines 93-122

Purpose:

Load balancing for service requests.

Methods:

__init__(self, strategy)

Lines 96-97

Purpose: Initialize the class instance with required parameters.

Parameters: – strategy: Strategy

select_instance(self, instances)

Lines 99-122

Purpose:

Select an instance based on load balancing strategy.

Parameters: – instances: Instances

Returns: <ast.Subscript object at 0x000001A06D8271D0>

---

Class: CircuitBreaker

Lines 125-158

Purpose:

Circuit breaker for fault tolerance.

Methods:

__init__(self, failure_threshold, timeout)

Lines 128-132

Purpose: Initialize the class instance with required parameters.

Parameters: – failure_threshold: Failure threshold – timeout: Timeout

record_failure(self, service_id)

Lines 134-140

Purpose:

Record a service failure.

Parameters: – service_id: Service id

record_success(self, service_id)

Lines 142-147

Purpose:

Record a successful call.

Parameters: – service_id: Service id

is_open(self, service_id)

Lines 149-158

Purpose:

Check if circuit is open.

Parameters: – service_id: Service id

Returns: bool

---

Class: ServiceMesh

Lines 161-283

Purpose:

Service mesh for agent communication.

Methods:

__init__(self, model_name)

Lines 164-169

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

call_service(self, service_name, request_data)

Lines 171-250

Purpose:

Make a service call through the mesh.

Parameters: – service_name: Service name – request_data: Request data

Returns: <ast.Subscript object at 0x000001A06D64B650>

get_observability_report(self)

Lines 252-283

Purpose:

Generate observability report.

Parameters:

Returns: str

---

Functions

demonstrate_service_mesh()

Lines 286-350

Purpose:

Demonstrate service mesh pattern.

---

Key Concepts Summary

Pattern Overview

The Service Mesh Pattern 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

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. 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 .env files 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

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. 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

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 121
• Pattern 123

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 Service Mesh Pattern 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: 122_service_mesh_pattern.py
Total Lines: 355
Pattern Category: AI Agent Design Pattern #122

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 121: Event-Driven Architecture

Description:
    Agents react to events in the system using pub-sub patterns and
    event buses for loose coupling and scalability.

Components:
    - Event bus
    - Publishers
    - Subscribers
    - Event handlers

Use Cases:
    - Real-time systems
    - Microservices
    - Reactive applications

LangChain Implementation:
    Implements event-driven pattern with event handlers and async processing.
"""

import os
from typing import List, Dict, Any, Callable
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
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 EventType(Enum):
    """Types of events in the system."""
    USER_ACTION = "user_action"
    SYSTEM_STATE = "system_state"
    DATA_UPDATE = "data_update"
    ERROR = "error"
    NOTIFICATION = "notification"


@dataclass
class Event:
    """Represents an event in the system."""
    id: str
    type: EventType
    source: str
    data: Dict[str, Any]
    timestamp: str
    priority: int = 5


class EventBus:
    """Central event bus for publish-subscribe pattern."""
    
    def __init__(self):
        self.subscribers: Dict[EventType, List[Callable]] = {}
        self.event_history: List[Event] = []
        
    def subscribe(self, event_type: EventType, handler: Callable):
        """Subscribe a handler to an event type."""
        if event_type not in self.subscribers:
            self.subscribers[event_type] = []
        self.subscribers[event_type].append(handler)
        print(f"✓ Subscribed handler to {event_type.value} events")
    
    def publish(self, event: Event):
        """Publish an event to all subscribers."""
        print(f"\n📤 Publishing event: {event.type.value} from {event.source}")
        print(f"   Data: {event.data}")
        
        self.event_history.append(event)
        
        # Notify subscribers
        if event.type in self.subscribers:
            for handler in self.subscribers[event.type]:
                try:
                    handler(event)
                except Exception as e:
                    print(f"   ⚠ Handler error: {e}")
    
    def get_event_count(self, event_type: EventType = None) -> int:
        """Get count of events."""
        if event_type:
            return sum(1 for e in self.event_history if e.type == event_type)
        return len(self.event_history)


class EventDrivenAgent:
    """Agent that reacts to events."""
    
    def __init__(self, name: str, event_bus: EventBus, model_name: str = "gpt-4"):
        self.name = name
        self.event_bus = event_bus
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.handled_events = []
        
    def handle_event(self, event: Event):
        """Handle an incoming event."""
        print(f"   📥 {self.name} received event")
        
        # Process event with LLM
        response_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an event-driven agent named {agent_name}.
React appropriately to the event based on its type and data."""),
            ("user", """Event Type: {event_type}
Source: {source}
Data: {data}

Provide your response action:""")
        ])
        
        chain = response_prompt | self.llm | StrOutputParser()
        response = chain.invoke({
            "agent_name": self.name,
            "event_type": event.type.value,
            "source": event.source,
            "data": str(event.data)
        })
        
        print(f"   🔧 {self.name} action: {response[:100]}...")
        
        self.handled_events.append({
            "event_id": event.id,
            "response": response
        })
        
        # Potentially publish new events
        if "notify" in response.lower():
            self.publish_notification(event)
    
    def publish_notification(self, original_event: Event):
        """Publish a notification event."""
        notification_event = Event(
            id=f"notif_{len(self.event_bus.event_history)}",
            type=EventType.NOTIFICATION,
            source=self.name,
            data={"message": f"Processed event {original_event.id}"},
            timestamp=datetime.now().isoformat()
        )
        self.event_bus.publish(notification_event)


class EventAnalyzer:
    """Analyzes event patterns."""
    
    def __init__(self, event_bus: EventBus, model_name: str = "gpt-4"):
        self.event_bus = event_bus
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
    
    def analyze_patterns(self) -> str:
        """Analyze event patterns using LLM."""
        events_summary = "\n".join([
            f"{e.type.value}: {e.source} -> {e.data}"
            for e in self.event_bus.event_history[-20:]
        ])
        
        analysis_prompt = ChatPromptTemplate.from_messages([
            ("system", "Analyze these event patterns and identify trends."),
            ("user", """Recent Events:
{events}

Provide:
1. Event frequency patterns
2. Common event chains
3. Potential issues
4. Optimization suggestions""")
        ])
        
        chain = analysis_prompt | self.llm | StrOutputParser()
        return chain.invoke({"events": events_summary})


def demonstrate_event_driven_architecture():
    """Demonstrate event-driven architecture pattern."""
    print("=== Event-Driven Architecture Pattern ===\n")
    
    # Create event bus
    event_bus = EventBus()
    
    # Create agents
    print("1. Creating Event-Driven Agents")
    print("-" * 50)
    
    monitoring_agent = EventDrivenAgent("MonitoringAgent", event_bus)
    analytics_agent = EventDrivenAgent("AnalyticsAgent", event_bus)
    notification_agent = EventDrivenAgent("NotificationAgent", event_bus)
    
    # Subscribe agents to events
    print("\n2. Subscribing to Events")
    print("-" * 50)
    
    event_bus.subscribe(EventType.USER_ACTION, monitoring_agent.handle_event)
    event_bus.subscribe(EventType.USER_ACTION, analytics_agent.handle_event)
    
    event_bus.subscribe(EventType.SYSTEM_STATE, monitoring_agent.handle_event)
    
    event_bus.subscribe(EventType.ERROR, monitoring_agent.handle_event)
    event_bus.subscribe(EventType.ERROR, notification_agent.handle_event)
    
    event_bus.subscribe(EventType.NOTIFICATION, notification_agent.handle_event)
    
    print()
    
    # Publish events
    print("3. Publishing Events")
    print("-" * 50)
    
    # User action event
    event_bus.publish(Event(
        id="evt_001",
        type=EventType.USER_ACTION,
        source="WebApp",
        data={"action": "login", "user_id": "user123"},
        timestamp=datetime.now().isoformat(),
        priority=5
    ))
    
    # System state event
    event_bus.publish(Event(
        id="evt_002",
        type=EventType.SYSTEM_STATE,
        source="SystemMonitor",
        data={"cpu": 75, "memory": 60, "status": "healthy"},
        timestamp=datetime.now().isoformat(),
        priority=3
    ))
    
    # Data update event
    event_bus.publish(Event(
        id="evt_003",
        type=EventType.DATA_UPDATE,
        source="Database",
        data={"table": "users", "operation": "insert", "count": 5},
        timestamp=datetime.now().isoformat(),
        priority=4
    ))
    
    # Error event
    event_bus.publish(Event(
        id="evt_004",
        type=EventType.ERROR,
        source="APIGateway",
        data={"error": "RateLimitExceeded", "endpoint": "/api/data"},
        timestamp=datetime.now().isoformat(),
        priority=9
    ))
    
    # Another user action
    event_bus.publish(Event(
        id="evt_005",
        type=EventType.USER_ACTION,
        source="MobileApp",
        data={"action": "purchase", "amount": 99.99},
        timestamp=datetime.now().isoformat(),
        priority=7
    ))
    
    # Analyze event patterns
    print("\n4. Analyzing Event Patterns")
    print("-" * 50)
    
    analyzer = EventAnalyzer(event_bus)
    analysis = analyzer.analyze_patterns()
    print(analysis)
    
    # Statistics
    print("\n5. Event Statistics")
    print("-" * 50)
    print(f"Total events: {event_bus.get_event_count()}")
    print(f"User actions: {event_bus.get_event_count(EventType.USER_ACTION)}")
    print(f"System states: {event_bus.get_event_count(EventType.SYSTEM_STATE)}")
    print(f"Errors: {event_bus.get_event_count(EventType.ERROR)}")
    print(f"Notifications: {event_bus.get_event_count(EventType.NOTIFICATION)}")
    
    print(f"\nMonitoring agent handled: {len(monitoring_agent.handled_events)} events")
    print(f"Analytics agent handled: {len(analytics_agent.handled_events)} events")
    print(f"Notification agent handled: {len(notification_agent.handled_events)} events")
    
    print("\n=== Summary ===")
    print("Event-driven architecture demonstrated with:")
    print("- Event bus (pub-sub pattern)")
    print("- Multiple event types")
    print("- Event subscribers and handlers")
    print("- Event-driven agents")
    print("- Event chain reactions")
    print("- Pattern analysis")
    print("- Loose coupling between components")


if __name__ == "__main__":
    demonstrate_event_driven_architecture()

    

Agentic Design Pattern: Event Driven Architecture

Pattern 121: Event Driven Architecture – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Event Driven Architecture pattern implementation.

---

Pattern Description

Pattern 121: Event-Driven Architecture

Description: Agents react to events in the system using pub-sub patterns and event buses for loose coupling and scalability.

Components: – Event bus – Publishers – Subscribers – Event handlers

Use Cases: – Real-time systems – Microservices – Reactive applications

LangChain Implementation: Implements event-driven pattern with event handlers and async processing.

---

File Statistics

• Total Lines: 297
• Classes: 5
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 121: Event-Driven Architecture

Description:
    Agents react to events in the system using pub-sub patterns and
    event buses for loose coupling and scalability.

Components:
    - Event bus
    - Publishers
    - Subscribers
    - Event handlers

Use Cases:
    - Real-time systems
    - Microservices
    - Reactive applications

LangChain Implementation:
    Implements event-driven pattern with event handlers and async processing.
"""

Purpose: This module implements the Event Driven Architecture pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import datetime.datetime
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Callable
import typing.Dict
import typing.List

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – 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)

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: EventType

Lines 37-43

Inherits from: Enum

Purpose:

Types of events in the system.

---

Class: Event

Lines 47-54

Purpose:

Represents an event in the system.

---

Class: EventBus

Lines 57-90

Purpose:

Central event bus for publish-subscribe pattern.

Methods:

__init__(self)

Lines 60-62

Purpose: Initialize the class instance with required parameters.

Parameters:

subscribe(self, event_type, handler)

Lines 64-69

Purpose:

Subscribe a handler to an event type.

Parameters: – event_type: Event type – handler: Handler

publish(self, event)

Lines 71-84

Purpose:

Publish an event to all subscribers.

Parameters: – event: Event

get_event_count(self, event_type)

Lines 86-90

Purpose:

Get count of events.

Parameters: – event_type: Event type

Returns: int

---

Class: EventDrivenAgent

Lines 93-145

Purpose:

Agent that reacts to events.

Methods:

__init__(self, name, event_bus, model_name)

Lines 96-100

Purpose: Initialize the class instance with required parameters.

Parameters: – name: Name – event_bus: Event bus – model_name: Language model instance

handle_event(self, event)

Lines 102-134

Purpose:

Handle an incoming event.

Parameters: – event: Event

publish_notification(self, original_event)

Lines 136-145

Purpose:

Publish a notification event.

Parameters: – original_event: Original event

---

Class: EventAnalyzer

Lines 148-175

Purpose:

Analyzes event patterns.

Methods:

__init__(self, event_bus, model_name)

Lines 151-153

Purpose: Initialize the class instance with required parameters.

Parameters: – event_bus: Event bus – model_name: Language model instance

analyze_patterns(self)

Lines 155-175

Purpose:

Analyze event patterns using LLM.

Parameters:

Returns: str

---

Functions

demonstrate_event_driven_architecture()

Lines 178-292

Purpose:

Demonstrate event-driven architecture pattern.

---

Key Concepts Summary

Pattern Overview

The Event Driven Architecture 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

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. 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 .env files 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

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. 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

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 120
• Pattern 122

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 Event Driven Architecture 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: 121_event_driven_architecture.py
Total Lines: 297
Pattern Category: AI Agent Design Pattern #121

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 120: Workflow Orchestration

Description:
    Manages complex multi-step workflows with dependencies, branching,
    error handling, and monitoring.

Components:
    - Workflow engine
    - Step dependencies
    - Error handling
    - Progress monitoring

Use Cases:
    - Business processes
    - Data pipelines
    - Complex automation

LangChain Implementation:
    Uses LangGraph for state machine-based workflow orchestration.
"""

import os
from typing import List, Dict, Any, Callable, Optional
from enum import Enum
from dataclasses import dataclass
from datetime import datetime
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 StepStatus(Enum):
    """Status of workflow step."""
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    SKIPPED = "skipped"


@dataclass
class WorkflowStep:
    """Represents a step in the workflow."""
    name: str
    description: str
    dependencies: List[str]
    action: Callable
    status: StepStatus = StepStatus.PENDING
    result: Any = None
    error: str = None
    retry_count: int = 0
    max_retries: int = 3


class WorkflowOrchestrator:
    """Orchestrates complex multi-step workflows."""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.steps: Dict[str, WorkflowStep] = {}
        self.execution_log: List[Dict[str, Any]] = []
        self.context: Dict[str, Any] = {}
        
    def add_step(self, step: WorkflowStep):
        """Add a step to the workflow."""
        self.steps[step.name] = step
        print(f"✓ Added step: {step.name}")
    
    def validate_workflow(self) -> bool:
        """Validate workflow structure and dependencies."""
        print("\n=== Validating Workflow ===")
        
        # Check for circular dependencies
        for step_name, step in self.steps.items():
            if self._has_circular_dependency(step_name, set()):
                print(f"✗ Circular dependency detected for step: {step_name}")
                return False
        
        # Check all dependencies exist
        for step_name, step in self.steps.items():
            for dep in step.dependencies:
                if dep not in self.steps:
                    print(f"✗ Missing dependency: {dep} for step {step_name}")
                    return False
        
        print("✓ Workflow validation passed")
        return True
    
    def _has_circular_dependency(self, step_name: str, visited: set) -> bool:
        """Check for circular dependencies."""
        if step_name in visited:
            return True
        
        visited.add(step_name)
        step = self.steps.get(step_name)
        
        if step:
            for dep in step.dependencies:
                if self._has_circular_dependency(dep, visited.copy()):
                    return True
        
        return False
    
    def get_ready_steps(self) -> List[WorkflowStep]:
        """Get steps that are ready to execute."""
        ready = []
        
        for step in self.steps.values():
            if step.status != StepStatus.PENDING:
                continue
            
            # Check if all dependencies are completed
            deps_completed = all(
                self.steps[dep].status == StepStatus.COMPLETED
                for dep in step.dependencies
            )
            
            if deps_completed:
                ready.append(step)
        
        return ready
    
    def execute_step(self, step: WorkflowStep) -> bool:
        """Execute a single workflow step."""
        print(f"\n▶ Executing: {step.name}")
        print(f"  Description: {step.description}")
        
        step.status = StepStatus.RUNNING
        start_time = datetime.now()
        
        try:
            # Execute the step's action
            step.result = step.action(self.context)
            step.status = StepStatus.COMPLETED
            
            # Log execution
            log_entry = {
                "step": step.name,
                "status": "completed",
                "duration": (datetime.now() - start_time).total_seconds(),
                "timestamp": datetime.now().isoformat()
            }
            self.execution_log.append(log_entry)
            
            print(f"  ✓ Completed")
            return True
            
        except Exception as e:
            step.error = str(e)
            step.retry_count += 1
            
            if step.retry_count < step.max_retries:
                print(f"  ⚠ Failed (attempt {step.retry_count}/{step.max_retries}): {e}")
                step.status = StepStatus.PENDING  # Retry
            else:
                step.status = StepStatus.FAILED
                print(f"  ✗ Failed after {step.retry_count} attempts: {e}")
            
            log_entry = {
                "step": step.name,
                "status": "failed",
                "error": str(e),
                "retry_count": step.retry_count,
                "timestamp": datetime.now().isoformat()
            }
            self.execution_log.append(log_entry)
            
            return False
    
    def execute_workflow(self):
        """Execute the entire workflow."""
        print("\n=== Executing Workflow ===")
        
        if not self.validate_workflow():
            print("✗ Workflow validation failed")
            return False
        
        total_steps = len(self.steps)
        completed = 0
        
        while completed  0:
                    break
                else:
                    continue
            
            # Execute ready steps (can be parallelized)
            for step in ready_steps:
                self.execute_step(step)
            
            completed = sum(1 for s in self.steps.values() if s.status == StepStatus.COMPLETED)
        
        # Generate execution summary
        self.generate_summary()
        
        return all(s.status == StepStatus.COMPLETED for s in self.steps.values())
    
    def generate_summary(self):
        """Generate workflow execution summary."""
        print("\n=== Workflow Summary ===")
        
        completed = sum(1 for s in self.steps.values() if s.status == StepStatus.COMPLETED)
        failed = sum(1 for s in self.steps.values() if s.status == StepStatus.FAILED)
        total = len(self.steps)
        
        print(f"Total Steps: {total}")
        print(f"Completed: {completed}")
        print(f"Failed: {failed}")
        print(f"Success Rate: {completed/total*100:.1f}%")
        
        if failed > 0:
            print("\nFailed Steps:")
            for step in self.steps.values():
                if step.status == StepStatus.FAILED:
                    print(f"  - {step.name}: {step.error}")
        
        # LLM-based analysis
        if self.execution_log:
            analysis_prompt = ChatPromptTemplate.from_messages([
                ("system", "Analyze this workflow execution and provide insights."),
                ("user", """Execution Log:
{log}

Provide:
1. Performance insights
2. Bottlenecks identified
3. Optimization suggestions""")
            ])
            
            log_str = "\n".join([
                f"{entry['step']}: {entry['status']} ({entry.get('duration', 'N/A')}s)"
                for entry in self.execution_log
            ])
            
            chain = analysis_prompt | self.llm | StrOutputParser()
            analysis = chain.invoke({"log": log_str})
            
            print("\nWorkflow Analysis:")
            print(analysis)


def demonstrate_workflow_orchestration():
    """Demonstrate workflow orchestration pattern."""
    print("=== Workflow Orchestration Pattern ===\n")
    
    orchestrator = WorkflowOrchestrator()
    
    # Define workflow steps for data processing pipeline
    print("1. Defining Workflow Steps")
    print("-" * 50)
    
    # Step 1: Data Extraction
    def extract_data(context):
        print("    Extracting data from source...")
        context['raw_data'] = [1, 2, 3, 4, 5]
        return "Data extracted"
    
    orchestrator.add_step(WorkflowStep(
        name="extract_data",
        description="Extract data from source system",
        dependencies=[],
        action=extract_data
    ))
    
    # Step 2: Data Validation
    def validate_data(context):
        print("    Validating data...")
        data = context.get('raw_data', [])
        if not data:
            raise ValueError("No data to validate")
        context['validated_data'] = data
        return "Data validated"
    
    orchestrator.add_step(WorkflowStep(
        name="validate_data",
        description="Validate extracted data",
        dependencies=["extract_data"],
        action=validate_data
    ))
    
    # Step 3: Data Transformation
    def transform_data(context):
        print("    Transforming data...")
        data = context.get('validated_data', [])
        context['transformed_data'] = [x * 2 for x in data]
        return "Data transformed"
    
    orchestrator.add_step(WorkflowStep(
        name="transform_data",
        description="Transform data format",
        dependencies=["validate_data"],
        action=transform_data
    ))
    
    # Step 4: Data Enrichment
    def enrich_data(context):
        print("    Enriching data...")
        data = context.get('transformed_data', [])
        context['enriched_data'] = [{"value": x, "category": "processed"} for x in data]
        return "Data enriched"
    
    orchestrator.add_step(WorkflowStep(
        name="enrich_data",
        description="Enrich data with additional information",
        dependencies=["transform_data"],
        action=enrich_data
    ))
    
    # Step 5: Load to Database
    def load_to_db(context):
        print("    Loading to database...")
        data = context.get('enriched_data', [])
        context['db_records'] = len(data)
        return f"Loaded {len(data)} records"
    
    orchestrator.add_step(WorkflowStep(
        name="load_to_db",
        description="Load data to database",
        dependencies=["enrich_data"],
        action=load_to_db
    ))
    
    # Step 6: Generate Report
    def generate_report(context):
        print("    Generating report...")
        records = context.get('db_records', 0)
        context['report'] = f"Successfully processed {records} records"
        return "Report generated"
    
    orchestrator.add_step(WorkflowStep(
        name="generate_report",
        description="Generate processing report",
        dependencies=["load_to_db"],
        action=generate_report
    ))
    
    print()
    
    # Execute workflow
    print("2. Executing Workflow")
    print("-" * 50)
    success = orchestrator.execute_workflow()
    
    print("\n=== Results ===")
    print(f"Workflow {'succeeded' if success else 'failed'}")
    print(f"Final context: {orchestrator.context}")
    
    print("\n=== Summary ===")
    print("Workflow orchestration demonstrated with:")
    print("- Multi-step pipeline")
    print("- Dependency management")
    print("- Error handling and retries")
    print("- Progress tracking")
    print("- Execution logging")
    print("- Performance analysis")


if __name__ == "__main__":
    demonstrate_workflow_orchestration()

    

Agentic Design Pattern: Workflow Orchestration

Pattern 120: Workflow Orchestration – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Workflow Orchestration pattern implementation.

---

Pattern Description

Pattern 120: Workflow Orchestration

Description: Manages complex multi-step workflows with dependencies, branching, error handling, and monitoring.

Components: – Workflow engine – Step dependencies – Error handling – Progress monitoring

Use Cases: – Business processes – Data pipelines – Complex automation

LangChain Implementation: Uses LangGraph for state machine-based workflow orchestration.

---

File Statistics

• Total Lines: 372
• Classes: 3
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 120: Workflow Orchestration

Description:
    Manages complex multi-step workflows with dependencies, branching,
    error handling, and monitoring.

Components:
    - Workflow engine
    - Step dependencies
    - Error handling
    - Progress monitoring

Use Cases:
    - Business processes
    - Data pipelines
    - Complex automation

LangChain Implementation:
    Uses LangGraph for state machine-based workflow orchestration.
"""

Purpose: This module implements the Workflow Orchestration pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import datetime.datetime
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Callable
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – 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)

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: StepStatus

Lines 37-43

Inherits from: Enum

Purpose:

Status of workflow step.

---

Class: WorkflowStep

Lines 47-57

Purpose:

Represents a step in the workflow.

---

Class: WorkflowOrchestrator

Lines 60-251

Purpose:

Orchestrates complex multi-step workflows.

Methods:

__init__(self, model_name)

Lines 63-67

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

add_step(self, step)

Lines 69-72

Purpose:

Add a step to the workflow.

Parameters: – step: Step

validate_workflow(self)

Lines 74-92

Purpose:

Validate workflow structure and dependencies.

Parameters:

Returns: bool

_has_circular_dependency(self, step_name, visited)

Lines 94-107

Purpose:

Check for circular dependencies.

Parameters: – step_name: Step name – visited: Visited

Returns: bool

get_ready_steps(self)

Lines 109-126

Purpose:

Get steps that are ready to execute.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D889CD0>

execute_step(self, step)

Lines 128-173

Purpose:

Execute a single workflow step.

Parameters: – step: Step

Returns: bool

execute_workflow(self)

Lines 175-208

Purpose:

Execute the entire workflow.

Parameters:

generate_summary(self)

Lines 210-251

Purpose:

Generate workflow execution summary.

Parameters:

---

Functions

demonstrate_workflow_orchestration()

Lines 254-367

Purpose:

Demonstrate workflow orchestration pattern.

---

Key Concepts Summary

Pattern Overview

The Workflow Orchestration 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

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. 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 .env files 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

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. 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

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 119
• Pattern 121

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 Workflow Orchestration 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: 120_workflow_orchestration.py
Total Lines: 372
Pattern Category: AI Agent Design Pattern #120

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 119: Task Allocation & Scheduling

Description:
    Assigns tasks to agents based on capabilities, load, and constraints
    using auction-based or optimal assignment algorithms.

Components:
    - Task queue
    - Agent capabilities
    - Allocation algorithm
    - Load balancing

Use Cases:
    - Multi-agent systems
    - Distributed computing
    - Resource optimization

LangChain Implementation:
    Uses LLM-based task analysis and agent matching with LangGraph for orchestration.
"""

import os
from typing import List, Dict, Any
from dataclasses import dataclass
from datetime import datetime
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()


@dataclass
class Task:
    """Represents a task to be allocated."""
    id: str
    description: str
    required_skills: List[str]
    priority: int
    estimated_duration: int
    deadline: str = None


@dataclass
class Agent:
    """Represents an agent with capabilities."""
    id: str
    name: str
    skills: List[str]
    current_load: int = 0
    max_capacity: int = 100
    performance_history: List[float] = None
    
    def __post_init__(self):
        if self.performance_history is None:
            self.performance_history = []


class TaskAllocationScheduler:
    """Schedules and allocates tasks to agents optimally."""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.agents: List[Agent] = []
        self.task_queue: List[Task] = []
        self.assignments: Dict[str, str] = {}  # task_id -> agent_id
        
    def register_agent(self, agent: Agent):
        """Register an agent in the system."""
        self.agents.append(agent)
        print(f"✓ Registered agent: {agent.name} with skills {agent.skills}")
    
    def add_task(self, task: Task):
        """Add task to queue."""
        self.task_queue.append(task)
        print(f"✓ Added task: {task.description} (Priority: {task.priority})")
    
    def analyze_task_requirements(self, task: Task) -> Dict[str, Any]:
        """Analyze task requirements using LLM."""
        analysis_prompt = ChatPromptTemplate.from_messages([
            ("system", "Analyze this task and provide detailed requirements."),
            ("user", """Task: {description}
Required Skills: {skills}
Priority: {priority}

Provide:
1. Complexity level (1-10)
2. Required expertise level
3. Any special considerations""")
        ])
        
        chain = analysis_prompt | self.llm | StrOutputParser()
        analysis = chain.invoke({
            "description": task.description,
            "skills": ", ".join(task.required_skills),
            "priority": task.priority
        })
        
        return {"analysis": analysis}
    
    def calculate_agent_score(self, agent: Agent, task: Task) -> float:
        """Calculate how well an agent matches a task."""
        scoring_prompt = ChatPromptTemplate.from_messages([
            ("system", "Score how well this agent matches the task requirements (0-100)."),
            ("user", """Agent: {agent_name}
Agent Skills: {agent_skills}
Current Load: {current_load}/{max_capacity}
Performance History: {performance}

Task: {task_description}
Required Skills: {required_skills}
Priority: {priority}

Return only a numerical score (0-100):""")
        ])
        
        chain = scoring_prompt | self.llm | StrOutputParser()
        response = chain.invoke({
            "agent_name": agent.name,
            "agent_skills": ", ".join(agent.skills),
            "current_load": agent.current_load,
            "max_capacity": agent.max_capacity,
            "performance": sum(agent.performance_history) / len(agent.performance_history) if agent.performance_history else "No history",
            "task_description": task.description,
            "required_skills": ", ".join(task.required_skills),
            "priority": task.priority
        })
        
        try:
            score = float(response.strip().split()[0])
            
            # Adjust for current load
            load_factor = (agent.max_capacity - agent.current_load) / agent.max_capacity
            adjusted_score = score * load_factor
            
            return max(0.0, min(100.0, adjusted_score))
        except:
            return 0.0
    
    def allocate_tasks(self):
        """Allocate tasks to agents using optimal matching."""
        print("\n=== Task Allocation Process ===\n")
        
        # Sort tasks by priority
        sorted_tasks = sorted(self.task_queue, key=lambda t: t.priority, reverse=True)
        
        for task in sorted_tasks:
            print(f"Allocating: {task.description}")
            
            # Analyze task
            analysis = self.analyze_task_requirements(task)
            print(f"  Task Analysis: {analysis['analysis'][:100]}...")
            
            # Score all agents
            agent_scores = []
            for agent in self.agents:
                score = self.calculate_agent_score(agent, task)
                agent_scores.append((agent, score))
                print(f"  {agent.name}: {score:.1f}")
            
            # Select best agent
            if agent_scores:
                best_agent, best_score = max(agent_scores, key=lambda x: x[1])
                
                if best_score > 0:
                    self.assignments[task.id] = best_agent.id
                    best_agent.current_load += task.estimated_duration
                    print(f"  ✓ Assigned to: {best_agent.name} (Score: {best_score:.1f})")
                else:
                    print(f"  ✗ No suitable agent found")
            print()
        
        self.task_queue.clear()
    
    def get_schedule_summary(self) -> str:
        """Generate schedule summary."""
        summary_prompt = ChatPromptTemplate.from_messages([
            ("system", "Summarize this task allocation and schedule."),
            ("user", """Assignments:
{assignments}

Agent Loads:
{agent_loads}

Provide:
1. Overall load distribution
2. Potential bottlenecks
3. Optimization suggestions""")
        ])
        
        assignments_str = "\n".join([
            f"Task {task_id} -> Agent {agent_id}"
            for task_id, agent_id in self.assignments.items()
        ])
        
        agent_loads_str = "\n".join([
            f"{agent.name}: {agent.current_load}/{agent.max_capacity} ({agent.current_load/agent.max_capacity*100:.1f}%)"
            for agent in self.agents
        ])
        
        chain = summary_prompt | self.llm | StrOutputParser()
        return chain.invoke({
            "assignments": assignments_str,
            "agent_loads": agent_loads_str
        })
    
    def rebalance_load(self):
        """Rebalance load across agents if needed."""
        avg_load = sum(a.current_load for a in self.agents) / len(self.agents)
        overloaded = [a for a in self.agents if a.current_load > avg_load * 1.5]
        underloaded = [a for a in self.agents if a.current_load < avg_load * 0.5]
        
        if overloaded and underloaded:
            print("\n=== Load Rebalancing ===")
            print(f"Overloaded agents: {[a.name for a in overloaded]}")
            print(f"Underloaded agents: {[a.name for a in underloaded]}")
            print("Recommending task redistribution...")


def demonstrate_task_allocation():
    """Demonstrate task allocation and scheduling pattern."""
    print("=== Task Allocation & Scheduling Pattern ===\n")
    
    scheduler = TaskAllocationScheduler()
    
    # Register agents
    print("1. Registering Agents")
    print("-" * 50)
    
    scheduler.register_agent(Agent(
        id="agent1",
        name="Alice",
        skills=["python", "data_analysis", "machine_learning"],
        max_capacity=100,
        performance_history=[0.85, 0.90, 0.88]
    ))
    
    scheduler.register_agent(Agent(
        id="agent2",
        name="Bob",
        skills=["javascript", "frontend", "ui_design"],
        max_capacity=100,
        performance_history=[0.92, 0.89, 0.91]
    ))
    
    scheduler.register_agent(Agent(
        id="agent3",
        name="Charlie",
        skills=["devops", "cloud", "kubernetes"],
        max_capacity=100,
        performance_history=[0.88, 0.87, 0.90]
    ))
    
    scheduler.register_agent(Agent(
        id="agent4",
        name="Diana",
        skills=["python", "backend", "database"],
        max_capacity=100,
        performance_history=[0.90, 0.93, 0.91]
    ))
    
    print()
    
    # Add tasks
    print("2. Adding Tasks to Queue")
    print("-" * 50)
    
    scheduler.add_task(Task(
        id="task1",
        description="Build machine learning model for customer churn prediction",
        required_skills=["python", "machine_learning", "data_analysis"],
        priority=9,
        estimated_duration=40
    ))
    
    scheduler.add_task(Task(
        id="task2",
        description="Design and implement responsive dashboard UI",
        required_skills=["javascript", "frontend", "ui_design"],
        priority=8,
        estimated_duration=30
    ))
    
    scheduler.add_task(Task(
        id="task3",
        description="Set up CI/CD pipeline with Kubernetes deployment",
        required_skills=["devops", "cloud", "kubernetes"],
        priority=7,
        estimated_duration=35
    ))
    
    scheduler.add_task(Task(
        id="task4",
        description="Optimize database queries and add indexes",
        required_skills=["database", "backend", "python"],
        priority=8,
        estimated_duration=25
    ))
    
    scheduler.add_task(Task(
        id="task5",
        description="Implement user authentication system",
        required_skills=["backend", "python", "security"],
        priority=9,
        estimated_duration=45
    ))
    
    print()
    
    # Allocate tasks
    print("3. Allocating Tasks to Agents")
    print("-" * 50)
    scheduler.allocate_tasks()
    
    # Schedule summary
    print("4. Schedule Summary")
    print("-" * 50)
    summary = scheduler.get_schedule_summary()
    print(summary)
    print()
    
    # Check for rebalancing needs
    print("5. Load Balancing Check")
    print("-" * 50)
    scheduler.rebalance_load()
    
    print("\n=== Summary ===")
    print(f"Total assignments: {len(scheduler.assignments)}")
    print("Task allocation demonstrated with:")
    print("- Skill-based matching")
    print("- Load-aware assignment")
    print("- Priority-based scheduling")
    print("- Performance history consideration")
    print("- Load balancing recommendations")


if __name__ == "__main__":
    demonstrate_task_allocation()

    

Agentic Design Pattern: Task Allocation Scheduling

Pattern 119: Task Allocation Scheduling – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Task Allocation Scheduling pattern implementation.

---

Pattern Description

Pattern 119: Task Allocation & Scheduling

Description: Assigns tasks to agents based on capabilities, load, and constraints using auction-based or optimal assignment algorithms.

Components: – Task queue – Agent capabilities – Allocation algorithm – Load balancing

Use Cases: – Multi-agent systems – Distributed computing – Resource optimization

LangChain Implementation: Uses LLM-based task analysis and agent matching with LangGraph for orchestration.

---

File Statistics

• Total Lines: 342
• Classes: 3
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 119: Task Allocation & Scheduling

Description:
    Assigns tasks to agents based on capabilities, load, and constraints
    using auction-based or optimal assignment algorithms.

Components:
    - Task queue
    - Agent capabilities
    - Allocation algorithm
    - Load balancing

Use Cases:
    - Multi-agent systems
    - Distributed computing
    - Resource optimization

LangChain Implementation:
    Uses LLM-based task analysis and agent matching with LangGraph for orchestration.
"""

Purpose: This module implements the Task Allocation Scheduling pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import datetime.datetime
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: 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)

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: Task

Lines 37-44

Purpose:

Represents a task to be allocated.

---

Class: Agent

Lines 48-59

Purpose:

Represents an agent with capabilities.

Methods:

__post_init__(self)

Lines 57-59

Purpose: Post Init operation.

Parameters:

---

Class: TaskAllocationScheduler

Lines 62-220

Purpose:

Schedules and allocates tasks to agents optimally.

Methods:

__init__(self, model_name)

Lines 65-69

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

register_agent(self, agent)

Lines 71-74

Purpose:

Register an agent in the system.

Parameters: – agent: Agent

add_task(self, task)

Lines 76-79

Purpose:

Add task to queue.

Parameters: – task: Task

analyze_task_requirements(self, task)

Lines 81-102

Purpose:

Analyze task requirements using LLM.

Parameters: – task: Task

Returns: <ast.Subscript object at 0x000001A06D765F50>

calculate_agent_score(self, agent, task)

Lines 104-141

Purpose:

Calculate how well an agent matches a task.

Parameters: – agent: Agent – task: Task

Returns: float

allocate_tasks(self)

Lines 143-176

Purpose:

Allocate tasks to agents using optimal matching.

Parameters:

get_schedule_summary(self)

Lines 178-208

Purpose:

Generate schedule summary.

Parameters:

Returns: str

rebalance_load(self)

Lines 210-220

Purpose:

Rebalance load across agents if needed.

Parameters:

---

Functions

demonstrate_task_allocation()

Lines 223-337

Purpose:

Demonstrate task allocation and scheduling pattern.

---

Key Concepts Summary

Pattern Overview

The Task Allocation Scheduling 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

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. 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 .env files 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

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. 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

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 118
• Pattern 120

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 Task Allocation Scheduling 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: 119_task_allocation_scheduling.py
Total Lines: 342
Pattern Category: AI Agent Design Pattern #119

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 118: Curiosity-Driven Exploration

Description:
    Agent explores environment based on intrinsic motivation driven by
    novelty, information gain, and prediction error.

Components:
    - Novelty detection
    - Information gain calculation
    - Exploration strategy
    - Intrinsic reward

Use Cases:
    - Open-ended learning
    - Exploration tasks
    - Discovery systems

LangChain Implementation:
    Uses LLM to generate curious questions and exploration strategies based on
    knowledge gaps and novelty.
"""

import os
from typing import List, Dict, Any, Set
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 CuriosityDrivenAgent:
    """Agent that explores based on intrinsic curiosity."""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.8)
        self.knowledge_base = set()
        self.explored_topics = []
        self.curiosity_log = []
        
    def assess_novelty(self, topic: str) -> float:
        """Assess how novel a topic is based on existing knowledge."""
        novelty_prompt = ChatPromptTemplate.from_messages([
            ("system", "Rate the novelty of this topic given existing knowledge. Return a score from 0.0 (completely known) to 1.0 (completely novel)."),
            ("user", """Topic: {topic}

Known Knowledge:
{knowledge}

Novelty score (0.0-1.0):""")
        ])
        
        knowledge_str = "\n".join([f"- {k}" for k in list(self.knowledge_base)[:20]]) or "Empty"
        
        chain = novelty_prompt | self.llm | StrOutputParser()
        response = chain.invoke({"topic": topic, "knowledge": knowledge_str})
        
        try:
            score = float(response.strip())
            return max(0.0, min(1.0, score))
        except:
            return 0.5
    
    def generate_curious_questions(self, context: str) -> List[str]:
        """Generate questions driven by curiosity about knowledge gaps."""
        curiosity_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a curious agent exploring knowledge.
Generate 3-5 interesting questions that would expand understanding in unexplored directions.
Focus on:
- Novel connections
- Surprising implications
- Unexplored aspects
- Deep mechanisms"""),
            ("user", """Current Context: {context}

Known Knowledge:
{knowledge}

Generate curious questions:""")
        ])
        
        knowledge_str = "\n".join([f"- {k}" for k in list(self.knowledge_base)[-10:]]) or "Empty"
        
        chain = curiosity_prompt | self.llm | StrOutputParser()
        response = chain.invoke({"context": context, "knowledge": knowledge_str})
        
        # Parse questions
        questions = [line.strip('- ').strip() for line in response.split('\n') if line.strip()]
        return [q for q in questions if '?' in q][:5]
    
    def calculate_information_gain(self, potential_exploration: str) -> float:
        """Calculate expected information gain from exploring a topic."""
        gain_prompt = ChatPromptTemplate.from_messages([
            ("system", "Estimate information gain (0.0-1.0) from exploring this topic given current knowledge."),
            ("user", """Potential Exploration: {exploration}

Current Knowledge:
{knowledge}

Information gain score (0.0-1.0):""")
        ])
        
        knowledge_str = "\n".join([f"- {k}" for k in list(self.knowledge_base)[-15:]]) or "Empty"
        
        chain = gain_prompt | self.llm | StrOutputParser()
        response = chain.invoke({"exploration": potential_exploration, "knowledge": knowledge_str})
        
        try:
            score = float(response.strip())
            return max(0.0, min(1.0, score))
        except:
            return 0.5
    
    def explore(self, topic: str) -> Dict[str, Any]:
        """Explore a topic and gain knowledge."""
        print(f"\n🔍 Exploring: {topic}")
        
        # Assess novelty
        novelty = self.assess_novelty(topic)
        print(f"  Novelty: {novelty:.2f}")
        
        # Calculate information gain
        info_gain = self.calculate_information_gain(topic)
        print(f"  Information Gain: {info_gain:.2f}")
        
        # Intrinsic reward
        intrinsic_reward = (novelty + info_gain) / 2
        print(f"  Intrinsic Reward: {intrinsic_reward:.2f}")
        
        # Explore the topic
        exploration_prompt = ChatPromptTemplate.from_messages([
            ("system", "Explore this topic deeply and discover interesting insights."),
            ("user", "Topic: {topic}\n\nProvide interesting discoveries:")
        ])
        
        chain = exploration_prompt | self.llm | StrOutputParser()
        discoveries = chain.invoke({"topic": topic})
        
        # Update knowledge base
        self.knowledge_base.add(topic)
        self.explored_topics.append(topic)
        
        result = {
            "topic": topic,
            "novelty": novelty,
            "information_gain": info_gain,
            "intrinsic_reward": intrinsic_reward,
            "discoveries": discoveries
        }
        
        self.curiosity_log.append(result)
        
        print(f"\n  Discoveries:")
        for line in discoveries.split('\n')[:3]:
            if line.strip():
                print(f"    • {line.strip()}")
        
        return result
    
    def choose_next_exploration(self, candidates: List[str]) -> str:
        """Choose next exploration target based on curiosity."""
        print("\n📊 Evaluating exploration candidates...")
        
        scores = []
        for candidate in candidates:
            novelty = self.assess_novelty(candidate)
            info_gain = self.calculate_information_gain(candidate)
            curiosity_score = (novelty + info_gain) / 2
            scores.append((candidate, curiosity_score))
            print(f"  {candidate}: {curiosity_score:.2f}")
        
        # Choose highest curiosity score
        best_candidate = max(scores, key=lambda x: x[1])
        print(f"\n✨ Selected: {best_candidate[0]} (score: {best_candidate[1]:.2f})")
        
        return best_candidate[0]
    
    def get_exploration_summary(self) -> str:
        """Summarize exploration journey."""
        summary_prompt = ChatPromptTemplate.from_messages([
            ("system", "Summarize this curiosity-driven exploration journey."),
            ("user", """Explored Topics:
{topics}

Curiosity Log:
{log}

Provide a summary of:
1. Key discoveries
2. Exploration patterns
3. Knowledge growth
4. Most rewarding explorations""")
        ])
        
        topics = "\n".join([f"{i+1}. {t}" for i, t in enumerate(self.explored_topics)])
        log = "\n".join([
            f"Topic: {entry['topic']}, Reward: {entry['intrinsic_reward']:.2f}"
            for entry in self.curiosity_log
        ])
        
        chain = summary_prompt | self.llm | StrOutputParser()
        return chain.invoke({"topics": topics, "log": log})


def demonstrate_curiosity_driven_exploration():
    """Demonstrate curiosity-driven exploration pattern."""
    print("=== Curiosity-Driven Exploration Pattern ===\n")
    
    agent = CuriosityDrivenAgent()
    
    # Initial topic
    print("1. Starting Exploration from Initial Topic")
    print("-" * 50)
    agent.explore("Quantum Computing")
    
    # Generate curious questions
    print("\n2. Generating Curious Questions")
    print("-" * 50)
    questions = agent.generate_curious_questions("Quantum Computing")
    print("Curious questions generated:")
    for i, q in enumerate(questions, 1):
        print(f"{i}. {q}")
    
    # Explore based on curiosity
    print("\n3. Curiosity-Driven Exploration")
    print("-" * 50)
    
    # Candidate topics
    candidates = [
        "Quantum Entanglement Applications",
        "Quantum Error Correction",
        "Superconducting Qubits",
        "Quantum Algorithms for Chemistry"
    ]
    
    # Choose and explore
    for i in range(3):
        print(f"\n--- Exploration Round {i+1} ---")
        next_topic = agent.choose_next_exploration(candidates)
        agent.explore(next_topic)
        
        # Remove explored topic and add related ones
        candidates.remove(next_topic)
        
        # Generate new candidates based on discoveries
        if i < 2:
            new_candidates_prompt = ChatPromptTemplate.from_messages([
                ("system", "Suggest 2 related unexplored topics based on recent discovery."),
                ("user", "Recent exploration: {topic}\n\nSuggest new topics:")
            ])
            chain = new_candidates_prompt | agent.llm | StrOutputParser()
            new_suggestions = chain.invoke({"topic": next_topic})
            new_topics = [line.strip('- ').strip() for line in new_suggestions.split('\n') if line.strip()][:2]
            candidates.extend(new_topics)
    
    # Summary
    print("\n4. Exploration Journey Summary")
    print("-" * 50)
    summary = agent.get_exploration_summary()
    print(summary)
    
    print("\n=== Summary ===")
    print(f"Total explorations: {len(agent.explored_topics)}")
    print(f"Knowledge base size: {len(agent.knowledge_base)}")
    print(f"Average intrinsic reward: {sum(e['intrinsic_reward'] for e in agent.curiosity_log) / len(agent.curiosity_log):.2f}")
    print("\nCuriosity-driven exploration demonstrated with:")
    print("- Novelty assessment")
    print("- Information gain calculation")
    print("- Intrinsic reward-based decisions")
    print("- Autonomous exploration strategy")
    print("- Continuous knowledge expansion")


if __name__ == "__main__":
    demonstrate_curiosity_driven_exploration()

    

Agentic Design Pattern: Curiosity Driven Exploration

Pattern 118: Curiosity Driven Exploration – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Curiosity Driven Exploration pattern implementation.

---

Pattern Description

Pattern 118: Curiosity-Driven Exploration

Description: Agent explores environment based on intrinsic motivation driven by novelty, information gain, and prediction error.

Components: – Novelty detection – Information gain calculation – Exploration strategy – Intrinsic reward

Use Cases: – Open-ended learning – Exploration tasks – Discovery systems

LangChain Implementation: Uses LLM to generate curious questions and exploration strategies based on knowledge gaps and novelty.

---

File Statistics

• Total Lines: 279
• Classes: 1
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 118: Curiosity-Driven Exploration

Description:
    Agent explores environment based on intrinsic motivation driven by
    novelty, information gain, and prediction error.

Components:
    - Novelty detection
    - Information gain calculation
    - Exploration strategy
    - Intrinsic reward

Use Cases:
    - Open-ended learning
    - Exploration tasks
    - Discovery systems

LangChain Implementation:
    Uses LLM to generate curious questions and exploration strategies based on
    knowledge gaps and novelty.
"""

Purpose: This module implements the Curiosity Driven Exploration pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Set

Purpose: Core Python functionality: – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: 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.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: CuriosityDrivenAgent

Lines 35-205

Purpose:

Agent that explores based on intrinsic curiosity.

Methods:

__init__(self, model_name)

Lines 38-42

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

assess_novelty(self, topic)

Lines 44-65

Purpose:

Assess how novel a topic is based on existing knowledge.

Parameters: – topic: Topic

Returns: float

generate_curious_questions(self, context)

Lines 67-92

Purpose:

Generate questions driven by curiosity about knowledge gaps.

Parameters: – context: Context

Returns: <ast.Subscript object at 0x000001A06D7CC0D0>

calculate_information_gain(self, potential_exploration)

Lines 94-115

Purpose:

Calculate expected information gain from exploring a topic.

Parameters: – potential_exploration: Potential exploration

Returns: float

explore(self, topic)

Lines 117-161

Purpose:

Explore a topic and gain knowledge.

Parameters: – topic: Topic

Returns: <ast.Subscript object at 0x000001A06D88B210>

choose_next_exploration(self, candidates)

Lines 163-179

Purpose:

Choose next exploration target based on curiosity.

Parameters: – candidates: Candidates

Returns: str

get_exploration_summary(self)

Lines 181-205

Purpose:

Summarize exploration journey.

Parameters:

Returns: str

---

Functions

demonstrate_curiosity_driven_exploration()

Lines 208-274

Purpose:

Demonstrate curiosity-driven exploration pattern.

---

Key Concepts Summary

Pattern Overview

The Curiosity Driven Exploration 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

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. 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 .env files 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

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. 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

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 117
• Pattern 119

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 Curiosity Driven Exploration 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: 118_curiosity_driven_exploration.py
Total Lines: 279
Pattern Category: AI Agent Design Pattern #118

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 117: Imitation Learning

Description:
    Agent learns by observing and imitating expert behavior through
    behavioral cloning and inverse reinforcement learning approaches.

Components:
    - Expert demonstrations
    - Behavioral cloning
    - Policy learning

Use Cases:
    - Robotics
    - Complex skill acquisition
    - Learning from examples

LangChain Implementation:
    Uses few-shot learning and demonstration-based prompting to imitate expert behavior.
"""

import os
from typing import List, Dict, Any
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


class ImitationLearningAgent:
    """Agent that learns by observing and imitating expert demonstrations."""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.expert_demonstrations = []
        self.learned_policy = None
        
    def observe_expert(self, situation: str, expert_action: str, reasoning: str):
        """Observe and record expert demonstration."""
        demonstration = {
            "situation": situation,
            "action": expert_action,
            "reasoning": reasoning
        }
        self.expert_demonstrations.append(demonstration)
        print(f"Observed demonstration {len(self.expert_demonstrations)}:")
        print(f"  Situation: {situation}")
        print(f"  Action: {expert_action}")
        print(f"  Reasoning: {reasoning}")
        print()
        
    def learn_from_demonstrations(self):
        """Learn policy from all observed demonstrations."""
        learning_prompt = ChatPromptTemplate.from_messages([
            ("system", "Analyze these expert demonstrations and extract the underlying policy, patterns, and decision-making strategy."),
            ("user", """Expert Demonstrations:
{demonstrations}

Extract:
1. Common patterns in decision-making
2. Key factors considered
3. General policy rules
4. Decision strategy""")
        ])
        
        demos_text = "\n\n".join([
            f"Demo {i+1}:\nSituation: {d['situation']}\nAction: {d['action']}\nReasoning: {d['reasoning']}"
            for i, d in enumerate(self.expert_demonstrations)
        ])
        
        chain = learning_prompt | self.llm | StrOutputParser()
        self.learned_policy = chain.invoke({"demonstrations": demos_text})
        
        print("=== Learned Policy ===")
        print(self.learned_policy)
        print()
        
    def imitate_expert(self, new_situation: str) -> Dict[str, str]:
        """Apply learned policy to a new situation."""
        if not self.learned_policy:
            return {"error": "No policy learned yet. Need more demonstrations."}
        
        # Create few-shot prompt with demonstrations
        example_template = """Situation: {situation}
Expert Action: {action}
Reasoning: {reasoning}"""
        
        few_shot_prompt = FewShotChatMessagePromptTemplate(
            example_prompt=ChatPromptTemplate.from_messages([
                ("user", "Situation: {situation}"),
                ("assistant", "Action: {action}\nReasoning: {reasoning}")
            ]),
            examples=self.expert_demonstrations,
        )
        
        full_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are imitating an expert's decision-making style.

Learned Policy:
{learned_policy}

Use the learned policy and demonstrated patterns to decide actions for new situations."""),
            few_shot_prompt,
            ("user", "Situation: {new_situation}\n\nWhat action should be taken? Provide action and reasoning.")
        ])
        
        chain = full_prompt | self.llm | StrOutputParser()
        response = chain.invoke({
            "learned_policy": self.learned_policy,
            "new_situation": new_situation
        })
        
        # Parse response
        lines = response.split('\n')
        action = ""
        reasoning = ""
        
        for line in lines:
            if line.startswith("Action:"):
                action = line.replace("Action:", "").strip()
            elif line.startswith("Reasoning:"):
                reasoning = line.replace("Reasoning:", "").strip()
            elif action and not reasoning:
                reasoning = line.strip()
        
        return {
            "situation": new_situation,
            "action": action or response,
            "reasoning": reasoning or "Based on learned policy"
        }
    
    def evaluate_imitation(self, test_situation: str, expected_action: str) -> bool:
        """Evaluate how well the agent imitates expert behavior."""
        result = self.imitate_expert(test_situation)
        
        print(f"Test Situation: {test_situation}")
        print(f"Expected Action: {expected_action}")
        print(f"Agent Action: {result['action']}")
        print(f"Reasoning: {result['reasoning']}")
        
        # Simple similarity check
        similarity = expected_action.lower() in result['action'].lower()
        print(f"Match: {'✓' if similarity else '✗'}")
        print()
        
        return similarity


def demonstrate_imitation_learning():
    """Demonstrate imitation learning pattern."""
    print("=== Imitation Learning Pattern ===\n")
    
    agent = ImitationLearningAgent()
    
    # Scenario: Learning customer service expert behavior
    print("1. Observing Expert Demonstrations")
    print("-" * 50)
    
    # Demonstration 1: Angry customer
    agent.observe_expert(
        situation="Customer is angry about a delayed delivery",
        expert_action="Apologize sincerely, explain the situation, offer compensation (discount or free shipping on next order)",
        reasoning="Acknowledge emotions first, show empathy, provide solution to restore trust"
    )
    
    # Demonstration 2: Confused customer
    agent.observe_expert(
        situation="Customer is confused about how to use a product feature",
        expert_action="Ask clarifying questions, provide step-by-step guidance, offer visual aids or video tutorial",
        reasoning="Understand specific confusion point, break down into simple steps, provide multiple learning formats"
    )
    
    # Demonstration 3: Billing inquiry
    agent.observe_expert(
        situation="Customer questions an unexpected charge on their bill",
        expert_action="Review the charge details, explain clearly what it's for, provide itemized breakdown if needed",
        reasoning="Be transparent about charges, ensure customer understands, address concerns directly"
    )
    
    # Demonstration 4: Feature request
    agent.observe_expert(
        situation="Customer requests a feature that doesn't exist",
        expert_action="Thank them for feedback, explain current capabilities, document request for product team, suggest workaround if possible",
        reasoning="Value customer input, manage expectations, show commitment to improvement, help immediately if possible"
    )
    
    print("✓ Observed 4 expert demonstrations\n")
    
    # Learn from demonstrations
    print("2. Learning Policy from Demonstrations")
    print("-" * 50)
    agent.learn_from_demonstrations()
    
    # Test imitation on new situations
    print("3. Testing Imitation on New Situations")
    print("-" * 50)
    
    # Test 1: Product defect
    print("Test 1: Product Defect Situation")
    result = agent.imitate_expert("Customer reports a defective product that stopped working after one week")
    print(f"Situation: {result['situation']}")
    print(f"Action: {result['action']}")
    print(f"Reasoning: {result['reasoning']}")
    print()
    
    # Test 2: Refund request
    print("Test 2: Refund Request Situation")
    result = agent.imitate_expert("Customer wants a refund for a product they bought 2 months ago and barely used")
    print(f"Situation: {result['situation']}")
    print(f"Action: {result['action']}")
    print(f"Reasoning: {result['reasoning']}")
    print()
    
    # Test 3: Complex technical issue
    print("Test 3: Technical Issue Situation")
    result = agent.imitate_expert("Customer experiencing intermittent connectivity issues with the app")
    print(f"Situation: {result['situation']}")
    print(f"Action: {result['action']}")
    print(f"Reasoning: {result['reasoning']}")
    print()
    
    # Evaluation
    print("4. Evaluating Imitation Quality")
    print("-" * 50)
    agent.evaluate_imitation(
        "Customer complaint about rude staff member",
        "Apologize for experience, escalate to manager, ensure follow-up"
    )
    
    print("=== Summary ===")
    print("Imitation learning demonstrated with:")
    print("- Expert demonstration observation")
    print("- Policy learning from demonstrations")
    print("- Behavioral cloning for new situations")
    print("- Few-shot learning approach")
    print("- Pattern extraction from expert behavior")


if __name__ == "__main__":
    demonstrate_imitation_learning()

    

Agentic Design Pattern: Imitation Learning

Pattern 117: Imitation Learning – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Imitation Learning pattern implementation.

---

Pattern Description

Pattern 117: Imitation Learning

Description: Agent learns by observing and imitating expert behavior through behavioral cloning and inverse reinforcement learning approaches.

Components: – Expert demonstrations – Behavioral cloning – Policy learning

Use Cases: – Robotics – Complex skill acquisition – Learning from examples

LangChain Implementation: Uses few-shot learning and demonstration-based prompting to imitate expert behavior.

---

File Statistics

• Total Lines: 244
• Classes: 1
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 117: Imitation Learning

Description:
    Agent learns by observing and imitating expert behavior through
    behavioral cloning and inverse reinforcement learning approaches.

Components:
    - Expert demonstrations
    - Behavioral cloning
    - Policy learning

Use Cases:
    - Robotics
    - Complex skill acquisition
    - Learning from examples

LangChain Implementation:
    Uses few-shot learning and demonstration-based prompting to imitate expert behavior.
"""

Purpose: This module implements the Imitation Learning pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import langchain_core.prompts.FewShotChatMessagePromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List

Purpose: Core Python functionality: – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.prompts.FewShotChatMessagePromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: 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)

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: ImitationLearningAgent

Lines 33-149

Purpose:

Agent that learns by observing and imitating expert demonstrations.

Methods:

__init__(self, model_name)

Lines 36-39

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

observe_expert(self, situation, expert_action, reasoning)

Lines 41-53

Purpose:

Observe and record expert demonstration.

Parameters: – situation: Situation – expert_action: Expert action – reasoning: Reasoning

learn_from_demonstrations(self)

Lines 55-79

Purpose:

Learn policy from all observed demonstrations.

Parameters:

imitate_expert(self, new_situation)

Lines 81-133

Purpose:

Apply learned policy to a new situation.

Parameters: – new_situation: New situation

Returns: <ast.Subscript object at 0x000001A06D7F4090>

evaluate_imitation(self, test_situation, expected_action)

Lines 135-149

Purpose:

Evaluate how well the agent imitates expert behavior.

Parameters: – test_situation: Test situation – expected_action: Expected action

Returns: bool

---

Functions

demonstrate_imitation_learning()

Lines 152-239

Purpose:

Demonstrate imitation learning pattern.

---

Key Concepts Summary

Pattern Overview

The Imitation Learning 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

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. 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 .env files 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

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. 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

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 116
• Pattern 118

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 Imitation Learning 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: 117_imitation_learning.py
Total Lines: 244
Pattern Category: AI Agent Design Pattern #117

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 116: Multi-Task Learning

Description:
    Agent learns multiple related tasks simultaneously using shared representations
    with task-specific heads for specialization.

Components:
    - Shared representations
    - Task-specific heads
    - Multi-task optimization

Use Cases:
    - Related task sets
    - Efficient training
    - Knowledge sharing

LangChain Implementation:
    Uses multi-task prompt templates and shared context for learning across tasks.
"""

import os
from typing import List, Dict, Any
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 MultiTaskLearningAgent:
    """Agent that learns multiple related tasks simultaneously."""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.tasks = {}
        self.shared_knowledge = {}
        self.task_performance = {}
        
    def register_task(self, task_name: str, task_description: str, examples: List[Dict[str, str]]):
        """Register a new task for multi-task learning."""
        self.tasks[task_name] = {
            "description": task_description,
            "examples": examples
        }
        self.task_performance[task_name] = []
        
    def learn_from_shared_representations(self):
        """Extract shared knowledge from all registered tasks."""
        shared_prompt = ChatPromptTemplate.from_messages([
            ("system", "Analyze these tasks and identify shared patterns, concepts, and skills that apply across them."),
            ("user", "Tasks:\n{tasks}\n\nIdentify shared knowledge:")
        ])
        
        tasks_description = "\n\n".join([
            f"{name}: {info['description']}\nExamples: {info['examples']}"
            for name, info in self.tasks.items()
        ])
        
        chain = shared_prompt | self.llm | StrOutputParser()
        self.shared_knowledge["patterns"] = chain.invoke({"tasks": tasks_description})
        
        print("Shared Knowledge Extracted:")
        print(self.shared_knowledge["patterns"])
        print()
        
    def execute_task(self, task_name: str, input_data: str) -> str:
        """Execute a specific task using shared knowledge and task-specific head."""
        if task_name not in self.tasks:
            return f"Task '{task_name}' not registered"
        
        task_info = self.tasks[task_name]
        
        # Create task-specific prompt with shared knowledge
        task_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a multi-task learning agent.

Shared Knowledge (applies to all tasks):
{shared_knowledge}

Current Task: {task_name}
Description: {task_description}

Examples:
{examples}

Use both shared knowledge and task-specific patterns."""),
            ("user", "{input}")
        ])
        
        examples_str = "\n".join([
            f"Input: {ex['input']}\nOutput: {ex['output']}"
            for ex in task_info['examples']
        ])
        
        chain = task_prompt | self.llm | StrOutputParser()
        
        result = chain.invoke({
            "shared_knowledge": self.shared_knowledge.get("patterns", "None extracted yet"),
            "task_name": task_name,
            "task_description": task_info['description'],
            "examples": examples_str,
            "input": input_data
        })
        
        return result
    
    def evaluate_task(self, task_name: str, test_cases: List[Dict[str, str]]):
        """Evaluate performance on a specific task."""
        if task_name not in self.tasks:
            print(f"Task '{task_name}' not registered")
            return
        
        print(f"\n=== Evaluating Task: {task_name} ===")
        correct = 0
        total = len(test_cases)
        
        for case in test_cases:
            result = self.execute_task(task_name, case['input'])
            is_correct = case['expected'].lower() in result.lower()
            if is_correct:
                correct += 1
            print(f"Input: {case['input']}")
            print(f"Expected: {case['expected']}")
            print(f"Got: {result}")
            print(f"✓" if is_correct else "✗")
            print()
        
        accuracy = correct / total if total > 0 else 0
        self.task_performance[task_name].append(accuracy)
        print(f"Accuracy: {accuracy:.2%}\n")
        
    def get_task_correlations(self) -> str:
        """Analyze how tasks benefit from shared learning."""
        correlation_prompt = ChatPromptTemplate.from_messages([
            ("system", "Analyze how these tasks benefit from shared multi-task learning."),
            ("user", """Tasks and Performance:
{task_info}

Shared Knowledge:
{shared_knowledge}

Explain how tasks benefit from each other and which skills transfer across tasks.""")
        ])
        
        task_info = "\n".join([
            f"{name}: {self.tasks[name]['description']} - Performance: {self.task_performance[name]}"
            for name in self.tasks
        ])
        
        chain = correlation_prompt | self.llm | StrOutputParser()
        return chain.invoke({
            "task_info": task_info,
            "shared_knowledge": self.shared_knowledge.get("patterns", "Not extracted")
        })


def demonstrate_multi_task_learning():
    """Demonstrate multi-task learning pattern."""
    print("=== Multi-Task Learning Pattern ===\n")
    
    agent = MultiTaskLearningAgent()
    
    # Register related NLP tasks
    print("1. Registering Multiple Related Tasks")
    print("-" * 50)
    
    # Task 1: Sentiment Analysis
    agent.register_task(
        "sentiment_analysis",
        "Classify the sentiment of text as positive, negative, or neutral",
        [
            {"input": "I love this product!", "output": "positive"},
            {"input": "This is terrible.", "output": "negative"},
            {"input": "It's okay, nothing special.", "output": "neutral"}
        ]
    )
    
    # Task 2: Text Classification
    agent.register_task(
        "text_classification",
        "Classify text into categories: technology, sports, politics, entertainment",
        [
            {"input": "New AI breakthrough announced", "output": "technology"},
            {"input": "Championship game tonight", "output": "sports"},
            {"input": "Election results are in", "output": "politics"}
        ]
    )
    
    # Task 3: Intent Detection
    agent.register_task(
        "intent_detection",
        "Detect user intent: question, command, statement, request",
        [
            {"input": "What time is it?", "output": "question"},
            {"input": "Please send the report.", "output": "command"},
            {"input": "I think it will rain.", "output": "statement"}
        ]
    )
    
    print("✓ Registered 3 related NLP tasks\n")
    
    # Extract shared knowledge
    print("2. Learning Shared Representations")
    print("-" * 50)
    agent.learn_from_shared_representations()
    
    # Test on sentiment analysis
    print("3. Testing on Sentiment Analysis Task")
    print("-" * 50)
    test_cases = [
        {"input": "This is amazing and wonderful!", "expected": "positive"},
        {"input": "Worst experience ever", "expected": "negative"},
    ]
    agent.evaluate_task("sentiment_analysis", test_cases)
    
    # Test on text classification
    print("4. Testing on Text Classification Task")
    print("-" * 50)
    test_cases = [
        {"input": "Latest smartphone release announced", "expected": "technology"},
        {"input": "Team wins the finals", "expected": "sports"},
    ]
    agent.evaluate_task("text_classification", test_cases)
    
    # Analyze task correlations
    print("5. Analyzing Task Correlations and Knowledge Transfer")
    print("-" * 50)
    correlations = agent.get_task_correlations()
    print(correlations)
    print()
    
    print("=== Summary ===")
    print("Multi-task learning demonstrated with:")
    print("- Multiple related NLP tasks")
    print("- Shared knowledge extraction")
    print("- Task-specific execution with shared representations")
    print("- Knowledge transfer analysis")


if __name__ == "__main__":
    demonstrate_multi_task_learning()

    

Agentic Design Pattern: Multi Task Learning

Pattern 116: Multi Task Learning – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Multi Task Learning pattern implementation.

---

Pattern Description

Pattern 116: Multi-Task Learning

Description: Agent learns multiple related tasks simultaneously using shared representations with task-specific heads for specialization.

Components: – Shared representations – Task-specific heads – Multi-task optimization

Use Cases: – Related task sets – Efficient training – Knowledge sharing

LangChain Implementation: Uses multi-task prompt templates and shared context for learning across tasks.

---

File Statistics

• Total Lines: 245
• Classes: 1
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 116: Multi-Task Learning

Description:
    Agent learns multiple related tasks simultaneously using shared representations
    with task-specific heads for specialization.

Components:
    - Shared representations
    - Task-specific heads
    - Multi-task optimization

Use Cases:
    - Related task sets
    - Efficient training
    - Knowledge sharing

LangChain Implementation:
    Uses multi-task prompt templates and shared context for learning across tasks.
"""

Purpose: This module implements the Multi Task Learning pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List

Purpose: Core Python functionality: – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: 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)

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: MultiTaskLearningAgent

Lines 33-157

Purpose:

Agent that learns multiple related tasks simultaneously.

Methods:

__init__(self, model_name)

Lines 36-40

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

register_task(self, task_name, task_description, examples)

Lines 42-48

Purpose:

Register a new task for multi-task learning.

Parameters: – task_name: Task name – task_description: Task description – examples: Examples

learn_from_shared_representations(self)

Lines 50-67

Purpose:

Extract shared knowledge from all registered tasks.

Parameters:

execute_task(self, task_name, input_data)

Lines 69-108

Purpose:

Execute a specific task using shared knowledge and task-specific head.

Parameters: – task_name: Task name – input_data: Input data

Returns: str

evaluate_task(self, task_name, test_cases)

Lines 110-133

Purpose:

Evaluate performance on a specific task.

Parameters: – task_name: Task name – test_cases: Test cases

get_task_correlations(self)

Lines 135-157

Purpose:

Analyze how tasks benefit from shared learning.

Parameters:

Returns: str

---

Functions

demonstrate_multi_task_learning()

Lines 160-240

Purpose:

Demonstrate multi-task learning pattern.

---

Key Concepts Summary

Pattern Overview

The Multi Task Learning 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

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. 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 .env files 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

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. 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

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 115
• Pattern 117

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 Multi Task Learning 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: 116_multi_task_learning.py
Total Lines: 245
Pattern Category: AI Agent Design Pattern #116

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 115: Transfer Learning Agent

Description:
    Transfer Learning enables agents to leverage knowledge learned from one
    task or domain to improve performance on related tasks. This pattern
    identifies transferable knowledge, adapts it to new contexts, and combines
    it with task-specific learning for faster adaptation and better performance.
    
    Essential for agents that need to work across multiple related domains or
    tasks, transfer learning reduces the data and time needed to achieve good
    performance in new areas.

Key Components:
    1. Source Knowledge Extractor: Identifies transferable knowledge
    2. Domain Adapter: Adapts knowledge to new domain
    3. Fine-tuner: Specializes for target task
    4. Transfer Strategy Selector: Chooses transfer approach
    5. Performance Evaluator: Assesses transfer effectiveness

Transfer Strategies:
    - Feature Transfer: Use learned representations
    - Parameter Transfer: Transfer model parameters
    - Instance Transfer: Leverage source examples
    - Relational Transfer: Transfer relationships
    - Meta-transfer: Transfer learning strategies

Use Cases:
    - Multi-domain chatbots
    - Cross-lingual applications
    - Domain adaptation (e.g., medical to legal)
    - Few-shot learning in new domains
    - Skill transfer in robotics

LangChain Implementation:
    Uses prompt engineering, few-shot examples, and domain adaptation
    techniques to transfer knowledge between tasks and domains.
"""

import os
from typing import List, Dict, Any, Optional
from dotenv import load_dotenv
from datetime import datetime

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
from langchain_core.output_parsers import StrOutputParser

load_dotenv()


class DomainKnowledge:
    """Represents knowledge from a source domain"""
    
    def __init__(self, domain_name: str):
        self.domain_name = domain_name
        self.examples: List[Dict[str, str]] = []
        self.patterns: List[str] = []
        self.concepts: Dict[str, str] = {}
    
    def add_example(self, input_text: str, output_text: str, pattern: str = ""):
        """Add an example from this domain"""
        self.examples.append({
            "input": input_text,
            "output": output_text,
            "pattern": pattern
        })
    
    def add_pattern(self, pattern: str):
        """Add a learned pattern"""
        if pattern not in self.patterns:
            self.patterns.append(pattern)
    
    def add_concept(self, concept: str, description: str):
        """Add a domain concept"""
        self.concepts[concept] = description


class TransferLearningAgent:
    """Agent that transfers knowledge between domains"""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.source_domains: Dict[str, DomainKnowledge] = {}
        self.current_domain: Optional[str] = None
    
    def add_source_domain(self, domain: DomainKnowledge):
        """Add a source domain with learned knowledge"""
        self.source_domains[domain.domain_name] = domain
        print(f"Added source domain: {domain.domain_name}")
        print(f"  - {len(domain.examples)} examples")
        print(f"  - {len(domain.patterns)} patterns")
        print(f"  - {len(domain.concepts)} concepts")
    
    def identify_relevant_source(self, target_task: str) -> Optional[DomainKnowledge]:
        """Identify most relevant source domain for transfer"""
        if not self.source_domains:
            return None
        
        # Use LLM to identify most relevant source
        domains_desc = "\n".join([
            f"- {name}: {len(d.examples)} examples, patterns: {', '.join(d.patterns[:3])}"
            for name, d in self.source_domains.items()
        ])
        
        selection_prompt = f"""Given a target task and available source domains, identify which source domain would provide the most useful transferable knowledge.

Target task: {target_task}

Available source domains:
{domains_desc}

Which source domain is most relevant? Answer with just the domain name."""
        
        response = self.llm.invoke(selection_prompt).content.strip()
        
        # Find matching domain
        for domain_name in self.source_domains:
            if domain_name.lower() in response.lower():
                print(f"\nSelected source domain: {domain_name}")
                return self.source_domains[domain_name]
        
        # Default to first domain
        return list(self.source_domains.values())[0]
    
    def adapt_examples(self, source_domain: DomainKnowledge, 
                      target_domain: str, n: int = 3) -> List[Dict[str, str]]:
        """Adapt source examples to target domain"""
        if not source_domain.examples:
            return []
        
        # Take first n examples
        source_examples = source_domain.examples[:n]
        
        # Create adaptation prompt
        examples_str = "\n\n".join([
            f"Input: {ex['input']}\nOutput: {ex['output']}"
            for ex in source_examples
        ])
        
        adaptation_prompt = f"""Adapt these examples from {source_domain.domain_name} domain to {target_domain} domain.
Keep the same pattern and structure, but change the content to fit the target domain.

Source examples ({source_domain.domain_name}):
{examples_str}

Generate {n} adapted examples for {target_domain} domain in the same format."""
        
        response = self.llm.invoke(adaptation_prompt).content
        
        # Parse adapted examples (simplified parsing)
        adapted = []
        lines = response.split('\n')
        current_input = None
        
        for line in lines:
            if line.startswith('Input:'):
                current_input = line.replace('Input:', '').strip()
            elif line.startswith('Output:') and current_input:
                output = line.replace('Output:', '').strip()
                adapted.append({"input": current_input, "output": output})
                current_input = None
        
        return adapted[:n] if adapted else source_examples[:n]
    
    def transfer_and_respond(self, target_task: str, target_domain: str,
                            query: str) -> Dict[str, Any]:
        """Transfer knowledge and respond to query"""
        print(f"\n{'='*60}")
        print(f"Target Domain: {target_domain}")
        print(f"Target Task: {target_task}")
        print(f"Query: {query}")
        print(f"{'='*60}")
        
        # Identify relevant source domain
        source_domain = self.identify_relevant_source(target_task)
        
        if source_domain:
            print(f"\nTransferring knowledge from: {source_domain.domain_name}")
            
            # Adapt examples to target domain
            adapted_examples = self.adapt_examples(source_domain, target_domain, n=3)
            print(f"Adapted {len(adapted_examples)} examples to target domain")
            
            # Extract transferable patterns
            patterns_str = "\n".join([f"- {p}" for p in source_domain.patterns])
            
            # Create transfer-enhanced prompt
            if adapted_examples:
                example_prompt = ChatPromptTemplate.from_messages([
                    ("human", "{input}"),
                    ("ai", "{output}")
                ])
                
                few_shot_prompt = FewShotChatMessagePromptTemplate(
                    example_prompt=example_prompt,
                    examples=adapted_examples
                )
                
                prompt = ChatPromptTemplate.from_messages([
                    ("system", f"""You are an AI assistant for {target_domain}.

You are leveraging transferred knowledge from {source_domain.domain_name}.

Transferable patterns:
{patterns_str}

Apply these learned patterns to the current domain:"""),
                    few_shot_prompt,
                    ("human", "{input}")
                ])
            else:
                prompt = ChatPromptTemplate.from_messages([
                    ("system", f"""You are an AI assistant for {target_domain}.

You are leveraging transferred knowledge from {source_domain.domain_name}.

Transferable patterns:
{patterns_str}

Apply these learned patterns to the current task."""),
                    ("human", "{input}")
                ])
            
            chain = prompt | self.llm | StrOutputParser()
            response = chain.invoke({"input": query})
        else:
            # No source domain available, use base prompt
            print("\nNo source domain available, using base knowledge")
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are an AI assistant for {target_domain}."),
                ("human", "{input}")
            ])
            
            chain = prompt | self.llm | StrOutputParser()
            response = chain.invoke({"input": query})
        
        print(f"\nResponse:")
        print(response)
        
        return {
            "target_domain": target_domain,
            "target_task": target_task,
            "query": query,
            "source_domain": source_domain.domain_name if source_domain else None,
            "response": response,
            "timestamp": datetime.now().isoformat()
        }
    
    def compare_with_without_transfer(self, target_task: str, target_domain: str,
                                     query: str) -> Dict[str, Any]:
        """Compare performance with and without transfer learning"""
        print(f"\n{'='*70}")
        print(f"COMPARISON: With vs Without Transfer Learning")
        print(f"{'='*70}")
        
        # Response with transfer
        print("\n--- WITH TRANSFER LEARNING ---")
        with_transfer = self.transfer_and_respond(target_task, target_domain, query)
        
        # Response without transfer (base model)
        print("\n--- WITHOUT TRANSFER LEARNING (Baseline) ---")
        base_prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are an AI assistant for {target_domain}."),
            ("human", "{input}")
        ])
        base_chain = base_prompt | self.llm | StrOutputParser()
        without_transfer = base_chain.invoke({"input": query})
        
        print(f"\nBaseline Response:")
        print(without_transfer)
        
        return {
            "with_transfer": with_transfer,
            "without_transfer": without_transfer
        }


def demonstrate_transfer_learning():
    """Demonstrate transfer learning pattern"""
    print("\n" + "="*70)
    print("TRANSFER LEARNING AGENT DEMONSTRATION")
    print("="*70)
    
    agent = TransferLearningAgent()
    
    # Create source domain: Customer Support
    print("\n" + "="*70)
    print("Setting Up Source Domain Knowledge")
    print("="*70)
    
    customer_support = DomainKnowledge("customer_support")
    
    # Add examples
    customer_support.add_example(
        "How do I reset my password?",
        "To reset your password: 1) Click 'Forgot Password' on login page, "
        "2) Enter your email, 3) Check your email for reset link, 4) Create new password.",
        "step_by_step_instructions"
    )
    customer_support.add_example(
        "Where is my order?",
        "To track your order: 1) Log into your account, 2) Go to 'Orders', "
        "3) Find your order number, 4) Click 'Track' to see current status.",
        "step_by_step_instructions"
    )
    customer_support.add_example(
        "How do I cancel my subscription?",
        "To cancel subscription: 1) Go to Account Settings, 2) Select 'Subscriptions', "
        "3) Choose subscription to cancel, 4) Confirm cancellation.",
        "step_by_step_instructions"
    )
    
    # Add patterns
    customer_support.add_pattern("Provide step-by-step instructions")
    customer_support.add_pattern("Be clear and concise")
    customer_support.add_pattern("Offer alternative solutions")
    customer_support.add_pattern("Include troubleshooting tips")
    
    # Add concepts
    customer_support.add_concept("account_management", "User account operations")
    customer_support.add_concept("order_processing", "Order lifecycle and tracking")
    
    agent.add_source_domain(customer_support)
    
    # Example 1: Transfer to Technical Support
    print("\n" + "="*70)
    print("Example 1: Transfer to Technical Support Domain")
    print("="*70)
    
    result1 = agent.transfer_and_respond(
        target_task="Help users troubleshoot technical issues",
        target_domain="technical_support",
        query="How do I fix a blue screen error?"
    )
    
    # Example 2: Transfer to Medical Assistant
    print("\n" + "="*70)
    print("Example 2: Transfer to Medical Assistant Domain")
    print("="*70)
    
    result2 = agent.transfer_and_respond(
        target_task="Provide medical information and guidance",
        target_domain="medical_assistant",
        query="How do I check my blood pressure at home?"
    )
    
    # Example 3: Compare with and without transfer
    print("\n" + "="*70)
    print("Example 3: Comparing With/Without Transfer")
    print("="*70)
    
    comparison = agent.compare_with_without_transfer(
        target_task="Help users with cooking recipes",
        target_domain="cooking_assistant",
        query="How do I make homemade pasta?"
    )
    
    # Summary
    print("\n" + "="*70)
    print("SUMMARY")
    print("="*70)
    print("""
Transfer Learning Pattern demonstrated:

Key Features:
1. Source Domain Selection - Identify most relevant source knowledge
2. Knowledge Adaptation - Adapt examples and patterns to new domain
3. Pattern Transfer - Transfer learned patterns across domains
4. Few-Shot Enhancement - Use adapted examples for better performance
5. Comparative Evaluation - Compare with/without transfer

Transfer Mechanisms:
- Example adaptation across domains
- Pattern and structure transfer
- Concept mapping between domains
- Meta-learning of general strategies
- Domain-specific fine-tuning

Applications:
- Multi-domain chatbots (customer → technical support)
- Cross-lingual NLP (English → Spanish)
- Domain adaptation (medical → legal documentation)
- Few-shot learning in new domains
- Skill transfer in robotics

Benefits:
- Faster adaptation to new domains
- Better performance with limited data
- Leverages existing knowledge
- Reduces training time and data needs
- Improves generalization

The transfer learning pattern enables agents to leverage knowledge from
related domains, significantly improving performance on new tasks with
limited domain-specific training.
    """)


if __name__ == "__main__":
    demonstrate_transfer_learning()

    

Agentic Design Pattern: Transfer Learning

Pattern 115: Transfer Learning – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Transfer Learning pattern implementation.

---

Pattern Description

Pattern 115: Transfer Learning Agent

Description: Transfer Learning enables agents to leverage knowledge learned from one task or domain to improve performance on related tasks. This pattern identifies transferable knowledge, adapts it to new contexts, and combines it with task-specific learning for faster adaptation and better performance.

Essential for agents that need to work across multiple related domains or
tasks, transfer learning reduces the data and time needed to achieve good
performance in new areas.

Key Components: 1. Source Knowledge Extractor: Identifies transferable knowledge 2. Domain Adapter: Adapts knowledge to new domain 3. Fine-tuner: Specializes for target task 4. Transfer Strategy Selector: Chooses transfer approach 5. Performance Evaluator: Assesses transfer effectiveness

Transfer Strategies: – Feature Transfer: Use learned representations – Parameter Transfer: Transfer model parameters – Instance Transfer: Leverage source examples – Relational Transfer: Transfer relationships – Meta-transfer: Transfer learning strategies

Use Cases: – Multi-domain chatbots – Cross-lingual applications – Domain adaptation (e.g., medical to legal) – Few-shot learning in new domains – Skill transfer in robotics

LangChain Implementation: Uses prompt engineering, few-shot examples, and domain adaptation techniques to transfer knowledge between tasks and domains.

---

File Statistics

• Total Lines: 403
• Classes: 2
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 115: Transfer Learning Agent

Description:
    Transfer Learning enables agents to leverage knowledge learned from one
    task or domain to improve performance on related tasks. This pattern
    identifies transferable knowledge, adapts it to new contexts, and combines
    it with task-specific learning for faster adaptation and better performance.

    Essential for agents that need to work across multiple related domains or
    tasks, transfer learning reduces the data and time needed to achieve good
    performance in new areas.

Key Components:
    1. Source Knowledge Extractor: Identifies transferable knowledge
    2. Domain Adapter: Adapts knowledge to new domain
    3. Fine-tuner: Specializes for target task
    4. Transfer Strategy Selector: Chooses transfer approach
    5. Performance Evaluator: Assesses transfer effectiveness

Transfer Strategies:
    - Feature Transfer: Use learned representations
    - Parameter Transfer: Transfer model parameters
    - Instance Transfer: Leverage source examples
    - Relational Transfer: Transfer relationships
    - Meta-transfer: Transfer learning strategies

Use Cases:
    - Multi-domain chatbots
    - Cross-lingual applications
    - Domain adaptation (e.g., medical to legal)
    - Few-shot learning in new domains
    - Skill transfer in robotics

LangChain Implementation:
    Uses prompt engineering, few-shot examples, and domain adaptation
    techniques to transfer knowledge between tasks and domains.
"""

Purpose: This module implements the Transfer Learning pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import datetime.datetime
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import langchain_core.prompts.FewShotChatMessagePromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – datetime.datetime: Date and time handling – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.prompts.FewShotChatMessagePromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: 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)

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: DomainKnowledge

Lines 52-76

Purpose:

Represents knowledge from a source domain

Methods:

__init__(self, domain_name)

Lines 55-59

Purpose: Initialize the class instance with required parameters.

Parameters: – domain_name: Domain name

add_example(self, input_text, output_text, pattern)

Lines 61-67

Purpose:

Add an example from this domain

Parameters: – input_text: Input data – output_text: Output text – pattern: Pattern

add_pattern(self, pattern)

Lines 69-72

Purpose:

Add a learned pattern

Parameters: – pattern: Pattern

add_concept(self, concept, description)

Lines 74-76

Purpose:

Add a domain concept

Parameters: – concept: Concept – description: Description

---

Class: TransferLearningAgent

Lines 79-277

Purpose:

Agent that transfers knowledge between domains

Methods:

__init__(self, model_name)

Lines 82-85

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

add_source_domain(self, domain)

Lines 87-93

Purpose:

Add a source domain with learned knowledge

Parameters: – domain: Domain

identify_relevant_source(self, target_task)

Lines 95-124

Purpose:

Identify most relevant source domain for transfer

Parameters: – target_task: Target task

Returns: <ast.Subscript object at 0x000001A06D7FEF50>

adapt_examples(self, source_domain, target_domain, n)

Lines 126-164

Purpose:

Adapt source examples to target domain

Parameters: – source_domain: Source domain – target_domain: Target domain – n: N

Returns: <ast.Subscript object at 0x000001A06D72EDD0>

transfer_and_respond(self, target_task, target_domain, query)

Lines 166-249

Purpose:

Transfer knowledge and respond to query

Parameters: – target_task: Target task – target_domain: Target domain – query: Input query or question

Returns: <ast.Subscript object at 0x000001A06D79B310>

compare_with_without_transfer(self, target_task, target_domain, query)

Lines 251-277

Purpose:

Compare performance with and without transfer learning

Parameters: – target_task: Target task – target_domain: Target domain – query: Input query or question

Returns: <ast.Subscript object at 0x000001A06D7E43D0>

---

Functions

demonstrate_transfer_learning()

Lines 280-398

Purpose:

Demonstrate transfer learning pattern

---

Key Concepts Summary

Pattern Overview

The Transfer Learning 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

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. 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 .env files 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

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. 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

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 114
• Pattern 116

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 Transfer Learning 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: 115_transfer_learning.py
Total Lines: 403
Pattern Category: AI Agent Design Pattern #115

---

Generated from actual implementation code
For educational and reference purposes