Tag: llm

Agentic Design Pattern: Multi Turn Dialogue


    """
Pattern 128: Multi-Turn Dialogue Management

Description:
    Manages coherent multi-turn conversations with context tracking, dialogue state,
    and appropriate responses across multiple exchanges.

Components:
    - Dialogue state tracking
    - Context management
    - Intent recognition
    - Response generation

Use Cases:
    - Conversational AI assistants
    - Customer service chatbots
    - Interactive tutoring systems

LangChain Implementation:
    Uses LangChain memory, state management, and conversation chains.
"""

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

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.runnables import RunnablePassthrough
from langchain_core.pydantic_v1 import BaseModel, Field

load_dotenv()


class DialogueState(str, Enum):
    """Dialogue states"""
    GREETING = "greeting"
    INFO_GATHERING = "info_gathering"
    CLARIFICATION = "clarification"
    PROCESSING = "processing"
    RESPONSE = "response"
    CLOSING = "closing"


class UserIntent(BaseModel):
    """Detected user intent"""
    intent: str = Field(description="Primary intent")
    confidence: float = Field(description="Confidence score")
    entities: Dict[str, Any] = Field(description="Extracted entities")


class DialogueContext(BaseModel):
    """Dialogue context"""
    state: str = Field(description="Current dialogue state")
    topic: Optional[str] = Field(description="Current topic")
    user_goal: Optional[str] = Field(description="User's goal")
    collected_info: Dict[str, Any] = Field(default_factory=dict)


class MultiTurnDialogueAgent:
    """Agent for managing multi-turn conversations"""
    
    def __init__(self, model: str = "gpt-4o-mini", temperature: float = 0.7):
        self.llm = ChatOpenAI(model=model, temperature=temperature)
        self.conversation_history: List[Dict[str, str]] = []
        self.context: DialogueContext = DialogueContext(
            state=DialogueState.GREETING.value,
            topic=None,
            user_goal=None,
            collected_info={}
        )
        
        # Intent detection prompt
        self.intent_prompt = ChatPromptTemplate.from_messages([
            ("system", """Analyze the user's message and detect their intent.
Common intents: greeting, question, request_info, provide_info, complaint, thanks, goodbye

Return JSON:
{{
  "intent": "intent_name",
  "confidence": 0.9,
  "entities": {{"key": "value"}}
}}"""),
            ("user", "{message}")
        ])
        
        # Response generation with context
        self.response_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a helpful conversational AI assistant.
Current conversation context:
- State: {state}
- Topic: {topic}
- User Goal: {user_goal}
- Collected Info: {collected_info}

Provide a natural, contextually appropriate response that:
1. Acknowledges the user's message
2. Maintains conversation flow
3. Helps achieve the user's goal
4. Asks follow-up questions if needed"""),
            MessagesPlaceholder(variable_name="history"),
            ("user", "{message}")
        ])
        
        self.intent_parser = JsonOutputParser()
        self.response_parser = StrOutputParser()
    
    def detect_intent(self, message: str) -> UserIntent:
        """Detect user intent from message"""
        try:
            chain = self.intent_prompt | self.llm | self.intent_parser
            result = chain.invoke({"message": message})
            
            return UserIntent(
                intent=result["intent"],
                confidence=result["confidence"],
                entities=result.get("entities", {})
            )
        except Exception as e:
            print(f"Intent detection error: {e}")
            return UserIntent(intent="unknown", confidence=0.0, entities={})
    
    def update_context(self, intent: UserIntent, message: str):
        """Update dialogue context based on intent"""
        # Update state based on intent
        if intent.intent == "greeting":
            self.context.state = DialogueState.GREETING.value
        elif intent.intent in ["question", "request_info"]:
            self.context.state = DialogueState.INFO_GATHERING.value
        elif intent.intent == "provide_info":
            self.context.state = DialogueState.PROCESSING.value
            # Store provided information
            self.context.collected_info.update(intent.entities)
        elif intent.intent == "goodbye":
            self.context.state = DialogueState.CLOSING.value
        
        # Extract topic if not set
        if not self.context.topic and len(self.conversation_history) > 0:
            # Simple topic extraction (could be enhanced)
            for entity_key in intent.entities:
                if entity_key not in ["time", "date", "number"]:
                    self.context.topic = intent.entities[entity_key]
                    break
    
    def generate_response(self, message: str) -> str:
        """Generate contextually appropriate response"""
        # Detect intent
        intent = self.detect_intent(message)
        print(f"\n🎯 Detected Intent: {intent.intent} (confidence: {intent.confidence:.2f})")
        
        # Update context
        self.update_context(intent, message)
        print(f"📊 Dialogue State: {self.context.state}")
        if self.context.topic:
            print(f"💡 Topic: {self.context.topic}")
        
        # Prepare conversation history for prompt
        history = []
        for turn in self.conversation_history[-5:]:  # Last 5 turns
            if turn["role"] == "user":
                history.append(HumanMessage(content=turn["content"]))
            else:
                history.append(AIMessage(content=turn["content"]))
        
        # Generate response
        chain = self.response_prompt | self.llm | self.response_parser
        
        response = chain.invoke({
            "message": message,
            "state": self.context.state,
            "topic": self.context.topic or "not set",
            "user_goal": self.context.user_goal or "not determined",
            "collected_info": str(self.context.collected_info),
            "history": history
        })
        
        return response
    
    def chat(self, user_message: str) -> str:
        """Process user message and return response"""
        print(f"\n{'='*60}")
        print(f"👤 User: {user_message}")
        
        # Add to history
        self.conversation_history.append({
            "role": "user",
            "content": user_message,
            "timestamp": datetime.now().isoformat()
        })
        
        # Generate response
        response = self.generate_response(user_message)
        
        # Add response to history
        self.conversation_history.append({
            "role": "assistant",
            "content": response,
            "timestamp": datetime.now().isoformat()
        })
        
        print(f"🤖 Assistant: {response}")
        
        return response
    
    def get_conversation_summary(self) -> str:
        """Generate summary of conversation"""
        if not self.conversation_history:
            return "No conversation yet."
        
        summary_prompt = ChatPromptTemplate.from_messages([
            ("system", "Summarize the following conversation in 2-3 sentences."),
            ("user", "{conversation}")
        ])
        
        conversation_text = "\n".join([
            f"{turn['role']}: {turn['content']}"
            for turn in self.conversation_history
        ])
        
        chain = summary_prompt | self.llm | StrOutputParser()
        summary = chain.invoke({"conversation": conversation_text})
        
        return summary
    
    def reset_conversation(self):
        """Reset conversation state"""
        self.conversation_history = []
        self.context = DialogueContext(
            state=DialogueState.GREETING.value,
            topic=None,
            user_goal=None,
            collected_info={}
        )
        print("\n🔄 Conversation reset")


def demonstrate_multi_turn_dialogue():
    """Demonstrate multi-turn dialogue management"""
    print("=" * 80)
    print("Pattern 128: Multi-Turn Dialogue Management")
    print("=" * 80)
    
    agent = MultiTurnDialogueAgent()
    
    # Example 1: Customer service conversation
    print("\n" + "=" * 80)
    print("EXAMPLE 1: Customer Service Conversation")
    print("=" * 80)
    
    conversation_1 = [
        "Hi, I need help with my order",
        "My order number is 12345",
        "It was supposed to arrive yesterday but I haven't received it",
        "Yes, my address is 123 Main Street",
        "Thank you for your help!"
    ]
    
    for message in conversation_1:
        agent.chat(message)
    
    print("\n" + "-" * 60)
    print("📝 Conversation Summary:")
    print(agent.get_conversation_summary())
    
    # Example 2: Technical support
    print("\n" + "=" * 80)
    print("EXAMPLE 2: Technical Support Conversation")
    print("=" * 80)
    
    agent.reset_conversation()
    
    conversation_2 = [
        "Hello, my computer won't start",
        "It's a laptop, about 2 years old",
        "When I press the power button, nothing happens at all",
        "No, there are no lights or sounds",
        "Okay, I'll try that. Thanks!"
    ]
    
    for message in conversation_2:
        agent.chat(message)
    
    print("\n" + "-" * 60)
    print("📝 Conversation Summary:")
    print(agent.get_conversation_summary())
    
    # Example 3: Information gathering
    print("\n" + "=" * 80)
    print("EXAMPLE 3: Information Gathering Conversation")
    print("=" * 80)
    
    agent.reset_conversation()
    
    conversation_3 = [
        "I want to book a flight",
        "From New York to London",
        "Next week, preferably Tuesday",
        "Economy class is fine",
        "Great, please proceed with the booking"
    ]
    
    for message in conversation_3:
        agent.chat(message)
    
    print("\n" + "-" * 60)
    print("📝 Conversation Summary:")
    print(agent.get_conversation_summary())
    
    print("\n" + "-" * 60)
    print("📊 Final Context State:")
    print(f"  State: {agent.context.state}")
    print(f"  Topic: {agent.context.topic}")
    print(f"  Collected Info: {agent.context.collected_info}")
    print(f"  Total Turns: {len(agent.conversation_history)}")
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY")
    print("=" * 80)
    print("""
Multi-Turn Dialogue Management Pattern:
- Maintains context across conversation turns
- Tracks dialogue state and user intent
- Generates contextually appropriate responses
- Manages information collection

Key Features:
✓ Intent detection
✓ State tracking
✓ Context maintenance
✓ Natural conversation flow
✓ Information extraction

Dialogue States:
• Greeting - Initial interaction
• Info Gathering - Collecting information
• Clarification - Resolving ambiguities
• Processing - Handling requests
• Response - Providing answers
• Closing - Ending conversation

Benefits:
✓ Coherent multi-turn conversations
✓ Context-aware responses
✓ Better user experience
✓ Goal-oriented dialogue
✓ Efficient information gathering
    """)


if __name__ == "__main__":
    demonstrate_multi_turn_dialogue()

Agentic Design Pattern: Multi Turn Dialogue

Pattern 128: Multi Turn Dialogue – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Multi Turn Dialogue pattern implementation.

Pattern Description

Pattern 128: Multi-Turn Dialogue Management

Description: Manages coherent multi-turn conversations with context tracking, dialogue state, and appropriate responses across multiple exchanges.

Components: – Dialogue state tracking – Context management – Intent recognition – Response generation

Use Cases: – Conversational AI assistants – Customer service chatbots – Interactive tutoring systems

LangChain Implementation: Uses LangChain memory, state management, and conversation chains.

File Statistics

Total Lines: 357
Classes: 4
Functions: 1
Constants: 0

Module Documentation

"""
Pattern 128: Multi-Turn Dialogue Management

Description:
    Manages coherent multi-turn conversations with context tracking, dialogue state,
    and appropriate responses across multiple exchanges.

Components:
    - Dialogue state tracking
    - Context management
    - Intent recognition
    - Response generation

Use Cases:
    - Conversational AI assistants
    - Customer service chatbots
    - Interactive tutoring systems

LangChain Implementation:
    Uses LangChain memory, state management, and conversation chains.
"""

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

Import Statements

Standard Library Imports

import datetime.datetime
import enum.Enum
import langchain_core.messages.AIMessage
import langchain_core.messages.HumanMessage
import langchain_core.messages.SystemMessage
import langchain_core.output_parsers.JsonOutputParser
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import langchain_core.prompts.MessagesPlaceholder
import langchain_core.pydantic_v1.BaseModel
import langchain_core.pydantic_v1.Field
import langchain_core.runnables.RunnablePassthrough
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.messages.AIMessage: Regular expression operations – langchain_core.messages.HumanMessage: Regular expression operations – langchain_core.messages.SystemMessage: Regular expression operations – langchain_core.output_parsers.JsonOutputParser: Regular expression operations – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.prompts.MessagesPlaceholder: Regular expression operations – langchain_core.pydantic_v1.BaseModel: Regular expression operations – langchain_core.pydantic_v1.Field: Regular expression operations – langchain_core.runnables.RunnablePassthrough: 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: `DialogueState`

Lines 39-46

Inherits from: str, Enum

Purpose:

Dialogue states

Class: `UserIntent`

Lines 49-53

Inherits from: BaseModel

Purpose:

Detected user intent

Class: `DialogueContext`

Lines 56-61

Inherits from: BaseModel

Purpose:

Dialogue context

Class: `MultiTurnDialogueAgent`

Lines 64-238

Purpose:

Agent for managing multi-turn conversations

Methods:

`init(self, model, temperature)`

Lines 67-110

Purpose: Initialize the class instance with required parameters.

Parameters: – model: Language model instance – temperature: Sampling temperature (0-1)

`detect_intent(self, message)`

Lines 112-125

Purpose:

Detect user intent from message

Parameters: – message: Message

Returns: UserIntent

`update_context(self, intent, message)`

Lines 127-147

Purpose:

Update dialogue context based on intent

Parameters: – intent: Intent – message: Message

`generate_response(self, message)`

Lines 149-181

Purpose:

Generate contextually appropriate response

Parameters: – message: Message

Returns: str

`chat(self, user_message)`

Lines 183-207

Purpose:

Process user message and return response

Parameters: – user_message: User message

Returns: str

`get_conversation_summary(self)`

Lines 209-227

Purpose:

Generate summary of conversation

Parameters:

Returns: str

`reset_conversation(self)`

Lines 229-238

Purpose:

Reset conversation state

Parameters:

Functions

`demonstrate_multi_turn_dialogue()`

Lines 241-352

Purpose:

Demonstrate multi-turn dialogue management

Key Concepts Summary

Pattern Overview

The Multi Turn Dialogue pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

Agent Class: Main implementation of the pattern logic
Prompt Engineering: Carefully crafted prompts to guide LLM behavior
Tool Integration: Optional tools for extended functionality
Memory Management: State handling across interactions
Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

Best Practices Demonstrated

1. Environment Configuration

✅ Use .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

Prompt Optimization: Minimize tokens while maintaining clarity
Caching: Cache repeated queries/responses
Model Selection: Use smaller models when appropriate
Batch Processing: Group similar requests
Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

GPT-3.5-turbo: ~$0.002 per request
GPT-4: ~$0.03 per request
GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 127
Pattern 129

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 Turn Dialogue 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: 128_multi_turn_dialogue.py
Total Lines: 357
Pattern Category: AI Agent Design Pattern #128

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Agentic Design Pattern: Semantic Search Retrieval


    """
Pattern 127: Semantic Search & Retrieval

Description:
    Retrieves information based on semantic similarity rather than keyword matching,
    using embeddings and vector databases for intelligent information retrieval.

Components:
    - Text embeddings
    - Vector similarity search
    - Semantic ranking
    - Hybrid retrieval (keyword + semantic)

Use Cases:
    - Intelligent document search
    - Question answering systems
    - Recommendation systems

LangChain Implementation:
    Uses LangChain embeddings, vector stores, and retrievers for semantic search.
"""

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

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.documents import Document
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.runnables import RunnablePassthrough
from langchain.retrievers import EnsembleRetriever, ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_community.retrievers import BM25Retriever

load_dotenv()


class SemanticSearchAgent:
    """Agent for semantic search and retrieval"""
    
    def __init__(self, model: str = "gpt-4o-mini", temperature: float = 0.0):
        self.llm = ChatOpenAI(model=model, temperature=temperature)
        self.embeddings = OpenAIEmbeddings()
        self.vectorstore: Optional[Chroma] = None
        self.documents: List[Document] = []
        
        # Text splitter for chunking
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50,
            length_function=len
        )
        
        # QA prompt
        self.qa_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a helpful assistant that answers questions based on the provided context.
Use only the information from the context to answer. If you cannot answer from the context, say so.
Always cite which part of the context you used."""),
            ("user", """Context:
{context}

Question: {question}

Answer:""")
        ])
    
    def load_documents(self, documents: List[str], metadatas: Optional[List[Dict]] = None):
        """Load documents into the system"""
        print(f"\n📚 Loading {len(documents)} documents...")
        
        # Create Document objects
        if metadatas:
            self.documents = [
                Document(page_content=doc, metadata=meta)
                for doc, meta in zip(documents, metadatas)
            ]
        else:
            self.documents = [
                Document(page_content=doc, metadata={"id": i})
                for i, doc in enumerate(documents)
            ]
        
        # Split documents
        split_docs = self.text_splitter.split_documents(self.documents)
        print(f"   Split into {len(split_docs)} chunks")
        
        # Create vector store
        self.vectorstore = Chroma.from_documents(
            documents=split_docs,
            embedding=self.embeddings,
            collection_name="semantic_search"
        )
        
        print(f"   ✓ Vectorstore created with {len(split_docs)} embeddings")
    
    def semantic_search(self, query: str, k: int = 3) -> List[Document]:
        """Pure semantic search using vector similarity"""
        print(f"\n🔍 Semantic search for: '{query}'")
        
        if not self.vectorstore:
            print("   ❌ No documents loaded!")
            return []
        
        # Retrieve similar documents
        results = self.vectorstore.similarity_search_with_score(query, k=k)
        
        print(f"   Found {len(results)} results:\n")
        for i, (doc, score) in enumerate(results, 1):
            print(f"   {i}. Score: {score:.4f}")
            print(f"      {doc.page_content[:100]}...\n")
        
        return [doc for doc, _ in results]
    
    def hybrid_search(self, query: str, k: int = 3) -> List[Document]:
        """Hybrid search combining semantic and keyword matching"""
        print(f"\n🔍 Hybrid search for: '{query}'")
        
        if not self.vectorstore or not self.documents:
            print("   ❌ No documents loaded!")
            return []
        
        # Create semantic retriever
        semantic_retriever = self.vectorstore.as_retriever(search_kwargs={"k": k})
        
        # Create keyword retriever (BM25)
        bm25_retriever = BM25Retriever.from_documents(self.documents)
        bm25_retriever.k = k
        
        # Combine retrievers
        ensemble_retriever = EnsembleRetriever(
            retrievers=[semantic_retriever, bm25_retriever],
            weights=[0.7, 0.3]  # Favor semantic
        )
        
        results = ensemble_retriever.get_relevant_documents(query)
        
        print(f"   Found {len(results)} results (semantic + keyword)\n")
        for i, doc in enumerate(results[:k], 1):
            print(f"   {i}. {doc.page_content[:100]}...\n")
        
        return results[:k]
    
    def search_with_reranking(self, query: str, k: int = 3) -> List[Document]:
        """Search with LLM-based reranking/compression"""
        print(f"\n🔍 Search with reranking for: '{query}'")
        
        if not self.vectorstore:
            print("   ❌ No documents loaded!")
            return []
        
        # Base retriever
        base_retriever = self.vectorstore.as_retriever(search_kwargs={"k": k * 2})
        
        # Compressor for reranking
        compressor = LLMChainExtractor.from_llm(self.llm)
        
        # Compression retriever
        compression_retriever = ContextualCompressionRetriever(
            base_compressor=compressor,
            base_retriever=base_retriever
        )
        
        results = compression_retriever.get_relevant_documents(query)
        
        print(f"   Found and reranked {len(results)} results\n")
        for i, doc in enumerate(results[:k], 1):
            print(f"   {i}. {doc.page_content[:150]}...\n")
        
        return results[:k]
    
    def answer_question(self, question: str, search_method: str = "semantic") -> str:
        """Answer question using retrieved context"""
        print(f"\n💬 Answering question: '{question}'")
        print(f"   Using {search_method} search")
        
        # Retrieve context
        if search_method == "semantic":
            context_docs = self.semantic_search(question, k=3)
        elif search_method == "hybrid":
            context_docs = self.hybrid_search(question, k=3)
        else:  # reranking
            context_docs = self.search_with_reranking(question, k=3)
        
        if not context_docs:
            return "No relevant context found."
        
        # Combine context
        context = "\n\n".join([doc.page_content for doc in context_docs])
        
        # Generate answer
        chain = self.qa_prompt | self.llm | StrOutputParser()
        answer = chain.invoke({"context": context, "question": question})
        
        print(f"\n📝 Answer:\n{answer}")
        
        return answer
    
    def find_similar_documents(self, reference_text: str, k: int = 3) -> List[Document]:
        """Find documents similar to reference text"""
        print(f"\n🔍 Finding similar documents to reference text...")
        
        if not self.vectorstore:
            print("   ❌ No documents loaded!")
            return []
        
        results = self.vectorstore.similarity_search(reference_text, k=k)
        
        print(f"   Found {len(results)} similar documents:\n")
        for i, doc in enumerate(results, 1):
            print(f"   {i}. {doc.page_content[:100]}...")
            if 'id' in doc.metadata:
                print(f"      (Document ID: {doc.metadata['id']})\n")
        
        return results
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get search system statistics"""
        return {
            "total_documents": len(self.documents),
            "vectorstore_loaded": self.vectorstore is not None,
            "embedding_model": "text-embedding-ada-002"
        }


def demonstrate_semantic_search():
    """Demonstrate semantic search and retrieval"""
    print("=" * 80)
    print("Pattern 127: Semantic Search & Retrieval")
    print("=" * 80)
    
    agent = SemanticSearchAgent()
    
    # Example 1: Load documents
    print("\n" + "=" * 80)
    print("EXAMPLE 1: Loading Documents")
    print("=" * 80)
    
    documents = [
        """Python is a high-level, interpreted programming language known for its simplicity 
        and readability. Created by Guido van Rossum in 1991, Python emphasizes code 
        readability and allows programmers to express concepts in fewer lines of code.""",
        
        """Machine learning is a subset of artificial intelligence that enables systems to 
        learn and improve from experience without being explicitly programmed. It focuses 
        on developing computer programs that can access data and learn from it.""",
        
        """Deep learning is a subset of machine learning that uses neural networks with 
        multiple layers. These deep neural networks attempt to simulate the behavior of 
        the human brain, allowing it to learn from large amounts of data.""",
        
        """Natural Language Processing (NLP) is a branch of AI that helps computers understand, 
        interpret, and manipulate human language. NLP draws from many disciplines, including 
        computer science and computational linguistics.""",
        
        """Computer vision is a field of artificial intelligence that trains computers to 
        interpret and understand the visual world. Using digital images and deep learning 
        models, machines can accurately identify and classify objects.""",
        
        """The Transformer architecture, introduced in the paper 'Attention is All You Need', 
        revolutionized NLP. It relies entirely on attention mechanisms, dispensing with 
        recurrence and convolutions entirely."""
    ]
    
    metadatas = [
        {"topic": "programming", "language": "python"},
        {"topic": "AI", "subtopic": "machine_learning"},
        {"topic": "AI", "subtopic": "deep_learning"},
        {"topic": "AI", "subtopic": "NLP"},
        {"topic": "AI", "subtopic": "computer_vision"},
        {"topic": "AI", "subtopic": "transformers"}
    ]
    
    agent.load_documents(documents, metadatas)
    
    # Example 2: Semantic search
    print("\n" + "=" * 80)
    print("EXAMPLE 2: Pure Semantic Search")
    print("=" * 80)
    
    queries = [
        "neural networks and learning from data",
        "understanding human language with computers",
        "programming language for beginners"
    ]
    
    for query in queries:
        agent.semantic_search(query, k=2)
    
    # Example 3: Hybrid search
    print("\n" + "=" * 80)
    print("EXAMPLE 3: Hybrid Search (Semantic + Keyword)")
    print("=" * 80)
    
    agent.hybrid_search("deep neural networks", k=2)
    
    # Example 4: Search with reranking
    print("\n" + "=" * 80)
    print("EXAMPLE 4: Search with LLM Reranking")
    print("=" * 80)
    
    agent.search_with_reranking("AI techniques for processing text", k=2)
    
    # Example 5: Question answering with different search methods
    print("\n" + "=" * 80)
    print("EXAMPLE 5: Question Answering")
    print("=" * 80)
    
    questions = [
        "What is the Transformer architecture?",
        "Who created Python and when?",
        "What is the relationship between deep learning and machine learning?"
    ]
    
    methods = ["semantic", "hybrid", "reranking"]
    
    for question in questions[:1]:  # Just first question with all methods
        print(f"\nQuestion: {question}\n")
        for method in methods:
            print(f"\n--- Method: {method} ---")
            agent.answer_question(question, search_method=method)
    
    # Example 6: Find similar documents
    print("\n" + "=" * 80)
    print("EXAMPLE 6: Finding Similar Documents")
    print("=" * 80)
    
    reference = "Neural networks that can process and generate human language"
    agent.find_similar_documents(reference, k=3)
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY")
    print("=" * 80)
    print("""
Semantic Search & Retrieval Pattern:
- Finds information based on meaning, not just keywords
- Uses embeddings to capture semantic similarity
- Multiple search strategies available
- Enables intelligent question answering

Search Methods:
1. Pure Semantic: Vector similarity only
2. Hybrid: Combines semantic + keyword (BM25)
3. With Reranking: LLM compresses and reranks results

Key Benefits:
✓ Understanding user intent
✓ Finding conceptually similar content
✓ Better than keyword matching
✓ Handles synonyms and paraphrases
✓ Improved retrieval accuracy

Use Cases:
• Document search systems
• Question answering
• Recommendation engines
• Knowledge bases
• Research assistants
    """)


if __name__ == "__main__":
    demonstrate_semantic_search()

Agentic Design Pattern: Semantic Search Retrieval

Pattern 127: Semantic Search Retrieval – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Semantic Search Retrieval pattern implementation.

Pattern Description

Pattern 127: Semantic Search & Retrieval

Description: Retrieves information based on semantic similarity rather than keyword matching, using embeddings and vector databases for intelligent information retrieval.

Components: – Text embeddings – Vector similarity search – Semantic ranking – Hybrid retrieval (keyword + semantic)

Use Cases: – Intelligent document search – Question answering systems – Recommendation systems

LangChain Implementation: Uses LangChain embeddings, vector stores, and retrievers for semantic search.

File Statistics

Total Lines: 367
Classes: 1
Functions: 1
Constants: 0

Module Documentation

"""
Pattern 127: Semantic Search & Retrieval

Description:
    Retrieves information based on semantic similarity rather than keyword matching,
    using embeddings and vector databases for intelligent information retrieval.

Components:
    - Text embeddings
    - Vector similarity search
    - Semantic ranking
    - Hybrid retrieval (keyword + semantic)

Use Cases:
    - Intelligent document search
    - Question answering systems
    - Recommendation systems

LangChain Implementation:
    Uses LangChain embeddings, vector stores, and retrievers for semantic search.
"""

Purpose: This module implements the Semantic Search Retrieval pattern, providing a structured approach to agent design and implementation.

Import Statements

Standard Library Imports

import langchain.retrievers.ContextualCompressionRetriever
import langchain.retrievers.EnsembleRetriever
import langchain.retrievers.document_compressors.LLMChainExtractor
import langchain_community.retrievers.BM25Retriever
import langchain_community.vectorstores.Chroma
import langchain_core.documents.Document
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import langchain_core.runnables.RunnablePassthrough
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – langchain.retrievers.ContextualCompressionRetriever: Regular expression operations – langchain.retrievers.EnsembleRetriever: Regular expression operations – langchain.retrievers.document_compressors.LLMChainExtractor: Regular expression operations – langchain_community.retrievers.BM25Retriever: Regular expression operations – langchain_community.vectorstores.Chroma: Regular expression operations – langchain_core.documents.Document: Regular expression operations – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.runnables.RunnablePassthrough: 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.text_splitter.RecursiveCharacterTextSplitter
from langchain_openai.ChatOpenAI
from langchain_openai.OpenAIEmbeddings

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: `SemanticSearchAgent`

Lines 41-225

Purpose:

Agent for semantic search and retrieval

Methods:

`init(self, model, temperature)`

Lines 44-68

Purpose: Initialize the class instance with required parameters.

Parameters: – model: Language model instance – temperature: Sampling temperature (0-1)

`load_documents(self, documents, metadatas)`

Lines 70-97

Purpose:

Load documents into the system

Parameters: – documents: Documents – metadatas: Metadatas

`semantic_search(self, query, k)`

Lines 99-115

Purpose:

Pure semantic search using vector similarity

Parameters: – query: Input query or question – k: K

Returns: <ast.Subscript object at 0x000001A06D62A390>

`hybrid_search(self, query, k)`

Lines 117-144

Purpose:

Hybrid search combining semantic and keyword matching

Parameters: – query: Input query or question – k: K

Returns: <ast.Subscript object at 0x000001A06D77C1D0>

`search_with_reranking(self, query, k)`

Lines 146-172

Purpose:

Search with LLM-based reranking/compression

Parameters: – query: Input query or question – k: K

Returns: <ast.Subscript object at 0x000001A06D7EE650>

`answer_question(self, question, search_method)`

Lines 174-199

Purpose:

Answer question using retrieved context

Parameters: – question: Question to answer – search_method: Search method

Returns: str

`find_similar_documents(self, reference_text, k)`

Lines 201-217

Purpose:

Find documents similar to reference text

Parameters: – reference_text: Reference text – k: K

Returns: <ast.Subscript object at 0x000001A06D7B3890>

`get_statistics(self)`

Lines 219-225

Purpose:

Get search system statistics

Parameters:

Returns: <ast.Subscript object at 0x000001A06D7B0050>

Functions

`demonstrate_semantic_search()`

Lines 228-362

Purpose:

Demonstrate semantic search and retrieval

Key Concepts Summary

Pattern Overview

The Semantic Search Retrieval pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

Agent Class: Main implementation of the pattern logic
Prompt Engineering: Carefully crafted prompts to guide LLM behavior
Tool Integration: Optional tools for extended functionality
Memory Management: State handling across interactions
Output Parsing: Structured response processing

When to Use

❌ 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

Prompt Optimization: Minimize tokens while maintaining clarity
Caching: Cache repeated queries/responses
Model Selection: Use smaller models when appropriate
Batch Processing: Group similar requests
Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

GPT-3.5-turbo: ~$0.002 per request
GPT-4: ~$0.03 per request
GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 126
Pattern 128

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 Semantic Search Retrieval 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: 127_semantic_search_retrieval.py
Total Lines: 367
Pattern Category: AI Agent Design Pattern #127

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Agentic Design Pattern: Knowledge Fusion


    """
Pattern 126: Knowledge Fusion

Description:
    Combines knowledge from multiple heterogeneous sources, resolving conflicts
    and creating a unified, consistent knowledge representation.

Components:
    - Multi-source integration
    - Conflict detection and resolution
    - Consistency checking
    - Knowledge merging strategies

Use Cases:
    - Multi-source data integration
    - Fact verification across sources
    - Building comprehensive knowledge bases

LangChain Implementation:
    Uses LangChain for analyzing, comparing, and merging knowledge from multiple sources.
"""

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

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.pydantic_v1 import BaseModel, Field

load_dotenv()


class KnowledgeSource(BaseModel):
    """Represents a knowledge source"""
    source_id: str = Field(description="Unique source identifier")
    name: str = Field(description="Source name")
    reliability: float = Field(description="Source reliability score 0-1")
    data: Dict[str, Any] = Field(description="Knowledge data")


class FusedKnowledge(BaseModel):
    """Fused knowledge item"""
    key: str = Field(description="Knowledge key/identifier")
    value: Any = Field(description="Fused value")
    confidence: float = Field(description="Confidence in fused value")
    sources: List[str] = Field(description="Contributing sources")
    conflicts: List[Dict[str, Any]] = Field(description="Detected conflicts", default_factory=list)


class KnowledgeFusionAgent:
    """Agent for fusing knowledge from multiple sources"""
    
    def __init__(self, model: str = "gpt-4o-mini", temperature: float = 0.0):
        self.llm = ChatOpenAI(model=model, temperature=temperature)
        self.sources: List[KnowledgeSource] = []
        self.fused_knowledge: Dict[str, FusedKnowledge] = {}
        
        # Conflict resolution prompt
        self.conflict_resolution_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an expert at resolving conflicting information from multiple sources.
Given conflicting facts from different sources with their reliability scores, determine:
1. Which fact is most likely correct
2. If a synthesis/combination is possible
3. Confidence level in the resolution

Return JSON with format:
{{
  "resolved_value": "the resolved/synthesized value",
  "confidence": 0.85,
  "reasoning": "explanation of resolution",
  "method": "voting|weighted|synthesis|newest"
}}"""),
            ("user", """Conflicting information about: {key}

Sources and values:
{conflicts}

Resolve the conflict.""")
        ])
        
        self.parser = JsonOutputParser()
    
    def add_source(self, source: KnowledgeSource):
        """Add a knowledge source"""
        self.sources.append(source)
        print(f"✓ Added source: {source.name} (reliability: {source.reliability})")
    
    def detect_conflicts(self, key: str, values: List[Dict[str, Any]]) -> bool:
        """Check if there are conflicts for a key"""
        if len(values)  1
    
    def simple_fusion(self, key: str, values: List[Dict[str, Any]]) -> FusedKnowledge:
        """Simple fusion without conflicts (voting or consensus)"""
        if len(values) == 1:
            # Single source - use as is
            return FusedKnowledge(
                key=key,
                value=values[0]['value'],
                confidence=values[0]['reliability'],
                sources=[values[0]['source_id']]
            )
        
        # Multiple sources with same value - average confidence
        avg_confidence = sum(v['reliability'] for v in values) / len(values)
        sources = [v['source_id'] for v in values]
        
        return FusedKnowledge(
            key=key,
            value=values[0]['value'],
            confidence=avg_confidence,
            sources=sources
        )
    
    def resolve_conflict(self, key: str, values: List[Dict[str, Any]]) -> FusedKnowledge:
        """Resolve conflicting values using LLM"""
        print(f"\n⚠️  Conflict detected for '{key}'")
        
        # Prepare conflict description
        conflicts_text = ""
        conflict_list = []
        for v in values:
            conflicts_text += f"- Source '{v['source_name']}' (reliability: {v['reliability']}): {v['value']}\n"
            conflict_list.append({
                "source": v['source_name'],
                "value": v['value'],
                "reliability": v['reliability']
            })
        
        print(f"   Conflicting values:\n{conflicts_text}")
        
        try:
            # Use LLM to resolve
            chain = self.conflict_resolution_prompt | self.llm | self.parser
            result = chain.invoke({"key": key, "conflicts": conflicts_text})
            
            print(f"   Resolution: {result['resolved_value']}")
            print(f"   Method: {result['method']}, Confidence: {result['confidence']}")
            print(f"   Reasoning: {result['reasoning']}")
            
            return FusedKnowledge(
                key=key,
                value=result['resolved_value'],
                confidence=result['confidence'],
                sources=[v['source_id'] for v in values],
                conflicts=conflict_list
            )
        except Exception as e:
            print(f"   Error resolving conflict: {e}")
            # Fallback: use most reliable source
            best_source = max(values, key=lambda x: x['reliability'])
            return FusedKnowledge(
                key=key,
                value=best_source['value'],
                confidence=best_source['reliability'] * 0.8,  # Reduce confidence
                sources=[v['source_id'] for v in values],
                conflicts=conflict_list
            )
    
    def fuse_knowledge(self) -> Dict[str, FusedKnowledge]:
        """Fuse knowledge from all sources"""
        print("\n" + "=" * 80)
        print("🔄 Starting Knowledge Fusion Process")
        print("=" * 80)
        
        # Group knowledge by key
        knowledge_by_key = defaultdict(list)
        
        for source in self.sources:
            for key, value in source.data.items():
                knowledge_by_key[key].append({
                    'source_id': source.source_id,
                    'source_name': source.name,
                    'reliability': source.reliability,
                    'value': value
                })
        
        print(f"\n📊 Found {len(knowledge_by_key)} unique knowledge keys")
        
        # Fuse each key
        conflicts_detected = 0
        for key, values in knowledge_by_key.items():
            if self.detect_conflicts(key, values):
                conflicts_detected += 1
                self.fused_knowledge[key] = self.resolve_conflict(key, values)
            else:
                self.fused_knowledge[key] = self.simple_fusion(key, values)
        
        print(f"\n✓ Fusion complete!")
        print(f"  - Total keys: {len(self.fused_knowledge)}")
        print(f"  - Conflicts resolved: {conflicts_detected}")
        
        return self.fused_knowledge
    
    def query(self, key: str) -> Optional[FusedKnowledge]:
        """Query fused knowledge"""
        return self.fused_knowledge.get(key)
    
    def get_high_confidence_facts(self, threshold: float = 0.8) -> List[FusedKnowledge]:
        """Get facts with confidence above threshold"""
        return [
            fact for fact in self.fused_knowledge.values()
            if fact.confidence >= threshold
        ]
    
    def get_conflicted_facts(self) -> List[FusedKnowledge]:
        """Get facts that had conflicts"""
        return [
            fact for fact in self.fused_knowledge.values()
            if fact.conflicts
        ]
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get fusion statistics"""
        total_facts = len(self.fused_knowledge)
        conflicted = len(self.get_conflicted_facts())
        high_confidence = len(self.get_high_confidence_facts())
        
        avg_confidence = sum(f.confidence for f in self.fused_knowledge.values()) / total_facts if total_facts > 0 else 0
        
        return {
            "total_sources": len(self.sources),
            "total_facts": total_facts,
            "conflicted_facts": conflicted,
            "high_confidence_facts": high_confidence,
            "average_confidence": avg_confidence
        }


def demonstrate_knowledge_fusion():
    """Demonstrate knowledge fusion from multiple sources"""
    print("=" * 80)
    print("Pattern 126: Knowledge Fusion")
    print("=" * 80)
    
    agent = KnowledgeFusionAgent()
    
    # Example 1: Add multiple knowledge sources
    print("\n" + "=" * 80)
    print("EXAMPLE 1: Adding Knowledge Sources")
    print("=" * 80)
    
    # Source 1: Wikipedia (high reliability)
    wikipedia = KnowledgeSource(
        source_id="wikipedia",
        name="Wikipedia",
        reliability=0.85,
        data={
            "eiffel_tower_height": "330 meters",
            "eiffel_tower_built": "1889",
            "eiffel_tower_location": "Paris, France",
            "mount_everest_height": "8,849 meters",
            "python_creator": "Guido van Rossum"
        }
    )
    
    # Source 2: Tourist website (medium reliability)
    tourist_site = KnowledgeSource(
        source_id="tourist_site",
        name="Tourism Website",
        reliability=0.65,
        data={
            "eiffel_tower_height": "324 meters",  # Conflict!
            "eiffel_tower_built": "1889",
            "eiffel_tower_location": "Paris, France",
            "eiffel_tower_visitors": "7 million per year"
        }
    )
    
    # Source 3: News article (medium reliability)
    news = KnowledgeSource(
        source_id="news",
        name="News Article",
        reliability=0.70,
        data={
            "eiffel_tower_height": "330 meters including antenna",  # Conflict!
            "mount_everest_height": "8,848.86 meters",  # Conflict!
            "python_creator": "Guido van Rossum",
            "python_first_release": "1991"
        }
    )
    
    agent.add_source(wikipedia)
    agent.add_source(tourist_site)
    agent.add_source(news)
    
    # Example 2: Fuse knowledge
    print("\n" + "=" * 80)
    print("EXAMPLE 2: Fusing Knowledge")
    print("=" * 80)
    
    fused = agent.fuse_knowledge()
    
    # Example 3: Query fused knowledge
    print("\n" + "=" * 80)
    print("EXAMPLE 3: Querying Fused Knowledge")
    print("=" * 80)
    
    queries = [
        "eiffel_tower_height",
        "eiffel_tower_location",
        "python_creator",
        "mount_everest_height"
    ]
    
    for query in queries:
        result = agent.query(query)
        if result:
            print(f"\n🔍 Query: {query}")
            print(f"   Value: {result.value}")
            print(f"   Confidence: {result.confidence:.2f}")
            print(f"   Sources: {', '.join(result.sources)}")
            if result.conflicts:
                print(f"   ⚠️  Had conflicts from {len(result.conflicts)} sources")
    
    # Example 4: High confidence facts
    print("\n" + "=" * 80)
    print("EXAMPLE 4: High Confidence Facts")
    print("=" * 80)
    
    high_conf = agent.get_high_confidence_facts(threshold=0.75)
    print(f"\n✓ Found {len(high_conf)} high-confidence facts (>0.75):\n")
    for fact in high_conf:
        print(f"  - {fact.key}: {fact.value} (confidence: {fact.confidence:.2f})")
    
    # Example 5: Conflicted facts
    print("\n" + "=" * 80)
    print("EXAMPLE 5: Facts with Conflicts")
    print("=" * 80)
    
    conflicted = agent.get_conflicted_facts()
    print(f"\n⚠️  Found {len(conflicted)} facts with conflicts:\n")
    for fact in conflicted:
        print(f"  - {fact.key}:")
        print(f"      Resolved to: {fact.value}")
        print(f"      Original values:")
        for conflict in fact.conflicts:
            print(f"        • {conflict['source']}: {conflict['value']}")
    
    # Example 6: Statistics
    print("\n" + "=" * 80)
    print("EXAMPLE 6: Fusion Statistics")
    print("=" * 80)
    
    stats = agent.get_statistics()
    print(f"\n📊 Statistics:")
    print(f"  Total Sources: {stats['total_sources']}")
    print(f"  Total Facts: {stats['total_facts']}")
    print(f"  Conflicted Facts: {stats['conflicted_facts']}")
    print(f"  High Confidence Facts: {stats['high_confidence_facts']}")
    print(f"  Average Confidence: {stats['average_confidence']:.2f}")
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY")
    print("=" * 80)
    print("""
Knowledge Fusion Pattern:
- Integrates knowledge from multiple heterogeneous sources
- Detects and resolves conflicting information
- Uses source reliability scores for weighting
- Creates unified, consistent knowledge representation
- Maintains provenance and conflict information

Conflict Resolution Strategies:
✓ Voting (majority wins)
✓ Weighted by source reliability
✓ LLM-based synthesis
✓ Temporal (newest information)
✓ Consensus building

Key Benefits:
✓ Comprehensive knowledge coverage
✓ Improved accuracy through verification
✓ Conflict awareness and resolution
✓ Source tracking and attribution
✓ Confidence scoring
    """)


if __name__ == "__main__":
    demonstrate_knowledge_fusion()

Agentic Design Pattern: Knowledge Fusion

Pattern 126: Knowledge Fusion – Line-by-Line Explanation

Overview

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

Pattern Description

Pattern 126: Knowledge Fusion

Description: Combines knowledge from multiple heterogeneous sources, resolving conflicts and creating a unified, consistent knowledge representation.

Components: – Multi-source integration – Conflict detection and resolution – Consistency checking – Knowledge merging strategies

Use Cases: – Multi-source data integration – Fact verification across sources – Building comprehensive knowledge bases

LangChain Implementation: Uses LangChain for analyzing, comparing, and merging knowledge from multiple sources.

File Statistics

Total Lines: 396
Classes: 3
Functions: 1
Constants: 0

Module Documentation

"""
Pattern 126: Knowledge Fusion

Description:
    Combines knowledge from multiple heterogeneous sources, resolving conflicts
    and creating a unified, consistent knowledge representation.

Components:
    - Multi-source integration
    - Conflict detection and resolution
    - Consistency checking
    - Knowledge merging strategies

Use Cases:
    - Multi-source data integration
    - Fact verification across sources
    - Building comprehensive knowledge bases

LangChain Implementation:
    Uses LangChain for analyzing, comparing, and merging knowledge from multiple sources.
"""

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

Import Statements

Standard Library Imports

import collections.defaultdict
import datetime.datetime
import langchain_core.output_parsers.JsonOutputParser
import langchain_core.prompts.ChatPromptTemplate
import langchain_core.pydantic_v1.BaseModel
import langchain_core.pydantic_v1.Field
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Set

Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – datetime.datetime: Date and time handling – langchain_core.output_parsers.JsonOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.pydantic_v1.BaseModel: Regular expression operations – langchain_core.pydantic_v1.Field: 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) – 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: `KnowledgeSource`

Lines 37-42

Inherits from: BaseModel

Purpose:

Represents a knowledge source

Class: `FusedKnowledge`

Lines 45-51

Inherits from: BaseModel

Purpose:

Fused knowledge item

Class: `KnowledgeFusionAgent`

Lines 54-240

Purpose:

Agent for fusing knowledge from multiple sources

Methods:

`init(self, model, temperature)`

Lines 57-85

Purpose: Initialize the class instance with required parameters.

Parameters: – model: Language model instance – temperature: Sampling temperature (0-1)

`add_source(self, source)`

Lines 87-90

Purpose:

Add a knowledge source

Parameters: – source: Source

`detect_conflicts(self, key, values)`

Lines 92-104

Purpose:

Check if there are conflicts for a key

Parameters: – key: Key – values: Values

Returns: bool

`simple_fusion(self, key, values)`

Lines 106-126

Purpose:

Simple fusion without conflicts (voting or consensus)

Parameters: – key: Key – values: Values

Returns: FusedKnowledge

`resolve_conflict(self, key, values)`

Lines 128-171

Purpose:

Resolve conflicting values using LLM

Parameters: – key: Key – values: Values

Returns: FusedKnowledge

`fuse_knowledge(self)`

Lines 173-206

Purpose:

Fuse knowledge from all sources

Parameters:

Returns: <ast.Subscript object at 0x000001A06D88A210>

`query(self, key)`

Lines 208-210

Purpose:

Query fused knowledge

Parameters: – key: Key

Returns: <ast.Subscript object at 0x000001A06D889D10>

`get_high_confidence_facts(self, threshold)`

Lines 212-217

Purpose:

Get facts with confidence above threshold

Parameters: – threshold: Threshold

Returns: <ast.Subscript object at 0x000001A06D64A710>

`get_conflicted_facts(self)`

Lines 219-224

Purpose:

Get facts that had conflicts

Parameters:

Returns: <ast.Subscript object at 0x000001A06D64BD10>

`get_statistics(self)`

Lines 226-240

Purpose:

Get fusion statistics

Parameters:

Returns: <ast.Subscript object at 0x000001A06D748190>

Functions

`demonstrate_knowledge_fusion()`

Lines 243-391

Purpose:

Demonstrate knowledge fusion from multiple sources

Key Concepts Summary

Pattern Overview

The Knowledge Fusion pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

Agent Class: Main implementation of the pattern logic
Prompt Engineering: Carefully crafted prompts to guide LLM behavior
Tool Integration: Optional tools for extended functionality
Memory Management: State handling across interactions
Output Parsing: Structured response processing

When to Use

❌ 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

Prompt Optimization: Minimize tokens while maintaining clarity
Caching: Cache repeated queries/responses
Model Selection: Use smaller models when appropriate
Batch Processing: Group similar requests
Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

GPT-3.5-turbo: ~$0.002 per request
GPT-4: ~$0.03 per request
GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 125
Pattern 127

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 Fusion 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: 126_knowledge_fusion.py
Total Lines: 396
Pattern Category: AI Agent Design Pattern #126

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Agentic Design Pattern: Knowledge Extraction Mining


    """
Pattern 125: Knowledge Extraction & Mining

Description:
    Automatically extracts structured knowledge from unstructured data sources using NER,
    relation extraction, and information extraction techniques.

Components:
    - Named Entity Recognition (NER)
    - Relation extraction
    - Information extraction pipelines
    - Knowledge base building

Use Cases:
    - Building knowledge bases from documents
    - Extracting facts from text
    - Document processing and indexing

LangChain Implementation:
    Uses LangChain for entity extraction, relation identification, and knowledge structuring.
"""

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

from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.pydantic_v1 import BaseModel, Field

load_dotenv()


# Pydantic models for structured output
class Entity(BaseModel):
    """An entity extracted from text"""
    text: str = Field(description="The entity text")
    type: str = Field(description="Entity type (PERSON, ORG, LOCATION, DATE, etc.)")
    confidence: float = Field(description="Confidence score 0-1")


class Relation(BaseModel):
    """A relation between two entities"""
    subject: str = Field(description="Subject entity")
    predicate: str = Field(description="Relation type")
    object: str = Field(description="Object entity")
    confidence: float = Field(description="Confidence score 0-1")


class KnowledgeFact(BaseModel):
    """A structured knowledge fact"""
    entities: List[Entity] = Field(description="Entities in the fact")
    relations: List[Relation] = Field(description="Relations between entities")
    source_text: str = Field(description="Original text")
    timestamp: str = Field(description="Extraction timestamp")


class KnowledgeExtractionAgent:
    """Agent for extracting structured knowledge from text"""
    
    def __init__(self, model: str = "gpt-4o-mini", temperature: float = 0.0):
        self.llm = ChatOpenAI(model=model, temperature=temperature)
        self.knowledge_base: List[KnowledgeFact] = []
        
        # Entity extraction chain
        self.entity_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an expert at Named Entity Recognition (NER).
Extract all entities from the text and classify them into types:
PERSON, ORGANIZATION, LOCATION, DATE, EVENT, PRODUCT, TECHNOLOGY, CONCEPT, etc.

Return JSON with format:
{{
  "entities": [
    {{"text": "entity name", "type": "TYPE", "confidence": 0.95}}
  ]
}}"""),
            ("user", "Text: {text}")
        ])
        
        # Relation extraction chain
        self.relation_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an expert at extracting semantic relations between entities.
Given a text and its entities, extract meaningful relations in the form:
Subject - Predicate - Object

Common relation types:
- works_for, founded_by, located_in, part_of
- born_in, created_on, acquired_by
- invented, developed, collaborated_with

Return JSON with format:
{{
  "relations": [
    {{"subject": "Entity1", "predicate": "relation_type", "object": "Entity2", "confidence": 0.9}}
  ]
}}"""),
            ("user", "Text: {text}\n\nEntities: {entities}")
        ])
        
        self.entity_parser = JsonOutputParser()
        self.relation_parser = JsonOutputParser()
        
    def extract_entities(self, text: str) -> List[Entity]:
        """Extract named entities from text"""
        try:
            chain = self.entity_prompt | self.llm | self.entity_parser
            result = chain.invoke({"text": text})
            
            entities = []
            for e in result.get("entities", []):
                entities.append(Entity(
                    text=e["text"],
                    type=e["type"],
                    confidence=e.get("confidence", 0.8)
                ))
            return entities
        except Exception as e:
            print(f"Entity extraction error: {e}")
            return []
    
    def extract_relations(self, text: str, entities: List[Entity]) -> List[Relation]:
        """Extract relations between entities"""
        try:
            entity_str = ", ".join([f"{e.text} ({e.type})" for e in entities])
            
            chain = self.relation_prompt | self.llm | self.relation_parser
            result = chain.invoke({"text": text, "entities": entity_str})
            
            relations = []
            for r in result.get("relations", []):
                relations.append(Relation(
                    subject=r["subject"],
                    predicate=r["predicate"],
                    object=r["object"],
                    confidence=r.get("confidence", 0.8)
                ))
            return relations
        except Exception as e:
            print(f"Relation extraction error: {e}")
            return []
    
    def extract_knowledge(self, text: str) -> KnowledgeFact:
        """Extract complete knowledge from text"""
        print(f"\n🔍 Extracting knowledge from text...")
        
        # Step 1: Extract entities
        entities = self.extract_entities(text)
        print(f"\n📌 Extracted {len(entities)} entities:")
        for e in entities:
            print(f"  - {e.text} ({e.type}) - confidence: {e.confidence:.2f}")
        
        # Step 2: Extract relations
        relations = self.extract_relations(text, entities)
        print(f"\n🔗 Extracted {len(relations)} relations:")
        for r in relations:
            print(f"  - {r.subject} --[{r.predicate}]--> {r.object}")
        
        # Create knowledge fact
        fact = KnowledgeFact(
            entities=entities,
            relations=relations,
            source_text=text,
            timestamp=datetime.now().isoformat()
        )
        
        # Add to knowledge base
        self.knowledge_base.append(fact)
        
        return fact
    
    def batch_extract(self, texts: List[str]) -> List[KnowledgeFact]:
        """Extract knowledge from multiple texts"""
        print(f"\n📚 Processing {len(texts)} documents...")
        facts = []
        
        for i, text in enumerate(texts, 1):
            print(f"\n--- Document {i}/{len(texts)} ---")
            fact = self.extract_knowledge(text)
            facts.append(fact)
        
        return facts
    
    def query_knowledge_base(self, entity_name: str) -> List[Dict[str, Any]]:
        """Query knowledge base for entity"""
        results = []
        
        for fact in self.knowledge_base:
            # Check if entity appears in this fact
            for entity in fact.entities:
                if entity_name.lower() in entity.text.lower():
                    # Find related entities and relations
                    related = {
                        "entity": entity.text,
                        "type": entity.type,
                        "relations": [],
                        "source": fact.source_text[:100] + "..."
                    }
                    
                    for rel in fact.relations:
                        if entity_name.lower() in rel.subject.lower():
                            related["relations"].append({
                                "type": rel.predicate,
                                "target": rel.object
                            })
                        elif entity_name.lower() in rel.object.lower():
                            related["relations"].append({
                                "type": f"inverse_{rel.predicate}",
                                "target": rel.subject
                            })
                    
                    if related["relations"]:
                        results.append(related)
        
        return results
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get knowledge base statistics"""
        total_entities = sum(len(f.entities) for f in self.knowledge_base)
        total_relations = sum(len(f.relations) for f in self.knowledge_base)
        
        entity_types = {}
        for fact in self.knowledge_base:
            for entity in fact.entities:
                entity_types[entity.type] = entity_types.get(entity.type, 0) + 1
        
        relation_types = {}
        for fact in self.knowledge_base:
            for relation in fact.relations:
                relation_types[relation.predicate] = relation_types.get(relation.predicate, 0) + 1
        
        return {
            "total_documents": len(self.knowledge_base),
            "total_entities": total_entities,
            "total_relations": total_relations,
            "entity_types": entity_types,
            "relation_types": relation_types
        }


def demonstrate_knowledge_extraction():
    """Demonstrate knowledge extraction and mining"""
    print("=" * 80)
    print("Pattern 125: Knowledge Extraction & Mining")
    print("=" * 80)
    
    agent = KnowledgeExtractionAgent()
    
    # Example 1: Extract from single document
    print("\n" + "=" * 80)
    print("EXAMPLE 1: Single Document Extraction")
    print("=" * 80)
    
    text1 = """
    OpenAI was founded in December 2015 by Sam Altman, Elon Musk, and others in San Francisco.
    The company developed ChatGPT, which was released on November 30, 2022. ChatGPT uses GPT-4,
    a large language model trained on massive amounts of text data.
    """
    
    fact1 = agent.extract_knowledge(text1)
    
    # Example 2: Batch extraction
    print("\n" + "=" * 80)
    print("EXAMPLE 2: Batch Document Processing")
    print("=" * 80)
    
    documents = [
        """Microsoft was founded by Bill Gates and Paul Allen in 1975 in Albuquerque, New Mexico.
        The company later moved to Redmond, Washington.""",
        
        """Google was founded by Larry Page and Sergey Brin in 1998 while they were Ph.D. students
        at Stanford University. The company is headquartered in Mountain View, California.""",
        
        """Apple Inc. was co-founded by Steve Jobs, Steve Wozniak, and Ronald Wayne in 1976.
        The company introduced the iPhone in 2007, revolutionizing the smartphone industry."""
    ]
    
    facts = agent.batch_extract(documents)
    
    # Example 3: Query knowledge base
    print("\n" + "=" * 80)
    print("EXAMPLE 3: Querying Knowledge Base")
    print("=" * 80)
    
    queries = ["OpenAI", "Bill Gates", "iPhone"]
    
    for query in queries:
        print(f"\n🔍 Query: '{query}'")
        results = agent.query_knowledge_base(query)
        
        if results:
            for result in results:
                print(f"\n  Entity: {result['entity']} ({result['type']})")
                print(f"  Relations:")
                for rel in result['relations']:
                    print(f"    - {rel['type']}: {rel['target']}")
        else:
            print("  No results found")
    
    # Example 4: Knowledge base statistics
    print("\n" + "=" * 80)
    print("EXAMPLE 4: Knowledge Base Statistics")
    print("=" * 80)
    
    stats = agent.get_statistics()
    
    print(f"\n📊 Statistics:")
    print(f"  Total Documents: {stats['total_documents']}")
    print(f"  Total Entities: {stats['total_entities']}")
    print(f"  Total Relations: {stats['total_relations']}")
    
    print(f"\n📌 Entity Types:")
    for entity_type, count in sorted(stats['entity_types'].items(), key=lambda x: x[1], reverse=True):
        print(f"  - {entity_type}: {count}")
    
    print(f"\n🔗 Relation Types:")
    for rel_type, count in sorted(stats['relation_types'].items(), key=lambda x: x[1], reverse=True):
        print(f"  - {rel_type}: {count}")
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY")
    print("=" * 80)
    print("""
Knowledge Extraction & Mining Pattern:
- Automatically extracts structured knowledge from text
- Identifies entities (people, organizations, locations, etc.)
- Discovers relations between entities
- Builds queryable knowledge base
- Enables semantic search and knowledge discovery

Key Benefits:
✓ Automated knowledge acquisition
✓ Structured representation of information
✓ Relationship discovery
✓ Scalable processing
✓ Knowledge base building
    """)


if __name__ == "__main__":
    demonstrate_knowledge_extraction()

Agentic Design Pattern: Knowledge Extraction Mining

Pattern 125: Knowledge Extraction Mining – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Knowledge Extraction Mining pattern implementation.

Pattern Description

Pattern 125: Knowledge Extraction & Mining

Description: Automatically extracts structured knowledge from unstructured data sources using NER, relation extraction, and information extraction techniques.

Components: – Named Entity Recognition (NER) – Relation extraction – Information extraction pipelines – Knowledge base building

Use Cases: – Building knowledge bases from documents – Extracting facts from text – Document processing and indexing

LangChain Implementation: Uses LangChain for entity extraction, relation identification, and knowledge structuring.

File Statistics

Total Lines: 344
Classes: 4
Functions: 1
Constants: 0

Module Documentation

"""
Pattern 125: Knowledge Extraction & Mining

Description:
    Automatically extracts structured knowledge from unstructured data sources using NER,
    relation extraction, and information extraction techniques.

Components:
    - Named Entity Recognition (NER)
    - Relation extraction
    - Information extraction pipelines
    - Knowledge base building

Use Cases:
    - Building knowledge bases from documents
    - Extracting facts from text
    - Document processing and indexing

LangChain Implementation:
    Uses LangChain for entity extraction, relation identification, and knowledge structuring.
"""

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

Import Statements

Standard Library Imports

import datetime.datetime
import langchain_core.output_parsers.JsonOutputParser
import langchain_core.prompts.ChatPromptTemplate
import langchain_core.prompts.FewShotChatMessagePromptTemplate
import langchain_core.pydantic_v1.BaseModel
import langchain_core.pydantic_v1.Field
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Tuple

Purpose: Core Python functionality: – datetime.datetime: Date and time handling – langchain_core.output_parsers.JsonOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.prompts.FewShotChatMessagePromptTemplate: Regular expression operations – langchain_core.pydantic_v1.BaseModel: Regular expression operations – langchain_core.pydantic_v1.Field: 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) – 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: `Entity`

Lines 37-41

Inherits from: BaseModel

Purpose:

An entity extracted from text

Class: `Relation`

Lines 44-49

Inherits from: BaseModel

Purpose:

A relation between two entities

Class: `KnowledgeFact`

Lines 52-57

Inherits from: BaseModel

Purpose:

A structured knowledge fact

Class: `KnowledgeExtractionAgent`

Lines 60-239

Purpose:

Agent for extracting structured knowledge from text

Methods:

`init(self, model, temperature)`

Lines 63-103

Purpose: Initialize the class instance with required parameters.

Parameters: – model: Language model instance – temperature: Sampling temperature (0-1)

`extract_entities(self, text)`

Lines 105-121

Purpose:

Extract named entities from text

Parameters: – text: Text

Returns: <ast.Subscript object at 0x000001A06D7B11D0>

`extract_relations(self, text, entities)`

Lines 123-142

Purpose:

Extract relations between entities

Parameters: – text: Text – entities: Entities

Returns: <ast.Subscript object at 0x000001A06D812A50>

`extract_knowledge(self, text)`

Lines 144-171

Purpose:

Extract complete knowledge from text

Parameters: – text: Text

Returns: KnowledgeFact

`batch_extract(self, texts)`

Lines 173-183

Purpose:

Extract knowledge from multiple texts

Parameters: – texts: Texts

Returns: <ast.Subscript object at 0x000001A06D77E910>

`query_knowledge_base(self, entity_name)`

Lines 185-216

Purpose:

Query knowledge base for entity

Parameters: – entity_name: Entity name

Returns: <ast.Subscript object at 0x000001A06D63AB50>

`get_statistics(self)`

Lines 218-239

Purpose:

Get knowledge base statistics

Parameters:

Returns: <ast.Subscript object at 0x000001A06D7B8550>

Functions

`demonstrate_knowledge_extraction()`

Lines 242-339

Purpose:

Demonstrate knowledge extraction and mining

Key Concepts Summary

Pattern Overview

The Knowledge Extraction Mining pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

Agent Class: Main implementation of the pattern logic
Prompt Engineering: Carefully crafted prompts to guide LLM behavior
Tool Integration: Optional tools for extended functionality
Memory Management: State handling across interactions
Output Parsing: Structured response processing

When to Use

❌ 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

Prompt Optimization: Minimize tokens while maintaining clarity
Caching: Cache repeated queries/responses
Model Selection: Use smaller models when appropriate
Batch Processing: Group similar requests
Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

GPT-3.5-turbo: ~$0.002 per request
GPT-4: ~$0.03 per request
GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 124
Pattern 126

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 Extraction Mining 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: 125_knowledge_extraction_mining.py
Total Lines: 344
Pattern Category: AI Agent Design Pattern #125

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Agentic Design Pattern: Ontology Based Reasoning


    """
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

Agent Class: Main implementation of the pattern logic
Prompt Engineering: Carefully crafted prompts to guide LLM behavior
Tool Integration: Optional tools for extended functionality
Memory Management: State handling across interactions
Output Parsing: Structured response processing

When to Use

❌ 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

Prompt Optimization: Minimize tokens while maintaining clarity
Caching: Cache repeated queries/responses
Model Selection: Use smaller models when appropriate
Batch Processing: Group similar requests
Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

GPT-3.5-turbo: ~$0.002 per request
GPT-4: ~$0.03 per request
GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

Agentic Design Pattern: Knowledge Graph Integration


    """
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

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

Agent Class: Main implementation of the pattern logic
Prompt Engineering: Carefully crafted prompts to guide LLM behavior
Tool Integration: Optional tools for extended functionality
Memory Management: State handling across interactions
Output Parsing: Structured response processing

When to Use

❌ 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

Prompt Optimization: Minimize tokens while maintaining clarity
Caching: Cache repeated queries/responses
Model Selection: Use smaller models when appropriate
Batch Processing: Group similar requests
Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

GPT-3.5-turbo: ~$0.002 per request
GPT-4: ~$0.03 per request
GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

Agentic Design Pattern: Service Mesh Pattern


    """
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

Agent Class: Main implementation of the pattern logic
Prompt Engineering: Carefully crafted prompts to guide LLM behavior
Tool Integration: Optional tools for extended functionality
Memory Management: State handling across interactions
Output Parsing: Structured response processing

When to Use

❌ 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

Prompt Optimization: Minimize tokens while maintaining clarity
Caching: Cache repeated queries/responses
Model Selection: Use smaller models when appropriate
Batch Processing: Group similar requests
Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

GPT-3.5-turbo: ~$0.002 per request
GPT-4: ~$0.03 per request
GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

Agentic Design Pattern: Event Driven Architecture


    """
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

Agent Class: Main implementation of the pattern logic
Prompt Engineering: Carefully crafted prompts to guide LLM behavior
Tool Integration: Optional tools for extended functionality
Memory Management: State handling across interactions
Output Parsing: Structured response processing

When to Use

❌ 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

Prompt Optimization: Minimize tokens while maintaining clarity
Caching: Cache repeated queries/responses
Model Selection: Use smaller models when appropriate
Batch Processing: Group similar requests
Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

GPT-3.5-turbo: ~$0.002 per request
GPT-4: ~$0.03 per request
GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

Agentic Design Pattern: Workflow Orchestration


    """
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

Agent Class: Main implementation of the pattern logic
Prompt Engineering: Carefully crafted prompts to guide LLM behavior
Tool Integration: Optional tools for extended functionality
Memory Management: State handling across interactions
Output Parsing: Structured response processing

When to Use

❌ 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

Prompt Optimization: Minimize tokens while maintaining clarity
Caching: Cache repeated queries/responses
Model Selection: Use smaller models when appropriate
Batch Processing: Group similar requests
Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

GPT-3.5-turbo: ~$0.002 per request
GPT-4: ~$0.03 per request
GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

Agentic Design Pattern: Task Allocation Scheduling


    """
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

Agent Class: Main implementation of the pattern logic
Prompt Engineering: Carefully crafted prompts to guide LLM behavior
Tool Integration: Optional tools for extended functionality
Memory Management: State handling across interactions
Output Parsing: Structured response processing

When to Use

❌ 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

Prompt Optimization: Minimize tokens while maintaining clarity
Caching: Cache repeated queries/responses
Model Selection: Use smaller models when appropriate
Batch Processing: Group similar requests
Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

GPT-3.5-turbo: ~$0.002 per request
GPT-4: ~$0.03 per request
GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

Further Improvements

Production Enhancements

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

Feature Extensions

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

Code Quality

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

Related Patterns

Pattern 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

November 2, 2025

Tag: llm

Agentic Design Pattern: Multi Turn Dialogue

Agentic Design Pattern: Multi Turn Dialogue

Pattern 128: Multi Turn Dialogue – Line-by-Line Explanation

Overview

Pattern Description

File Statistics

Module Documentation

Import Statements

Standard Library Imports

LangChain Imports

Other Imports

Classes

Class: DialogueState

Class: UserIntent

Class: DialogueContext

Class: MultiTurnDialogueAgent

__init__(self, model, temperature)

detect_intent(self, message)

update_context(self, intent, message)

generate_response(self, message)

chat(self, user_message)

get_conversation_summary(self)

reset_conversation(self)

Functions

demonstrate_multi_turn_dialogue()

Key Concepts Summary

Pattern Overview

Core Components

When to Use

Best Practices Demonstrated

1. Environment Configuration

2. Error Handling

3. Prompt Engineering

4. Code Organization

5. Performance Optimization

LangChain Components Used

Execution Flow

Performance Considerations

Computational Cost

Optimization Strategies

Cost Estimates (Approximate)

Common Pitfalls to Avoid

1. Prompt Design Issues

2. Error Handling

3. Token Management

4. Security

5. Testing

6. Cost Management

Further Improvements

Production Enhancements

Feature Extensions

Code Quality

Related Patterns

Summary

Agentic Design Pattern: Semantic Search Retrieval

Agentic Design Pattern: Semantic Search Retrieval

Pattern 127: Semantic Search Retrieval – Line-by-Line Explanation

Overview

Pattern Description

File Statistics

Module Documentation

Import Statements

Standard Library Imports

LangChain Imports

Other Imports

Classes

Class: SemanticSearchAgent

__init__(self, model, temperature)

load_documents(self, documents, metadatas)

semantic_search(self, query, k)

hybrid_search(self, query, k)

search_with_reranking(self, query, k)

answer_question(self, question, search_method)

find_similar_documents(self, reference_text, k)

get_statistics(self)

Functions

demonstrate_semantic_search()

Key Concepts Summary

Pattern Overview

Class: `DialogueState`

Class: `UserIntent`

Class: `DialogueContext`

Class: `MultiTurnDialogueAgent`

`init(self, model, temperature)`

`detect_intent(self, message)`

`update_context(self, intent, message)`

`generate_response(self, message)`

`chat(self, user_message)`

`get_conversation_summary(self)`

`reset_conversation(self)`

`demonstrate_multi_turn_dialogue()`

Class: `SemanticSearchAgent`

`init(self, model, temperature)`

`load_documents(self, documents, metadatas)`

`semantic_search(self, query, k)`

`hybrid_search(self, query, k)`

`search_with_reranking(self, query, k)`

`answer_question(self, question, search_method)`

`find_similar_documents(self, reference_text, k)`

`get_statistics(self)`

`demonstrate_semantic_search()`

Class: `KnowledgeSource`

Class: `FusedKnowledge`

Class: `KnowledgeFusionAgent`

`init(self, model, temperature)`

`add_source(self, source)`

`detect_conflicts(self, key, values)`

`simple_fusion(self, key, values)`

`resolve_conflict(self, key, values)`

`fuse_knowledge(self)`

`query(self, key)`

`get_high_confidence_facts(self, threshold)`

`get_conflicted_facts(self)`

`get_statistics(self)`

`demonstrate_knowledge_fusion()`