Author: admin

Agentic Design Pattern: Constitutional Ai


    """
Pattern 010: Constitutional AI

Description:
    Ensures agent behavior aligns with explicit principles or "constitution."
    The agent generates responses, evaluates them against constitutional principles,
    and revises if violations are found. This creates safer, more aligned AI systems.

Key Concepts:
    - Constitution: Set of principles/rules the agent must follow
    - Self-Critique: Agent evaluates its own outputs against principles
    - Revision: Generate improved response that adheres to constitution
    - Iterative Alignment: Multiple rounds of critique and revision
    - Transparency: Explicit principles make behavior interpretable

Constitutional Principles Examples:
    - Harmlessness: Avoid harmful, dangerous, or offensive content
    - Helpfulness: Provide useful, relevant information
    - Honesty: Be truthful, acknowledge uncertainty
    - Privacy: Respect privacy and confidentiality
    - Fairness: Avoid bias and discrimination

Use Cases:
    - Production AI systems requiring safety
    - Content moderation and filtering
    - Ethical decision-making systems
    - Regulated domains (healthcare, finance, legal)

LangChain Implementation:
    Constitutional chain with critique and revision loops based on principles.
"""

import os
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
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 environment variables
load_dotenv()


@dataclass
class Principle:
    """Represents a constitutional principle."""
    name: str
    description: str
    critique_prompt: str
    weight: float = 1.0  # Importance weight


@dataclass
class ViolationReport:
    """Report of constitutional violations."""
    principle: str
    violated: bool
    explanation: str
    severity: float = 0.0  # 0.0 (no violation) to 1.0 (severe)


@dataclass
class ConstitutionalReview:
    """Complete review against constitution."""
    original_output: str
    violations: List[ViolationReport] = field(default_factory=list)
    total_severity: float = 0.0
    passes: bool = True
    
    def calculate_severity(self):
        """Calculate total severity of violations."""
        self.total_severity = sum(v.severity for v in self.violations if v.violated)
        self.passes = self.total_severity == 0.0


class Constitution:
    """Set of principles for constitutional AI."""
    
    def __init__(self, name: str = "Default Constitution"):
        """Initialize constitution with default or custom principles."""
        self.name = name
        self.principles: List[Principle] = []
    
    def add_principle(self, principle: Principle):
        """Add a principle to the constitution."""
        self.principles.append(principle)
    
    def get_principles_text(self) -> str:
        """Get formatted text of all principles."""
        lines = [f"Constitution: {self.name}\n"]
        for i, p in enumerate(self.principles, 1):
            lines.append(f"{i}. {p.name}: {p.description}")
        return "\n".join(lines)
    
    @staticmethod
    def get_default() -> 'Constitution':
        """Get default constitution with common principles."""
        constitution = Constitution("Default AI Constitution")
        
        # Harmlessness
        constitution.add_principle(Principle(
            name="Harmlessness",
            description="Responses must not include harmful, dangerous, illegal, or offensive content",
            critique_prompt="""Does the response contain any harmful, dangerous, illegal, or offensive content?
Consider: violence, illegal activities, self-harm, harassment, discrimination.
Answer YES if violations exist, NO if the response is harmless."""
        ))
        
        # Helpfulness
        constitution.add_principle(Principle(
            name="Helpfulness",
            description="Responses should be helpful, relevant, and answer the user's question",
            critique_prompt="""Is the response helpful and relevant to the user's query?
Does it provide useful information or assistance?
Answer YES if unhelpful, NO if the response is helpful."""
        ))
        
        # Honesty
        constitution.add_principle(Principle(
            name="Honesty",
            description="Responses should be truthful and acknowledge uncertainty when appropriate",
            critique_prompt="""Does the response contain false information or make claims without acknowledging uncertainty?
Answer YES if dishonest or overconfident, NO if honest and appropriately uncertain."""
        ))
        
        # Privacy
        constitution.add_principle(Principle(
            name="Privacy",
            description="Responses must respect privacy and not request or share personal information inappropriately",
            critique_prompt="""Does the response inappropriately request, share, or handle personal information?
Answer YES if privacy is violated, NO if privacy is respected."""
        ))
        
        # Fairness
        constitution.add_principle(Principle(
            name="Fairness",
            description="Responses should be fair and avoid bias or discrimination",
            critique_prompt="""Does the response show bias or discrimination based on race, gender, religion, or other protected characteristics?
Answer YES if biased or discriminatory, NO if fair and balanced."""
        ))
        
        return constitution


class ConstitutionalAI:
    """Agent that follows constitutional principles."""
    
    def __init__(self, constitution: Constitution,
                 model_name: str = "gpt-3.5-turbo",
                 max_revisions: int = 3,
                 severity_threshold: float = 0.3):
        """
        Initialize Constitutional AI agent.
        
        Args:
            constitution: The constitution to follow
            model_name: Name of the OpenAI model
            max_revisions: Maximum number of revision attempts
            severity_threshold: Severity threshold for triggering revision
        """
        self.constitution = constitution
        self.max_revisions = max_revisions
        self.severity_threshold = severity_threshold
        
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=0.7,
            api_key=os.getenv("OPENAI_API_KEY")
        )
        
        self.critic_llm = ChatOpenAI(
            model=model_name,
            temperature=0.2,  # Lower temperature for consistent critique
            api_key=os.getenv("OPENAI_API_KEY")
        )
    
    def generate_initial(self, query: str, context: str = "") -> str:
        """
        Generate initial response to query.
        
        Args:
            query: User query
            context: Additional context
            
        Returns:
            Initial response
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a helpful AI assistant. Respond to the user's query."),
            ("human", """{context}

Query: {query}

Response:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        response = chain.invoke({
            "query": query,
            "context": context if context else ""
        })
        
        return response.strip()
    
    def critique_against_principle(self, output: str, query: str,
                                   principle: Principle) -> ViolationReport:
        """
        Critique output against a single principle.
        
        Args:
            output: Output to critique
            query: Original query
            principle: Principle to check against
            
        Returns:
            ViolationReport indicating if principle is violated
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are evaluating whether a response violates the principle: "{principle.name}"

Principle: {principle.description}

{principle.critique_prompt}

If YES (violation), also rate severity from 0.1 (minor) to 1.0 (severe).

Format your response as:
VIOLATION: YES/NO
SEVERITY: [0.0-1.0]
EXPLANATION: [Brief explanation]"""),
            ("human", """Query: {query}

Response to evaluate:
{output}

Evaluation:""")
        ])
        
        chain = prompt | self.critic_llm | StrOutputParser()
        critique_text = chain.invoke({
            "query": query,
            "output": output
        })
        
        # Parse critique
        violated = False
        severity = 0.0
        explanation = ""
        
        for line in critique_text.split('\n'):
            line = line.strip()
            if line.upper().startswith('VIOLATION:'):
                violated = 'YES' in line.upper()
            elif line.upper().startswith('SEVERITY:'):
                try:
                    severity_str = line.split(':', 1)[1].strip()
                    severity = float(severity_str)
                except:
                    severity = 0.5 if violated else 0.0
            elif line.upper().startswith('EXPLANATION:'):
                explanation = line.split(':', 1)[1].strip()
        
        return ViolationReport(
            principle=principle.name,
            violated=violated,
            explanation=explanation,
            severity=severity if violated else 0.0
        )
    
    def review(self, output: str, query: str) -> ConstitutionalReview:
        """
        Review output against all constitutional principles.
        
        Args:
            output: Output to review
            query: Original query
            
        Returns:
            ConstitutionalReview with all violations
        """
        review = ConstitutionalReview(original_output=output)
        
        for principle in self.constitution.principles:
            violation = self.critique_against_principle(output, query, principle)
            review.violations.append(violation)
        
        review.calculate_severity()
        
        return review
    
    def revise(self, query: str, output: str, review: ConstitutionalReview) -> str:
        """
        Revise output to address constitutional violations.
        
        Args:
            query: Original query
            output: Current output with violations
            review: Review identifying violations
            
        Returns:
            Revised output addressing violations
        """
        # Build violation summary
        violations_text = []
        for v in review.violations:
            if v.violated:
                violations_text.append(f"- {v.principle}: {v.explanation} (severity: {v.severity:.2f})")
        
        violations_summary = "\n".join(violations_text) if violations_text else "No violations"
        
        # Get principles text
        principles_text = self.constitution.get_principles_text()
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are revising a response to address constitutional violations.

{principles_text}

Revise the response to:
1. Address all identified violations
2. Maintain helpfulness and relevance
3. Fully comply with all principles

Generate a new, improved response that adheres to the constitution."""),
            ("human", """Original Query: {query}

Current Response:
{output}

Violations Found:
{violations}

Revised Response:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        revised = chain.invoke({
            "query": query,
            "output": output,
            "violations": violations_summary
        })
        
        return revised.strip()
    
    def generate_with_constitution(self, query: str, context: str = "",
                                   verbose: bool = True) -> Dict[str, Any]:
        """
        Generate response following constitutional principles.
        
        Args:
            query: User query
            context: Additional context
            verbose: Whether to print progress
            
        Returns:
            Dictionary with final response and review history
        """
        if verbose:
            print(f"\nQuery: {query}\n")
            print(f"Constitution: {self.constitution.name}")
            print(f"Principles: {len(self.constitution.principles)}\n")
        
        # Generate initial response
        if verbose:
            print("="*60)
            print("INITIAL GENERATION")
            print("="*60)
        
        current_output = self.generate_initial(query, context)
        
        if verbose:
            print(f"\nGenerated:\n{current_output[:200]}..." if len(current_output) > 200 else f"\nGenerated:\n{current_output}")
        
        reviews = []
        revision_count = 0
        
        # Review and revise loop
        for iteration in range(self.max_revisions + 1):
            if verbose:
                print(f"\n{'='*60}")
                print(f"CONSTITUTIONAL REVIEW {iteration + 1}")
                print("="*60)
            
            # Review against constitution
            review = self.review(current_output, query)
            reviews.append(review)
            
            if verbose:
                print(f"\nViolations Found: {sum(1 for v in review.violations if v.violated)}/{len(review.violations)}")
                print(f"Total Severity: {review.total_severity:.2f}")
                
                for v in review.violations:
                    if v.violated:
                        print(f"  ⚠ {v.principle}: {v.explanation} (severity: {v.severity:.2f})")
                    else:
                        print(f"  ✓ {v.principle}: OK")
            
            # Check if passes constitutional review
            if review.passes or review.total_severity = self.max_revisions:
                if verbose:
                    print(f"\n⚠ Max revisions ({self.max_revisions}) reached")
                break
            
            # Revise
            if verbose:
                print(f"\n{'='*60}")
                print(f"REVISION {iteration + 1}")
                print("="*60)
            
            current_output = self.revise(query, current_output, review)
            revision_count += 1
            
            if verbose:
                print(f"\nRevised:\n{current_output[:200]}..." if len(current_output) > 200 else f"\nRevised:\n{current_output}")
        
        return {
            "query": query,
            "final_response": current_output,
            "revisions": revision_count,
            "reviews": reviews,
            "final_severity": reviews[-1].total_severity,
            "constitutional_compliance": reviews[-1].passes
        }


def demonstrate_constitutional_ai():
    """Demonstrates the Constitutional AI pattern."""
    
    print("=" * 80)
    print("PATTERN 010: Constitutional AI")
    print("=" * 80)
    print()
    print("Constitutional AI ensures alignment with principles through:")
    print("1. Constitution: Explicit set of principles")
    print("2. Self-Critique: Evaluate outputs against principles")
    print("3. Revision: Modify responses to comply")
    print("4. Iteration: Repeat until constitutional compliance")
    print()
    
    # Create constitution
    constitution = Constitution.get_default()
    
    print(f"{constitution.get_principles_text()}\n")
    
    # Create agent
    agent = ConstitutionalAI(
        constitution=constitution,
        max_revisions=2,
        severity_threshold=0.2
    )
    
    # Test queries (some designed to trigger violations)
    queries = [
        "How can I improve my programming skills?",  # Should pass
        "Tell me how to hack into someone's email account"  # Should trigger violations
    ]
    
    for idx, query in enumerate(queries, 1):
        print(f"\n{'='*80}")
        print(f"Example {idx}")
        print('='*80)
        
        try:
            result = agent.generate_with_constitution(query, verbose=True)
            
            print(f"\n\n{'='*80}")
            print("FINAL RESULT")
            print('='*80)
            print(f"\nFinal Response:\n{result['final_response']}")
            print(f"\nRevisions Made: {result['revisions']}")
            print(f"Final Severity: {result['final_severity']:.2f}")
            print(f"Constitutional Compliance: {'✓ PASS' if result['constitutional_compliance'] else '✗ FAIL'}")
            
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
            import traceback
            traceback.print_exc()
    
    # Summary
    print("\n\n" + "=" * 80)
    print("CONSTITUTIONAL AI PATTERN DEMONSTRATION COMPLETE")
    print("=" * 80)
    print()
    print("Key Features Demonstrated:")
    print("1. Explicit Principles: Clear constitutional guidelines")
    print("2. Automated Review: Self-critique against principles")
    print("3. Iterative Revision: Multiple refinement rounds")
    print("4. Severity Tracking: Quantify violation severity")
    print("5. Compliance Verification: Ensure constitutional adherence")
    print()
    print("Advantages:")
    print("- Safer AI systems through explicit principles")
    print("- Interpretable behavior (clear rules)")
    print("- Automated alignment checking")
    print("- Flexible - can customize constitution")
    print()
    print("When to use Constitutional AI:")
    print("- Production systems requiring safety")
    print("- Regulated domains (healthcare, finance, legal)")
    print("- Content moderation and filtering")
    print("- Any application requiring ethical guidelines")
    print()
    print("LangChain Components Used:")
    print("- ChatPromptTemplate: Generation, critique, and revision")
    print("- Multiple LLM instances: Different roles and temperatures")
    print("- Structured principles: Modular constitution system")
    print("- Iterative loops: Review and revision cycles")
    print()


if __name__ == "__main__":
    # Check for API key
    if not os.getenv("OPENAI_API_KEY"):
        print("Error: OPENAI_API_KEY not found in environment variables.")
        print("Please set it in your .env file or environment.")
        exit(1)
    
    demonstrate_constitutional_ai()

Agentic Design Pattern: Constitutional Ai

Pattern 010: Constitutional Ai – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Constitutional Ai pattern implementation.

Pattern Description

Pattern 010: Constitutional AI

Description: Ensures agent behavior aligns with explicit principles or “constitution.” The agent generates responses, evaluates them against constitutional principles, and revises if violations are found. This creates safer, more aligned AI systems.

Key Concepts: – Constitution: Set of principles/rules the agent must follow – Self-Critique: Agent evaluates its own outputs against principles – Revision: Generate improved response that adheres to constitution – Iterative Alignment: Multiple rounds of critique and revision – Transparency: Explicit principles make behavior interpretable

Constitutional Principles Examples: – Harmlessness: Avoid harmful, dangerous, or offensive content – Helpfulness: Provide useful, relevant information – Honesty: Be truthful, acknowledge uncertainty – Privacy: Respect privacy and confidentiality – Fairness: Avoid bias and discrimination

Use Cases: – Production AI systems requiring safety – Content moderation and filtering – Ethical decision-making systems – Regulated domains (healthcare, finance, legal)

LangChain Implementation: Constitutional chain with critique and revision loops based on principles.

File Statistics

Total Lines: 528
Classes: 5
Functions: 1
Constants: 0

Module Documentation

"""
Pattern 010: Constitutional AI

Description:
    Ensures agent behavior aligns with explicit principles or "constitution."
    The agent generates responses, evaluates them against constitutional principles,
    and revises if violations are found. This creates safer, more aligned AI systems.

Key Concepts:
    - Constitution: Set of principles/rules the agent must follow
    - Self-Critique: Agent evaluates its own outputs against principles
    - Revision: Generate improved response that adheres to constitution
    - Iterative Alignment: Multiple rounds of critique and revision
    - Transparency: Explicit principles make behavior interpretable

Constitutional Principles Examples:
    - Harmlessness: Avoid harmful, dangerous, or offensive content
    - Helpfulness: Provide useful, relevant information
    - Honesty: Be truthful, acknowledge uncertainty
    - Privacy: Respect privacy and confidentiality
    - Fairness: Avoid bias and discrimination

Use Cases:
    - Production AI systems requiring safety
    - Content moderation and filtering
    - Ethical decision-making systems
    - Regulated domains (healthcare, finance, legal)

LangChain Implementation:
    Constitutional chain with critique and revision loops based on principles.
"""

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

Import Statements

Standard Library Imports

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

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – 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.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
from traceback

Purpose: Additional dependencies for specialized functionality.

Classes

Class: `Principle`

Lines 48-53

Purpose:

Represents a constitutional principle.

Class: `ViolationReport`

Lines 57-62

Purpose:

Report of constitutional violations.

Class: `ConstitutionalReview`

Lines 66-76

Purpose:

Complete review against constitution.

Methods:

`calculate_severity(self)`

Lines 73-76

Purpose:

Calculate total severity of violations.

Parameters:

Class: `Constitution`

Lines 79-145

Purpose:

Set of principles for constitutional AI.

Methods:

`init(self, name)`

Lines 82-85

Purpose:

Initialize constitution with default or custom principles.

Parameters: – name: Name

`add_principle(self, principle)`

Lines 87-89

Purpose:

Add a principle to the constitution.

Parameters: – principle: Principle

`get_principles_text(self)`

Lines 91-96

Purpose:

Get formatted text of all principles.

Parameters:

Returns: str

`get_default()`

Lines 99-145

Decorators: staticmethod

Purpose:

Get default constitution with common principles.

Returns: Constitution

Class: `ConstitutionalAI`

Lines 148-432

Purpose:

Agent that follows constitutional principles.

Methods:

`init(self, constitution, model_name, max_revisions, severity_threshold)`

Lines 151-178

Purpose:

Initialize Constitutional AI agent.

Args:
    constitution: The constitution to follow
    model_name: Name of the OpenAI model
    max_revisions: Maximum number of revision attempts
    severity_threshold: Severity threshold for triggering revision

Parameters: – constitution: Constitution – model_name: Language model instance – max_revisions: Max revisions – severity_threshold: Severity threshold

`generate_initial(self, query, context)`

Lines 180-206

Purpose:

Generate initial response to query.

Args:
    query: User query
    context: Additional context

Returns:
    Initial response

Parameters: – query: Input query or question – context: Context

Returns: str

`critique_against_principle(self, output, query, principle)`

Lines 208-271

Purpose:

Critique output against a single principle.

Args:
    output: Output to critique
    query: Original query
    principle: Principle to check against

Returns:
    ViolationReport indicating if principle is violated

Parameters: – output: Output – query: Input query or question – principle: Principle

Returns: ViolationReport

`review(self, output, query)`

Lines 273-292

Purpose:

Review output against all constitutional principles.

Args:
    output: Output to review
    query: Original query

Returns:
    ConstitutionalReview with all violations

Parameters: – output: Output – query: Input query or question

Returns: ConstitutionalReview

`revise(self, query, output, review)`

Lines 294-346

Purpose:

Revise output to address constitutional violations.

Args:
    query: Original query
    output: Current output with violations
    review: Review identifying violations

Returns:
    Revised output addressing violations

Parameters: – query: Input query or question – output: Output – review: Review

Returns: str

`generate_with_constitution(self, query, context, verbose)`

Lines 348-432

Purpose:

Generate response following constitutional principles.

Args:
    query: User query
    context: Additional context
    verbose: Whether to print progress

Returns:
    Dictionary with final response and review history

Parameters: – query: Input query or question – context: Context – verbose: Enable detailed logging

Returns: <ast.Subscript object at 0x000001A06D607710>

Functions

`demonstrate_constitutional_ai()`

Lines 435-517

Purpose:

Demonstrates the Constitutional AI pattern.

Key Concepts Summary

Pattern Overview

The Constitutional Ai 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 009
Pattern 011
Other Reasoning Patterns (001-011)

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 Constitutional Ai 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: 010_constitutional_ai.py
Total Lines: 528
Pattern Category: AI Agent Design Pattern #010

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Agentic Design Pattern: Self Refine


    """
Pattern 009: Self-Refine

Description:
    Iteratively improves outputs through self-critique and refinement.
    The agent generates an initial output, evaluates it, identifies weaknesses,
    and produces an improved version. This cycle repeats until quality threshold
    is met or max iterations reached.

Key Concepts:
    - Initial Generation: Create first draft output
    - Self-Critique: Analyze weaknesses and areas for improvement
    - Refinement: Generate improved version based on critique
    - Iteration: Repeat until satisfactory or max iterations
    - Quality Tracking: Monitor improvement across iterations

Benefits:
    - Higher quality outputs through iteration
    - Self-improvement without external feedback
    - Transparency in refinement process
    - Handles complex generation tasks

Use Cases:
    - Content creation (writing, code, reports)
    - Problem-solving with iterative improvement
    - Design tasks requiring multiple revisions
    - Any task benefiting from refinement

LangChain Implementation:
    Chained prompts for generation, critique, and refinement with iteration control.
"""

import os
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
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 environment variables
load_dotenv()


class IterationStatus(Enum):
    """Status of refinement iteration."""
    INITIAL = "initial"
    REFINING = "refining"
    COMPLETED = "completed"
    MAX_ITERATIONS = "max_iterations_reached"


@dataclass
class Critique:
    """Represents a critique of generated output."""
    strengths: List[str] = field(default_factory=list)
    weaknesses: List[str] = field(default_factory=list)
    suggestions: List[str] = field(default_factory=list)
    score: float = 0.0  # Quality score 0-1


@dataclass
class RefinementIteration:
    """Represents one iteration of the refinement process."""
    iteration: int
    output: str
    critique: Optional[Critique] = None
    status: IterationStatus = IterationStatus.INITIAL
    
    def __str__(self):
        return f"Iteration {self.iteration}: {self.status.value} (score: {self.critique.score if self.critique else 'N/A'})"


class SelfRefineAgent:
    """Agent that uses self-refinement to improve outputs."""
    
    def __init__(self, model_name: str = "gpt-3.5-turbo", 
                 max_iterations: int = 3,
                 quality_threshold: float = 0.8):
        """
        Initialize the Self-Refine agent.
        
        Args:
            model_name: Name of the OpenAI model
            max_iterations: Maximum refinement iterations
            quality_threshold: Stop if quality score exceeds this
        """
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=0.7,
            api_key=os.getenv("OPENAI_API_KEY")
        )
        
        self.critic_llm = ChatOpenAI(
            model=model_name,
            temperature=0.3,  # Lower temperature for more consistent critique
            api_key=os.getenv("OPENAI_API_KEY")
        )
        
        self.max_iterations = max_iterations
        self.quality_threshold = quality_threshold
        self.history: List[RefinementIteration] = []
    
    def generate_initial(self, task: str, context: str = "") -> str:
        """
        Generate initial output for the task.
        
        Args:
            task: The task description
            context: Additional context
            
        Returns:
            Initial generated output
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Generate a response to the given task.
Focus on creating a solid first draft. It doesn't need to be perfect yet."""),
            ("human", """{context}

Task: {task}

Response:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        output = chain.invoke({
            "task": task,
            "context": context if context else "No additional context."
        })
        
        return output.strip()
    
    def critique(self, task: str, output: str, iteration: int) -> Critique:
        """
        Generate critique of current output.
        
        Args:
            task: Original task
            output: Current output to critique
            iteration: Current iteration number
            
        Returns:
            Critique object with feedback
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a constructive critic. Evaluate the response to the task.

Provide:
1. Strengths: What works well (2-3 points)
2. Weaknesses: What could be improved (2-3 points)
3. Suggestions: Specific improvements to make (2-3 points)
4. Score: Overall quality score from 0.0 to 1.0

Format your response as:
STRENGTHS:
- [strength 1]
- [strength 2]

WEAKNESSES:
- [weakness 1]
- [weakness 2]

SUGGESTIONS:
- [suggestion 1]
- [suggestion 2]

SCORE: [0.0-1.0]"""),
            ("human", """Task: {task}

Current Response (Iteration {iteration}):
{output}

Critique:""")
        ])
        
        chain = prompt | self.critic_llm | StrOutputParser()
        critique_text = chain.invoke({
            "task": task,
            "output": output,
            "iteration": iteration
        })
        
        # Parse critique
        critique = Critique()
        
        sections = {
            "strengths": [],
            "weaknesses": [],
            "suggestions": []
        }
        
        current_section = None
        for line in critique_text.split('\n'):
            line = line.strip()
            
            if line.upper().startswith('STRENGTHS:'):
                current_section = "strengths"
            elif line.upper().startswith('WEAKNESSES:'):
                current_section = "weaknesses"
            elif line.upper().startswith('SUGGESTIONS:'):
                current_section = "suggestions"
            elif line.upper().startswith('SCORE:'):
                try:
                    score_str = line.split(':', 1)[1].strip()
                    critique.score = float(score_str)
                except:
                    critique.score = 0.5
            elif line.startswith('-') and current_section:
                sections[current_section].append(line.lstrip('- ').strip())
        
        critique.strengths = sections["strengths"]
        critique.weaknesses = sections["weaknesses"]
        critique.suggestions = sections["suggestions"]
        
        # Ensure score is set
        if critique.score == 0.0 and (critique.strengths or not critique.weaknesses):
            critique.score = 0.6  # Default reasonable score
        
        return critique
    
    def refine(self, task: str, current_output: str, critique: Critique) -> str:
        """
        Generate refined version based on critique.
        
        Args:
            task: Original task
            current_output: Current output to refine
            critique: Critique with improvement suggestions
            
        Returns:
            Refined output
        """
        weaknesses_text = "\n".join([f"- {w}" for w in critique.weaknesses])
        suggestions_text = "\n".join([f"- {s}" for s in critique.suggestions])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Improve the response based on the critique provided.
Address the weaknesses and implement the suggestions while maintaining the strengths.
Produce a refined, higher-quality version."""),
            ("human", """Task: {task}

Current Response:
{current_output}

Weaknesses to Address:
{weaknesses}

Suggestions to Implement:
{suggestions}

Refined Response:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        refined = chain.invoke({
            "task": task,
            "current_output": current_output,
            "weaknesses": weaknesses_text,
            "suggestions": suggestions_text
        })
        
        return refined.strip()
    
    def refine_iteratively(self, task: str, context: str = "",
                          verbose: bool = True) -> Dict[str, Any]:
        """
        Perform iterative refinement on the task.
        
        Args:
            task: The task to complete
            context: Additional context
            verbose: Whether to print progress
            
        Returns:
            Dictionary with final output and refinement history
        """
        if verbose:
            print(f"\nTask: {task}\n")
            print("="*60)
            print("ITERATION 1: Initial Generation")
            print("="*60)
        
        # Generate initial output
        initial_output = self.generate_initial(task, context)
        
        iteration = RefinementIteration(
            iteration=1,
            output=initial_output,
            status=IterationStatus.INITIAL
        )
        self.history.append(iteration)
        
        if verbose:
            print(f"\nGenerated:\n{initial_output[:200]}..." if len(initial_output) > 200 else f"\nGenerated:\n{initial_output}")
        
        current_output = initial_output
        
        # Refinement loop
        for i in range(2, self.max_iterations + 2):
            if verbose:
                print(f"\n{'='*60}")
                print(f"ITERATION {i}: Critique & Refine")
                print("="*60)
            
            # Critique current output
            critique = self.critique(task, current_output, i - 1)
            
            # Update previous iteration with critique
            self.history[-1].critique = critique
            
            if verbose:
                print(f"\nCritique:")
                print(f"  Quality Score: {critique.score:.2f}")
                if critique.strengths:
                    print(f"  Strengths: {len(critique.strengths)} identified")
                if critique.weaknesses:
                    print(f"  Weaknesses: {len(critique.weaknesses)} identified")
                if critique.suggestions:
                    print(f"  Suggestions: {len(critique.suggestions)} provided")
            
            # Check if quality threshold met
            if critique.score >= self.quality_threshold:
                self.history[-1].status = IterationStatus.COMPLETED
                if verbose:
                    print(f"\n✓ Quality threshold ({self.quality_threshold}) met!")
                break
            
            # Check max iterations
            if i > self.max_iterations:
                self.history[-1].status = IterationStatus.MAX_ITERATIONS
                if verbose:
                    print(f"\n⚠ Max iterations ({self.max_iterations}) reached")
                break
            
            # Refine output
            if verbose:
                print(f"\nRefining...")
            
            refined_output = self.refine(task, current_output, critique)
            
            iteration = RefinementIteration(
                iteration=i,
                output=refined_output,
                status=IterationStatus.REFINING
            )
            self.history.append(iteration)
            
            if verbose:
                print(f"\nRefined:\n{refined_output[:200]}..." if len(refined_output) > 200 else f"\nRefined:\n{refined_output}")
            
            current_output = refined_output
        
        # Final critique
        if self.history[-1].critique is None:
            final_critique = self.critique(task, current_output, len(self.history))
            self.history[-1].critique = final_critique
        
        return {
            "task": task,
            "final_output": current_output,
            "iterations": len(self.history),
            "final_score": self.history[-1].critique.score if self.history[-1].critique else 0.0,
            "history": self.history,
            "status": self.history[-1].status.value
        }
    
    def get_improvement_summary(self) -> str:
        """Generate summary of improvements across iterations."""
        if len(self.history) < 2:
            return "Not enough iterations to compare."
        
        initial_score = self.history[0].critique.score if self.history[0].critique else 0.0
        final_score = self.history[-1].critique.score if self.history[-1].critique else 0.0
        
        improvement = final_score - initial_score
        improvement_pct = (improvement / max(initial_score, 0.1)) * 100
        
        summary = f"""
Improvement Summary:
  Initial Score: {initial_score:.2f}
  Final Score: {final_score:.2f}
  Improvement: {improvement:+.2f} ({improvement_pct:+.1f}%)
  Total Iterations: {len(self.history)}
"""
        return summary


def demonstrate_self_refine():
    """Demonstrates the Self-Refine pattern."""
    
    print("=" * 80)
    print("PATTERN 009: Self-Refine")
    print("=" * 80)
    print()
    print("Self-Refine iteratively improves outputs through:")
    print("1. Initial Generation: Create first draft")
    print("2. Self-Critique: Analyze strengths and weaknesses")
    print("3. Refinement: Generate improved version")
    print("4. Iteration: Repeat until quality threshold met")
    print()
    
    # Create agent
    agent = SelfRefineAgent(max_iterations=3, quality_threshold=0.85)
    
    # Test tasks
    tasks = [
        {
            "task": "Write a brief explanation of quantum entanglement for a general audience",
            "context": "Make it accessible but accurate, around 100 words"
        },
        {
            "task": "Create a function to calculate the Fibonacci sequence efficiently",
            "context": "Use Python, include docstring and handle edge cases"
        }
    ]
    
    for idx, test_case in enumerate(tasks, 1):
        print(f"\n{'='*80}")
        print(f"Example {idx}")
        print('='*80)
        
        try:
            # Reset history
            agent.history = []
            
            result = agent.refine_iteratively(
                task=test_case['task'],
                context=test_case['context'],
                verbose=True
            )
            
            print(f"\n\n{'='*80}")
            print("FINAL RESULT")
            print('='*80)
            print(f"\nFinal Output:\n{result['final_output']}")
            print(f"\n{agent.get_improvement_summary()}")
            print(f"Status: {result['status']}")
            
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
            import traceback
            traceback.print_exc()
    
    # Summary
    print("\n\n" + "=" * 80)
    print("SELF-REFINE PATTERN DEMONSTRATION COMPLETE")
    print("=" * 80)
    print()
    print("Key Features Demonstrated:")
    print("1. Iterative Improvement: Multiple refinement cycles")
    print("2. Self-Critique: Autonomous evaluation of outputs")
    print("3. Targeted Refinement: Addressing specific weaknesses")
    print("4. Quality Tracking: Score-based improvement monitoring")
    print("5. Convergence: Stops when threshold met or max iterations")
    print()
    print("Advantages:")
    print("- Higher quality outputs through iteration")
    print("- No external feedback required")
    print("- Transparent improvement process")
    print("- Handles complex generation tasks")
    print()
    print("When to use Self-Refine:")
    print("- Content creation (writing, code, reports)")
    print("- Tasks benefiting from multiple revisions")
    print("- When quality is more important than speed")
    print("- Complex generation requiring refinement")
    print()
    print("LangChain Components Used:")
    print("- ChatPromptTemplate: Generation, critique, and refinement prompts")
    print("- StrOutputParser: Parse LLM outputs")
    print("- Separate LLMs: Different temperatures for generation vs critique")
    print("- Iteration control: Quality threshold and max iterations")
    print()


if __name__ == "__main__":
    # Check for API key
    if not os.getenv("OPENAI_API_KEY"):
        print("Error: OPENAI_API_KEY not found in environment variables.")
        print("Please set it in your .env file or environment.")
        exit(1)
    
    demonstrate_self_refine()

Agentic Design Pattern: Self Refine

Pattern 009: Self Refine – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Self Refine pattern implementation.

Pattern Description

Pattern 009: Self-Refine

Description: Iteratively improves outputs through self-critique and refinement. The agent generates an initial output, evaluates it, identifies weaknesses, and produces an improved version. This cycle repeats until quality threshold is met or max iterations reached.

Key Concepts: – Initial Generation: Create first draft output – Self-Critique: Analyze weaknesses and areas for improvement – Refinement: Generate improved version based on critique – Iteration: Repeat until satisfactory or max iterations – Quality Tracking: Monitor improvement across iterations

Benefits: – Higher quality outputs through iteration – Self-improvement without external feedback – Transparency in refinement process – Handles complex generation tasks

Use Cases: – Content creation (writing, code, reports) – Problem-solving with iterative improvement – Design tasks requiring multiple revisions – Any task benefiting from refinement

LangChain Implementation: Chained prompts for generation, critique, and refinement with iteration control.

File Statistics

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

Module Documentation

"""
Pattern 009: Self-Refine

Description:
    Iteratively improves outputs through self-critique and refinement.
    The agent generates an initial output, evaluates it, identifies weaknesses,
    and produces an improved version. This cycle repeats until quality threshold
    is met or max iterations reached.

Key Concepts:
    - Initial Generation: Create first draft output
    - Self-Critique: Analyze weaknesses and areas for improvement
    - Refinement: Generate improved version based on critique
    - Iteration: Repeat until satisfactory or max iterations
    - Quality Tracking: Monitor improvement across iterations

Benefits:
    - Higher quality outputs through iteration
    - Self-improvement without external feedback
    - Transparency in refinement process
    - Handles complex generation tasks

Use Cases:
    - Content creation (writing, code, reports)
    - Problem-solving with iterative improvement
    - Design tasks requiring multiple revisions
    - Any task benefiting from refinement

LangChain Implementation:
    Chained prompts for generation, critique, and refinement with iteration control.
"""

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

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.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
from traceback

Purpose: Additional dependencies for specialized functionality.

Classes

Class: `IterationStatus`

Lines 47-52

Inherits from: Enum

Purpose:

Status of refinement iteration.

Class: `Critique`

Lines 56-61

Purpose:

Represents a critique of generated output.

Class: `RefinementIteration`

Lines 65-73

Purpose:

Represents one iteration of the refinement process.

Methods:

`str(self)`

Lines 72-73

Purpose: Str operation.

Parameters:

Class: `SelfRefineAgent`

Lines 76-387

Purpose:

Agent that uses self-refinement to improve outputs.

Methods:

`init(self, model_name, max_iterations, quality_threshold)`

Lines 79-104

Purpose:

Initialize the Self-Refine agent.

Args:
    model_name: Name of the OpenAI model
    max_iterations: Maximum refinement iterations
    quality_threshold: Stop if quality score exceeds this

Parameters: – model_name: Language model instance – max_iterations: Max iterations – quality_threshold: Quality threshold

`generate_initial(self, task, context)`

Lines 106-133

Purpose:

Generate initial output for the task.

Args:
    task: The task description
    context: Additional context

Returns:
    Initial generated output

Parameters: – task: Task – context: Context

Returns: str

`critique(self, task, output, iteration)`

Lines 135-221

Purpose:

Generate critique of current output.

Args:
    task: Original task
    output: Current output to critique
    iteration: Current iteration number

Returns:
    Critique object with feedback

Parameters: – task: Task – output: Output – iteration: Iteration

Returns: Critique

`refine(self, task, current_output, critique)`

Lines 223-264

Purpose:

Generate refined version based on critique.

Args:
    task: Original task
    current_output: Current output to refine
    critique: Critique with improvement suggestions

Returns:
    Refined output

Parameters: – task: Task – current_output: Current output – critique: Critique

Returns: str

`refine_iteratively(self, task, context, verbose)`

Lines 266-367

Purpose:

Perform iterative refinement on the task.

Args:
    task: The task to complete
    context: Additional context
    verbose: Whether to print progress

Returns:
    Dictionary with final output and refinement history

Parameters: – task: Task – context: Context – verbose: Enable detailed logging

Returns: <ast.Subscript object at 0x000001A06D658D90>

`get_improvement_summary(self)`

Lines 369-387

Purpose:

Generate summary of improvements across iterations.

Parameters:

Returns: str

Functions

`demonstrate_self_refine()`

Lines 390-475

Purpose:

Demonstrates the Self-Refine pattern.

Key Concepts Summary

Pattern Overview

The Self Refine 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 008
Pattern 010
Other Reasoning Patterns (001-011)

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 Self Refine 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: 009_self_refine.py
Total Lines: 486
Pattern Category: AI Agent Design Pattern #009

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Agentic Design Pattern: Llm Compiler


    """
Pattern 008: LLM Compiler

Description:
    Optimizes LLM execution by analyzing task dependencies, parallelizing
    independent operations, and optimizing the execution plan like a compiler
    optimizes code. Combines planning with execution optimization.

Key Concepts:
    - Task Decomposition: Break complex tasks into subtasks
    - Dependency Analysis: Build dependency graph (DAG)
    - Optimization: Identify parallelizable operations
    - Scheduling: Optimal execution order
    - Resource Management: Efficient LLM call allocation

Compiler Phases:
    1. Parsing: Understand and decompose the task
    2. Analysis: Build dependency graph
    3. Optimization: Find parallelization opportunities
    4. Code Generation: Create execution plan
    5. Execution: Run optimized plan

Use Cases:
    - Complex multi-step workflows
    - Research tasks with multiple branches
    - Data processing pipelines
    - Tasks with many independent subtasks

LangChain Implementation:
    Uses dependency graphs and parallel execution with LangChain tools.
"""

import os
from typing import List, Dict, Any, Set, Optional
from dataclasses import dataclass, field
from collections import defaultdict, deque
from concurrent.futures import ThreadPoolExecutor, as_completed
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 environment variables
load_dotenv()


class TaskStatus(Enum):
    """Status of a task in the execution plan."""
    PENDING = "pending"
    READY = "ready"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"


@dataclass
class Task:
    """Represents a single task in the execution plan."""
    id: str
    description: str
    dependencies: Set[str] = field(default_factory=set)
    status: TaskStatus = TaskStatus.PENDING
    result: Optional[str] = None
    error: Optional[str] = None
    parallel_group: Optional[int] = None  # For grouping parallel tasks
    
    def can_execute(self, completed_tasks: Set[str]) -> bool:
        """Check if this task can be executed."""
        return (self.status == TaskStatus.PENDING and 
                self.dependencies.issubset(completed_tasks))


@dataclass
class ExecutionPlan:
    """Represents the optimized execution plan."""
    goal: str
    tasks: Dict[str, Task] = field(default_factory=dict)
    dependency_graph: Dict[str, Set[str]] = field(default_factory=lambda: defaultdict(set))
    parallel_groups: List[List[str]] = field(default_factory=list)
    
    def add_task(self, task: Task):
        """Add a task to the plan."""
        self.tasks[task.id] = task
        for dep in task.dependencies:
            self.dependency_graph[dep].add(task.id)
    
    def get_ready_tasks(self, completed: Set[str]) -> List[Task]:
        """Get all tasks that are ready to execute."""
        ready = []
        for task in self.tasks.values():
            if task.can_execute(completed):
                ready.append(task)
        return ready
    
    def topological_sort(self) -> List[str]:
        """Perform topological sort on tasks."""
        in_degree = {task_id: len(task.dependencies) 
                     for task_id, task in self.tasks.items()}
        
        queue = deque([task_id for task_id, degree in in_degree.items() if degree == 0])
        sorted_tasks = []
        
        while queue:
            task_id = queue.popleft()
            sorted_tasks.append(task_id)
            
            for dependent in self.dependency_graph[task_id]:
                in_degree[dependent] -= 1
                if in_degree[dependent] == 0:
                    queue.append(dependent)
        
        return sorted_tasks


class LLMCompiler:
    """Compiler that optimizes LLM task execution."""
    
    def __init__(self, model_name: str = "gpt-3.5-turbo", max_parallel: int = 3):
        """
        Initialize the LLM Compiler.
        
        Args:
            model_name: Name of the OpenAI model
            max_parallel: Maximum number of parallel tasks
        """
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=0.3,
            api_key=os.getenv("OPENAI_API_KEY")
        )
        self.max_parallel = max_parallel
    
    def parse_task(self, goal: str) -> ExecutionPlan:
        """
        Parse the goal and decompose into subtasks.
        
        Args:
            goal: The high-level goal
            
        Returns:
            Initial execution plan with tasks
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Decompose the goal into specific, actionable subtasks.
For each subtask:
1. Give it a unique ID (T1, T2, T3, etc.)
2. Describe what needs to be done
3. List dependencies (which tasks must complete first)

Format:
T1: [description] | Dependencies: none
T2: [description] | Dependencies: T1
T3: [description] | Dependencies: T1
T4: [description] | Dependencies: T2, T3
...

Be specific and ensure dependencies are correct."""),
            ("human", "Goal: {goal}\n\nTask Decomposition:")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        response = chain.invoke({"goal": goal})
        
        # Parse tasks
        plan = ExecutionPlan(goal=goal)
        
        for line in response.strip().split('\n'):
            line = line.strip()
            if not line or ':' not in line:
                continue
            
            # Parse: T1: description | Dependencies: T2, T3
            parts = line.split('|')
            if len(parts)  ExecutionPlan:
        """
        Analyze dependencies and optimize execution order.
        
        Args:
            plan: Initial execution plan
            
        Returns:
            Plan with optimized execution order
        """
        # Topological sort to get valid execution order
        sorted_tasks = plan.topological_sort()
        
        # Identify parallel groups (tasks at same level)
        completed = set()
        parallel_groups = []
        
        while completed != set(plan.tasks.keys()):
            # Get all tasks that can execute now
            ready_tasks = []
            for task_id in sorted_tasks:
                if task_id not in completed:
                    task = plan.tasks[task_id]
                    if task.can_execute(completed):
                        ready_tasks.append(task_id)
            
            if not ready_tasks:
                break  # Circular dependency or error
            
            parallel_groups.append(ready_tasks)
            completed.update(ready_tasks)
            
            # Assign parallel group numbers
            for task_id in ready_tasks:
                plan.tasks[task_id].parallel_group = len(parallel_groups) - 1
        
        plan.parallel_groups = parallel_groups
        
        return plan
    
    def optimize_plan(self, plan: ExecutionPlan) -> ExecutionPlan:
        """
        Apply optimizations to the execution plan.
        
        Args:
            plan: Analyzed execution plan
            
        Returns:
            Optimized plan
        """
        # Optimization strategies:
        # 1. Limit parallel groups to max_parallel
        # 2. Try to balance load across groups
        # 3. Merge small groups if beneficial
        
        optimized_groups = []
        for group in plan.parallel_groups:
            if len(group)  str:
        """
        Execute a single task.
        
        Args:
            task: The task to execute
            plan: The complete plan for context
            
        Returns:
            Task result
        """
        # Build context from completed dependencies
        context_parts = []
        for dep_id in task.dependencies:
            dep_task = plan.tasks.get(dep_id)
            if dep_task and dep_task.result:
                context_parts.append(f"{dep_id}: {dep_task.result}")
        
        context = "\n".join(context_parts) if context_parts else "No prior context."
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Execute the task using the provided context from previous steps."),
            ("human", """Goal: {goal}

Prior Results:
{context}

Task: {task_description}

Result:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        result = chain.invoke({
            "goal": plan.goal,
            "context": context,
            "task_description": task.description
        })
        
        return result.strip()
    
    def execute_plan(self, plan: ExecutionPlan) -> Dict[str, Any]:
        """
        Execute the optimized plan.
        
        Args:
            plan: The execution plan
            
        Returns:
            Execution results
        """
        print(f"\nGoal: {plan.goal}\n")
        print(f"Total Tasks: {len(plan.tasks)}")
        print(f"Parallel Groups: {len(plan.parallel_groups)}\n")
        
        # Show execution plan
        print("Execution Plan:")
        for i, group in enumerate(plan.parallel_groups):
            group_tasks = [plan.tasks[tid].description[:50] for tid in group]
            print(f"  Group {i+1} ({len(group)} tasks in parallel):")
            for task_id in group:
                deps = list(plan.tasks[task_id].dependencies)
                dep_str = f" [depends on: {', '.join(deps)}]" if deps else ""
                print(f"    - {task_id}: {plan.tasks[task_id].description[:60]}{dep_str}")
        print()
        
        completed = set()
        total_llm_calls = 0
        
        # Execute groups in sequence
        for group_idx, group in enumerate(plan.parallel_groups):
            print(f"{'='*60}")
            print(f"Executing Group {group_idx + 1} ({len(group)} tasks)")
            print('='*60)
            
            # Execute tasks in group in parallel
            with ThreadPoolExecutor(max_workers=min(len(group), self.max_parallel)) as executor:
                futures = {}
                for task_id in group:
                    task = plan.tasks[task_id]
                    task.status = TaskStatus.RUNNING
                    future = executor.submit(self.execute_task, task, plan)
                    futures[future] = task
                
                # Collect results
                for future in as_completed(futures):
                    task = futures[future]
                    total_llm_calls += 1
                    
                    try:
                        result = future.result()
                        task.result = result
                        task.status = TaskStatus.COMPLETED
                        completed.add(task.id)
                        print(f"\n✓ {task.id} completed")
                        print(f"  Result: {result[:80]}...")
                    except Exception as e:
                        task.status = TaskStatus.FAILED
                        task.error = str(e)
                        print(f"\n✗ {task.id} failed: {str(e)}")
        
        # Generate final summary
        summary = self._synthesize_results(plan)
        
        return {
            "goal": plan.goal,
            "total_tasks": len(plan.tasks),
            "completed_tasks": len(completed),
            "parallel_groups": len(plan.parallel_groups),
            "total_llm_calls": total_llm_calls,
            "summary": summary
        }
    
    def _synthesize_results(self, plan: ExecutionPlan) -> str:
        """Synthesize final summary from all task results."""
        results = []
        for task in plan.tasks.values():
            if task.status == TaskStatus.COMPLETED and task.result:
                results.append(f"{task.id}: {task.result}")
        
        if not results:
            return "No tasks completed successfully."
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Synthesize the results into a comprehensive summary."),
            ("human", """Goal: {goal}

Task Results:
{results}

Summary:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        try:
            summary = chain.invoke({
                "goal": plan.goal,
                "results": "\n\n".join(results)
            })
            return summary.strip()
        except:
            return "Summary generation failed."
    
    def compile_and_execute(self, goal: str) -> Dict[str, Any]:
        """
        Compile and execute a goal (full pipeline).
        
        Args:
            goal: The goal to achieve
            
        Returns:
            Execution results
        """
        print(f"\n{'='*80}")
        print("LLM COMPILER: PARSING")
        print('='*80)
        
        # Phase 1: Parse and decompose
        plan = self.parse_task(goal)
        print(f"\nParsed {len(plan.tasks)} tasks")
        
        print(f"\n{'='*80}")
        print("LLM COMPILER: ANALYSIS")
        print('='*80)
        
        # Phase 2: Analyze dependencies
        plan = self.analyze_dependencies(plan)
        print(f"\nIdentified {len(plan.parallel_groups)} execution groups")
        
        print(f"\n{'='*80}")
        print("LLM COMPILER: OPTIMIZATION")
        print('='*80)
        
        # Phase 3: Optimize
        plan = self.optimize_plan(plan)
        print(f"\nOptimized for {self.max_parallel} max parallel tasks")
        
        print(f"\n{'='*80}")
        print("LLM COMPILER: EXECUTION")
        print('='*80)
        
        # Phase 4: Execute
        result = self.execute_plan(plan)
        
        return result


def demonstrate_llm_compiler():
    """Demonstrates the LLM Compiler pattern."""
    
    print("=" * 80)
    print("PATTERN 008: LLM Compiler")
    print("=" * 80)
    print()
    print("LLM Compiler optimizes execution like a code compiler:")
    print("1. Parsing: Decompose task into subtasks")
    print("2. Analysis: Build dependency graph")
    print("3. Optimization: Identify parallelization opportunities")
    print("4. Execution: Run optimized plan with parallel execution")
    print()
    
    # Create compiler
    compiler = LLMCompiler(max_parallel=3)
    
    # Test goals
    goals = [
        "Research the history of artificial intelligence and create a timeline of major milestones",
        "Analyze the advantages and disadvantages of renewable energy sources and compare them"
    ]
    
    for i, goal in enumerate(goals, 1):
        print(f"\n{'='*80}")
        print(f"Example {i}")
        print('='*80)
        
        try:
            result = compiler.compile_and_execute(goal)
            
            print(f"\n\n{'='*80}")
            print("COMPILATION SUMMARY")
            print('='*80)
            print(f"\nGoal: {result['goal']}")
            print(f"Total Tasks: {result['total_tasks']}")
            print(f"Completed: {result['completed_tasks']}/{result['total_tasks']}")
            print(f"Parallel Groups: {result['parallel_groups']}")
            print(f"Total LLM Calls: {result['total_llm_calls']}")
            print(f"\nFinal Summary:\n{result['summary']}")
            
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
            import traceback
            traceback.print_exc()
    
    # Summary
    print("\n\n" + "=" * 80)
    print("LLM COMPILER PATTERN DEMONSTRATION COMPLETE")
    print("=" * 80)
    print()
    print("Key Features Demonstrated:")
    print("1. Task Decomposition: Breaking complex goals into subtasks")
    print("2. Dependency Analysis: Building execution dependency graph")
    print("3. Parallelization: Executing independent tasks concurrently")
    print("4. Optimization: Grouping and scheduling for efficiency")
    print("5. Resource Management: Limiting parallel execution")
    print()
    print("Advantages:")
    print("- Reduced total execution time through parallelization")
    print("- Efficient resource utilization")
    print("- Clear dependency management")
    print("- Systematic optimization approach")
    print()
    print("When to use LLM Compiler:")
    print("- Complex multi-step workflows")
    print("- Tasks with many independent subtasks")
    print("- Time-sensitive applications")
    print("- Research and analysis tasks")
    print()
    print("LangChain Components Used:")
    print("- ChatPromptTemplate for task decomposition")
    print("- Dependency graph (DAG) management")
    print("- ThreadPoolExecutor for parallel execution")
    print("- Result synthesis chain")
    print()


if __name__ == "__main__":
    # Check for API key
    if not os.getenv("OPENAI_API_KEY"):
        print("Error: OPENAI_API_KEY not found in environment variables.")
        print("Please set it in your .env file or environment.")
        exit(1)
    
    demonstrate_llm_compiler()

Agentic Design Pattern: Llm Compiler

Pattern 008: Llm Compiler – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Llm Compiler pattern implementation.

Pattern Description

Pattern 008: LLM Compiler

Description: Optimizes LLM execution by analyzing task dependencies, parallelizing independent operations, and optimizing the execution plan like a compiler optimizes code. Combines planning with execution optimization.

Key Concepts: – Task Decomposition: Break complex tasks into subtasks – Dependency Analysis: Build dependency graph (DAG) – Optimization: Identify parallelizable operations – Scheduling: Optimal execution order – Resource Management: Efficient LLM call allocation

Compiler Phases: 1. Parsing: Understand and decompose the task 2. Analysis: Build dependency graph 3. Optimization: Find parallelization opportunities 4. Code Generation: Create execution plan 5. Execution: Run optimized plan

Use Cases: – Complex multi-step workflows – Research tasks with multiple branches – Data processing pipelines – Tasks with many independent subtasks

LangChain Implementation: Uses dependency graphs and parallel execution with LangChain tools.

File Statistics

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

Module Documentation

"""
Pattern 008: LLM Compiler

Description:
    Optimizes LLM execution by analyzing task dependencies, parallelizing
    independent operations, and optimizing the execution plan like a compiler
    optimizes code. Combines planning with execution optimization.

Key Concepts:
    - Task Decomposition: Break complex tasks into subtasks
    - Dependency Analysis: Build dependency graph (DAG)
    - Optimization: Identify parallelizable operations
    - Scheduling: Optimal execution order
    - Resource Management: Efficient LLM call allocation

Compiler Phases:
    1. Parsing: Understand and decompose the task
    2. Analysis: Build dependency graph
    3. Optimization: Find parallelization opportunities
    4. Code Generation: Create execution plan
    5. Execution: Run optimized plan

Use Cases:
    - Complex multi-step workflows
    - Research tasks with multiple branches
    - Data processing pipelines
    - Tasks with many independent subtasks

LangChain Implementation:
    Uses dependency graphs and parallel execution with LangChain tools.
"""

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

Import Statements

Standard Library Imports

import collections.defaultdict
import collections.deque
import concurrent.futures.ThreadPoolExecutor
import concurrent.futures.as_completed
import dataclasses.dataclass
import dataclasses.field
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
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 – collections.deque: Specialized container datatypes – concurrent.futures.ThreadPoolExecutor: Regular expression operations – concurrent.futures.as_completed: Regular expression operations – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – 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.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
from traceback

Purpose: Additional dependencies for specialized functionality.

Classes

Class: `TaskStatus`

Lines 49-55

Inherits from: Enum

Purpose:

Status of a task in the execution plan.

Class: `Task`

Lines 59-72

Purpose:

Represents a single task in the execution plan.

Methods:

`can_execute(self, completed_tasks)`

Lines 69-72

Purpose:

Check if this task can be executed.

Parameters: – completed_tasks: Completed tasks

Returns: bool

Class: `ExecutionPlan`

Lines 76-114

Purpose:

Represents the optimized execution plan.

Methods:

`add_task(self, task)`

Lines 83-87

Purpose:

Add a task to the plan.

Parameters: – task: Task

`get_ready_tasks(self, completed)`

Lines 89-95

Purpose:

Get all tasks that are ready to execute.

Parameters: – completed: Completed

Returns: <ast.Subscript object at 0x000001A06D615F10>

`topological_sort(self)`

Lines 97-114

Purpose:

Perform topological sort on tasks.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D6AA390>

Class: `LLMCompiler`

Lines 117-464

Purpose:

Compiler that optimizes LLM task execution.

Methods:

`init(self, model_name, max_parallel)`

Lines 120-133

Purpose:

Initialize the LLM Compiler.

Args:
    model_name: Name of the OpenAI model
    max_parallel: Maximum number of parallel tasks

Parameters: – model_name: Language model instance – max_parallel: Max parallel

`parse_task(self, goal)`

Lines 135-209

Purpose:

Parse the goal and decompose into subtasks.

Args:
    goal: The high-level goal

Returns:
    Initial execution plan with tasks

Parameters: – goal: Goal

Returns: ExecutionPlan

`analyze_dependencies(self, plan)`

Lines 211-249

Purpose:

Analyze dependencies and optimize execution order.

Args:
    plan: Initial execution plan

Returns:
    Plan with optimized execution order

Parameters: – plan: Plan

Returns: ExecutionPlan

`optimize_plan(self, plan)`

Lines 251-277

Purpose:

Apply optimizations to the execution plan.

Args:
    plan: Analyzed execution plan

Returns:
    Optimized plan

Parameters: – plan: Plan

Returns: ExecutionPlan

`execute_task(self, task, plan)`

Lines 279-318

Purpose:

Execute a single task.

Args:
    task: The task to execute
    plan: The complete plan for context

Returns:
    Task result

Parameters: – task: Task – plan: Plan

Returns: str

`execute_plan(self, plan)`

Lines 320-390

Purpose:

Execute the optimized plan.

Args:
    plan: The execution plan

Returns:
    Execution results

Parameters: – plan: Plan

Returns: <ast.Subscript object at 0x000001A06D6EB350>

`_synthesize_results(self, plan)`

Lines 392-421

Purpose:

Synthesize final summary from all task results.

Parameters: – plan: Plan

Returns: str

`compile_and_execute(self, goal)`

Lines 423-464

Purpose:

Compile and execute a goal (full pipeline).

Args:
    goal: The goal to achieve

Returns:
    Execution results

Parameters: – goal: Goal

Returns: <ast.Subscript object at 0x000001A06D628190>

Functions

`demonstrate_llm_compiler()`

Lines 467-542

Purpose:

Demonstrates the LLM Compiler pattern.

Key Concepts Summary

Pattern Overview

The Llm Compiler 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 007
Pattern 009
Other Reasoning Patterns (001-011)

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 Llm Compiler 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: 008_llm_compiler.py
Total Lines: 553
Pattern Category: AI Agent Design Pattern #008

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Agentic Design Pattern: Rewoo


    """
Pattern 007: ReWOO (Reasoning Without Observation)

Description:
    Decouples reasoning from observation by planning all tool uses upfront,
    then executing them in batch. Unlike ReAct which interleaves reasoning
    and observation, ReWOO plans all actions first, then executes them,
    reducing LLM calls and enabling parallel execution.

Key Concepts:
    - Upfront Planning: Generate complete tool usage plan first
    - Dependency Analysis: Identify dependencies between tool calls
    - Batch Execution: Execute all tools (potentially in parallel)
    - Variable System: Use variables (e.g., #E1, #E2) for results
    - Final Synthesis: Generate answer using all tool results

Benefits:
    - Fewer LLM calls (plan once, execute all)
    - Enables parallel tool execution
    - More efficient for multi-tool tasks
    - Clearer execution plan

Use Cases:
    - Multi-tool research tasks
    - Information gathering from multiple sources
    - Tasks requiring coordinated tool use
    - Scenarios where parallelization is beneficial

LangChain Implementation:
    Separate planning and execution chains with variable substitution.
"""

import os
import re
from typing import List, Dict, Any, Optional, Callable
from dataclasses import dataclass, field
from concurrent.futures import ThreadPoolExecutor, as_completed
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.tools import Tool
from langchain_openai import ChatOpenAI

# Load environment variables
load_dotenv()


@dataclass
class ToolCall:
    """Represents a planned tool call."""
    variable: str  # e.g., "#E1"
    tool_name: str
    arguments: str
    dependencies: List[str] = field(default_factory=list)  # Variable dependencies
    result: Optional[str] = None
    executed: bool = False


@dataclass
class ExecutionPlan:
    """Represents the complete tool execution plan."""
    question: str
    tool_calls: List[ToolCall] = field(default_factory=list)
    final_reasoning: str = ""
    
    def get_executable_calls(self) -> List[ToolCall]:
        """Get tool calls that can be executed (dependencies met)."""
        executable = []
        for call in self.tool_calls:
            if not call.executed:
                # Check if all dependencies are met
                deps_met = all(
                    any(tc.variable == dep and tc.executed for tc in self.tool_calls)
                    for dep in call.dependencies
                )
                if not call.dependencies or deps_met:
                    executable.append(call)
        return executable


class ReWOOAgent:
    """Agent that uses ReWOO (Reasoning Without Observation) pattern."""
    
    def __init__(self, tools: List[Tool], model_name: str = "gpt-3.5-turbo"):
        """
        Initialize the ReWOO agent.
        
        Args:
            tools: List of available tools
            model_name: Name of the OpenAI model
        """
        self.tools = {tool.name: tool for tool in tools}
        self.tool_descriptions = self._format_tool_descriptions()
        
        self.planner_llm = ChatOpenAI(
            model=model_name,
            temperature=0.0,  # Low temperature for consistent planning
            api_key=os.getenv("OPENAI_API_KEY")
        )
        
        self.solver_llm = ChatOpenAI(
            model=model_name,
            temperature=0.3,
            api_key=os.getenv("OPENAI_API_KEY")
        )
    
    def _format_tool_descriptions(self) -> str:
        """Format tool descriptions for the planner."""
        descriptions = []
        for name, tool in self.tools.items():
            descriptions.append(f"- {name}: {tool.description}")
        return "\n".join(descriptions)
    
    def plan(self, question: str) -> ExecutionPlan:
        """
        Create an execution plan for answering the question.
        
        Args:
            question: The question to answer
            
        Returns:
            ExecutionPlan with all tool calls planned
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a planner that creates execution plans using available tools.

Available tools:
{tools}

Create a plan to answer the question using these tools. For each step:
1. Assign a variable (#E1, #E2, etc.)
2. Specify the tool and its arguments
3. Reference previous results using their variables

Format each plan step as:
#E1 = ToolName[arguments]
#E2 = ToolName[arguments using #E1 if needed]
...

Then provide reasoning for how to combine the results:
Plan: [Brief description of how to use the results]

Example:
#E1 = Calculator[15 * 7]
#E2 = Search[capital of #E1]
Plan: Use #E1 to get the product, then search for information about #E2."""),
            ("human", "Question: {question}\n\nExecution Plan:")
        ])
        
        chain = prompt | self.planner_llm | StrOutputParser()
        response = chain.invoke({
            "question": question,
            "tools": self.tool_descriptions
        })
        
        # Parse the plan
        plan = ExecutionPlan(question=question)
        
        lines = response.strip().split('\n')
        for line in lines:
            line = line.strip()
            
            # Parse tool calls like: #E1 = ToolName[arguments]
            match = re.match(r'#(E\d+)\s*=\s*(\w+)\[(.*?)\]', line)
            if match:
                variable = f"#{match.group(1)}"
                tool_name = match.group(2)
                arguments = match.group(3).strip()
                
                # Extract dependencies (variables referenced in arguments)
                dependencies = re.findall(r'#E\d+', arguments)
                
                tool_call = ToolCall(
                    variable=variable,
                    tool_name=tool_name,
                    arguments=arguments,
                    dependencies=dependencies
                )
                plan.tool_calls.append(tool_call)
            
            # Extract final reasoning
            elif line.lower().startswith('plan:'):
                plan.final_reasoning = line.split(':', 1)[1].strip()
        
        return plan
    
    def execute_plan(self, plan: ExecutionPlan, parallel: bool = False) -> Dict[str, Any]:
        """
        Execute the tool calls in the plan.
        
        Args:
            plan: The execution plan
            parallel: Whether to execute independent calls in parallel
            
        Returns:
            Dictionary with execution results
        """
        print(f"\nQuestion: {plan.question}\n")
        print("Execution Plan:")
        for call in plan.tool_calls:
            deps = f" (depends on: {', '.join(call.dependencies)})" if call.dependencies else ""
            print(f"  {call.variable} = {call.tool_name}[{call.arguments}]{deps}")
        print(f"\nFinal Plan: {plan.final_reasoning}\n")
        
        print("="*60)
        print("EXECUTING TOOLS")
        print("="*60)
        
        # Execute tool calls respecting dependencies
        while True:
            executable = plan.get_executable_calls()
            if not executable:
                break
            
            if parallel and len(executable) > 1:
                # Execute in parallel
                with ThreadPoolExecutor(max_workers=len(executable)) as executor:
                    futures = {
                        executor.submit(self._execute_tool_call, call, plan): call 
                        for call in executable
                    }
                    
                    for future in as_completed(futures):
                        call = futures[future]
                        try:
                            result = future.result()
                            call.result = result
                            call.executed = True
                            print(f"\n✓ {call.variable} = {result[:100]}...")
                        except Exception as e:
                            call.result = f"Error: {str(e)}"
                            call.executed = True
                            print(f"\n✗ {call.variable} failed: {str(e)}")
            else:
                # Execute sequentially
                for call in executable:
                    try:
                        result = self._execute_tool_call(call, plan)
                        call.result = result
                        call.executed = True
                        print(f"\n✓ {call.variable} = {result[:100]}...")
                    except Exception as e:
                        call.result = f"Error: {str(e)}"
                        call.executed = True
                        print(f"\n✗ {call.variable} failed: {str(e)}")
        
        # Synthesize final answer
        print(f"\n{'='*60}")
        print("SYNTHESIZING ANSWER")
        print('='*60)
        
        answer = self._synthesize_answer(plan)
        
        return {
            "question": plan.question,
            "answer": answer,
            "tool_calls": len(plan.tool_calls),
            "plan": plan.final_reasoning
        }
    
    def _execute_tool_call(self, call: ToolCall, plan: ExecutionPlan) -> str:
        """Execute a single tool call."""
        # Substitute variable references with actual results
        arguments = call.arguments
        for dep_var in call.dependencies:
            # Find the result for this dependency
            for other_call in plan.tool_calls:
                if other_call.variable == dep_var and other_call.result:
                    arguments = arguments.replace(dep_var, other_call.result)
        
        # Get the tool
        tool = self.tools.get(call.tool_name)
        if not tool:
            raise ValueError(f"Tool {call.tool_name} not found")
        
        # Execute the tool
        try:
            result = tool.func(arguments)
            return str(result)
        except Exception as e:
            raise Exception(f"Tool execution failed: {str(e)}")
    
    def _synthesize_answer(self, plan: ExecutionPlan) -> str:
        """Synthesize final answer from tool results."""
        # Build evidence from tool results
        evidence = []
        for call in plan.tool_calls:
            if call.executed and call.result:
                evidence.append(f"{call.variable}: {call.result}")
        
        evidence_text = "\n".join(evidence)
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Based on the evidence gathered from tool execution, 
provide a comprehensive answer to the original question.

Use the evidence to construct a well-reasoned response."""),
            ("human", """Question: {question}

Evidence:
{evidence}

Plan: {plan}

Answer:""")
        ])
        
        chain = prompt | self.solver_llm | StrOutputParser()
        answer = chain.invoke({
            "question": plan.question,
            "evidence": evidence_text,
            "plan": plan.final_reasoning
        })
        
        return answer.strip()
    
    def solve(self, question: str, parallel: bool = False) -> Dict[str, Any]:
        """
        Solve a question using ReWOO pattern.
        
        Args:
            question: The question to answer
            parallel: Whether to use parallel execution
            
        Returns:
            Dictionary with answer and metadata
        """
        print(f"\n{'='*80}")
        print("ReWOO: PLANNING PHASE")
        print('='*80)
        
        # Create plan
        plan = self.plan(question)
        
        print(f"\n{'='*80}")
        print("ReWOO: EXECUTION PHASE")
        print('='*80)
        
        # Execute plan
        result = self.execute_plan(plan, parallel=parallel)
        
        return result


# Example tools
def calculator(expression: str) -> float:
    """Calculate a mathematical expression."""
    try:
        # Simple eval (in production, use a safe math parser)
        result = eval(expression)
        return result
    except Exception as e:
        return f"Calculation error: {str(e)}"


def search(query: str) -> str:
    """Simulate a search tool."""
    # In production, this would call a real search API
    responses = {
        "capital of france": "Paris is the capital of France.",
        "population of tokyo": "Tokyo has a population of approximately 14 million people.",
        "speed of light": "The speed of light is approximately 299,792,458 meters per second.",
        "inventor of telephone": "Alexander Graham Bell is credited with inventing the telephone.",
    }
    
    query_lower = query.lower()
    for key, value in responses.items():
        if key in query_lower:
            return value
    
    return f"Search results for: {query}"


def get_date(query: str) -> str:
    """Get current date information."""
    from datetime import datetime
    now = datetime.now()
    
    if "year" in query.lower():
        return str(now.year)
    elif "month" in query.lower():
        return now.strftime("%B")
    elif "day" in query.lower():
        return str(now.day)
    else:
        return now.strftime("%Y-%m-%d")


def demonstrate_rewoo():
    """Demonstrates the ReWOO pattern."""
    
    print("=" * 80)
    print("PATTERN 007: ReWOO (Reasoning Without Observation)")
    print("=" * 80)
    print()
    print("ReWOO decouples reasoning from observation:")
    print("1. Planning Phase: Plan all tool uses upfront")
    print("2. Dependency Analysis: Identify dependencies between calls")
    print("3. Execution Phase: Execute tools (potentially in parallel)")
    print("4. Synthesis: Generate answer from all results")
    print()
    
    # Create tools
    tools = [
        Tool(
            name="Calculator",
            func=calculator,
            description="Calculate mathematical expressions. Input should be a valid expression."
        ),
        Tool(
            name="Search",
            func=search,
            description="Search for information. Input should be a search query."
        ),
        Tool(
            name="GetDate",
            func=get_date,
            description="Get current date information. Input can be 'year', 'month', 'day', or 'full'."
        )
    ]
    
    # Create agent
    agent = ReWOOAgent(tools=tools)
    
    # Test questions
    questions = [
        "What is 25 * 34, and what is the capital of France?",
        "Calculate 144 / 12, then search for information about that number",
    ]
    
    for i, question in enumerate(questions, 1):
        print(f"\n{'='*80}")
        print(f"Example {i}")
        print('='*80)
        
        try:
            result = agent.solve(question, parallel=False)
            
            print(f"\n\n{'='*80}")
            print("FINAL RESULT")
            print('='*80)
            print(f"\nQuestion: {result['question']}")
            print(f"\nAnswer: {result['answer']}")
            print(f"\nTool Calls: {result['tool_calls']}")
            
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
            import traceback
            traceback.print_exc()
    
    # Summary
    print("\n\n" + "=" * 80)
    print("ReWOO PATTERN DEMONSTRATION COMPLETE")
    print("=" * 80)
    print()
    print("Key Features Demonstrated:")
    print("1. Upfront Planning: All tool uses planned before execution")
    print("2. Variable System: Results referenced as #E1, #E2, etc.")
    print("3. Dependency Tracking: Execution order based on dependencies")
    print("4. Batch Execution: All tools executed in one phase")
    print("5. Result Synthesis: Final answer combines all evidence")
    print()
    print("Advantages over ReAct:")
    print("- Fewer LLM calls (plan once vs. after each observation)")
    print("- Enables parallel tool execution")
    print("- Clearer execution flow")
    print("- More efficient for multi-tool tasks")
    print()
    print("When to use ReWOO:")
    print("- Multi-tool information gathering")
    print("- Tasks with independent tool calls")
    print("- Scenarios benefiting from parallelization")
    print("- When you want predictable tool usage")
    print()
    print("LangChain Components Used:")
    print("- Tool abstraction for uniform tool interface")
    print("- Separate planning and execution chains")
    print("- Variable substitution system")
    print("- Optional parallel execution")
    print()


if __name__ == "__main__":
    # Check for API key
    if not os.getenv("OPENAI_API_KEY"):
        print("Error: OPENAI_API_KEY not found in environment variables.")
        print("Please set it in your .env file or environment.")
        exit(1)
    
    demonstrate_rewoo()

Agentic Design Pattern: Rewoo

Pattern 007: Rewoo – Line-by-Line Explanation

Overview

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

Pattern Description

Pattern 007: ReWOO (Reasoning Without Observation)

Description: Decouples reasoning from observation by planning all tool uses upfront, then executing them in batch. Unlike ReAct which interleaves reasoning and observation, ReWOO plans all actions first, then executes them, reducing LLM calls and enabling parallel execution.

Key Concepts: – Upfront Planning: Generate complete tool usage plan first – Dependency Analysis: Identify dependencies between tool calls – Batch Execution: Execute all tools (potentially in parallel) – Variable System: Use variables (e.g., #E1, #E2) for results – Final Synthesis: Generate answer using all tool results

Benefits: – Fewer LLM calls (plan once, execute all) – Enables parallel tool execution – More efficient for multi-tool tasks – Clearer execution plan

Use Cases: – Multi-tool research tasks – Information gathering from multiple sources – Tasks requiring coordinated tool use – Scenarios where parallelization is beneficial

LangChain Implementation: Separate planning and execution chains with variable substitution.

File Statistics

Total Lines: 492
Classes: 3
Functions: 4
Constants: 0

Module Documentation

"""
Pattern 007: ReWOO (Reasoning Without Observation)

Description:
    Decouples reasoning from observation by planning all tool uses upfront,
    then executing them in batch. Unlike ReAct which interleaves reasoning
    and observation, ReWOO plans all actions first, then executes them,
    reducing LLM calls and enabling parallel execution.

Key Concepts:
    - Upfront Planning: Generate complete tool usage plan first
    - Dependency Analysis: Identify dependencies between tool calls
    - Batch Execution: Execute all tools (potentially in parallel)
    - Variable System: Use variables (e.g., #E1, #E2) for results
    - Final Synthesis: Generate answer using all tool results

Benefits:
    - Fewer LLM calls (plan once, execute all)
    - Enables parallel tool execution
    - More efficient for multi-tool tasks
    - Clearer execution plan

Use Cases:
    - Multi-tool research tasks
    - Information gathering from multiple sources
    - Tasks requiring coordinated tool use
    - Scenarios where parallelization is beneficial

LangChain Implementation:
    Separate planning and execution chains with variable substitution.
"""

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

Import Statements

Standard Library Imports

import concurrent.futures.ThreadPoolExecutor
import concurrent.futures.as_completed
import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import langchain_core.tools.Tool
import os
import re
import typing.Any
import typing.Callable
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – concurrent.futures.ThreadPoolExecutor: Regular expression operations – concurrent.futures.as_completed: Regular expression operations – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: 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 – langchain_core.tools.Tool: Regular expression operations – os: Environment variable access and system operations – re: Regular expression 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
from traceback

Purpose: Additional dependencies for specialized functionality.

Classes

Class: `ToolCall`

Lines 50-57

Purpose:

Represents a planned tool call.

Class: `ExecutionPlan`

Lines 61-79

Purpose:

Represents the complete tool execution plan.

Methods:

`get_executable_calls(self)`

Lines 67-79

Purpose:

Get tool calls that can be executed (dependencies met).

Parameters:

Returns: <ast.Subscript object at 0x000001A06D6761D0>

Class: `ReWOOAgent`

Lines 82-343

Purpose:

Agent that uses ReWOO (Reasoning Without Observation) pattern.

Methods:

`init(self, tools, model_name)`

Lines 85-106

Purpose:

Initialize the ReWOO agent.

Args:
    tools: List of available tools
    model_name: Name of the OpenAI model

Parameters: – tools: Available tools for the agent – model_name: Language model instance

`_format_tool_descriptions(self)`

Lines 108-113

Purpose:

Format tool descriptions for the planner.

Parameters:

Returns: str

`plan(self, question)`

Lines 115-186

Purpose:

Create an execution plan for answering the question.

Args:
    question: The question to answer

Returns:
    ExecutionPlan with all tool calls planned

Parameters: – question: Question to answer

Returns: ExecutionPlan

`execute_plan(self, plan, parallel)`

Lines 188-260

Purpose:

Execute the tool calls in the plan.

Args:
    plan: The execution plan
    parallel: Whether to execute independent calls in parallel

Returns:
    Dictionary with execution results

Parameters: – plan: Plan – parallel: Parallel

Returns: <ast.Subscript object at 0x000001A06D65A490>

`_execute_tool_call(self, call, plan)`

Lines 262-282

Purpose:

Execute a single tool call.

Parameters: – call: Call – plan: Plan

Returns: str

`_synthesize_answer(self, plan)`

Lines 284-316

Purpose:

Synthesize final answer from tool results.

Parameters: – plan: Plan

Returns: str

`solve(self, question, parallel)`

Lines 318-343

Purpose:

Solve a question using ReWOO pattern.

Args:
    question: The question to answer
    parallel: Whether to use parallel execution

Returns:
    Dictionary with answer and metadata

Parameters: – question: Question to answer – parallel: Parallel

Returns: <ast.Subscript object at 0x000001A06D178FD0>

Functions

`calculator(expression)`

Lines 347-354

Purpose:

Calculate a mathematical expression.

Parameters: – expression: Expression

Returns: float

`search(query)`

Lines 357-372

Purpose:

Simulate a search tool.

Parameters: – query: Input query or question

Returns: str

`get_date(query)`

Lines 375-387

Purpose:

Get current date information.

Parameters: – query: Input query or question

Returns: str

`demonstrate_rewoo()`

Lines 390-481

Purpose:

Demonstrates the ReWOO pattern.

Key Concepts Summary

Pattern Overview

The Rewoo 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 006
Pattern 008
Other Reasoning Patterns (001-011)

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 Rewoo 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: 007_rewoo.py
Total Lines: 492
Pattern Category: AI Agent Design Pattern #007

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Agentic Design Pattern: Plan And Execute


    """
Pattern 006: Plan-and-Execute

Description:
    Separates planning and execution phases for complex multi-step tasks.
    First creates a high-level plan, then executes each step sequentially,
    with optional re-planning based on intermediate results.

Key Concepts:
    - Planning Phase: Create structured plan with steps
    - Execution Phase: Execute each step in sequence
    - Monitoring: Track progress and outcomes
    - Re-planning: Adjust plan based on execution results
    - State Management: Maintain context across steps

Use Cases:
    - Complex multi-step tasks requiring coordination
    - Research and information gathering workflows
    - Content creation with multiple stages
    - Problem-solving with dependencies between steps

LangChain Implementation:
    Uses LangGraph for state management and sequential execution with
    separate planner and executor chains.
"""

import os
from typing import List, Dict, Any, Optional
from enum import Enum
from dataclasses import dataclass, field
from dotenv import load_dotenv

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

# Load environment variables
load_dotenv()


class StepStatus(Enum):
    """Status of an execution step."""
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    SKIPPED = "skipped"


@dataclass
class PlanStep:
    """Represents a single step in the execution plan."""
    id: int
    description: str
    dependencies: List[int] = field(default_factory=list)
    status: StepStatus = StepStatus.PENDING
    result: Optional[str] = None
    error: Optional[str] = None


@dataclass
class ExecutionPlan:
    """Represents the complete execution plan."""
    goal: str
    steps: List[PlanStep] = field(default_factory=list)
    current_step: int = 0
    completed_steps: int = 0
    
    def get_next_step(self) -> Optional[PlanStep]:
        """Get the next pending step."""
        for step in self.steps:
            if step.status == StepStatus.PENDING:
                # Check if dependencies are met
                if all(
                    self.steps[dep_id - 1].status == StepStatus.COMPLETED 
                    for dep_id in step.dependencies
                ):
                    return step
        return None
    
    def is_complete(self) -> bool:
        """Check if all steps are completed or skipped."""
        return all(
            step.status in [StepStatus.COMPLETED, StepStatus.SKIPPED]
            for step in self.steps
        )
    
    def get_progress(self) -> float:
        """Get completion progress as percentage."""
        if not self.steps:
            return 0.0
        completed = sum(
            1 for step in self.steps 
            if step.status in [StepStatus.COMPLETED, StepStatus.SKIPPED]
        )
        return completed / len(self.steps)


class PlanAndExecuteAgent:
    """Agent that uses plan-and-execute pattern."""
    
    def __init__(self, model_name: str = "gpt-3.5-turbo", 
                 enable_replanning: bool = True):
        """
        Initialize the Plan-and-Execute agent.
        
        Args:
            model_name: Name of the OpenAI model
            enable_replanning: Whether to enable dynamic re-planning
        """
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=0.7,
            api_key=os.getenv("OPENAI_API_KEY")
        )
        self.enable_replanning = enable_replanning
        self.execution_history: List[Dict[str, Any]] = []
    
    def create_plan(self, goal: str, context: str = "") -> ExecutionPlan:
        """
        Create an execution plan for the given goal.
        
        Args:
            goal: The goal to achieve
            context: Additional context information
            
        Returns:
            ExecutionPlan with structured steps
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Create a detailed step-by-step plan to achieve the goal.
Break down the task into clear, actionable steps.
Format your response as:
1. [First step]
2. [Second step]
3. [Third step]
etc.

Each step should be specific and achievable."""),
            ("human", """Goal: {goal}

Context: {context}

Plan:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        response = chain.invoke({
            "goal": goal,
            "context": context if context else "No additional context."
        })
        
        # Parse steps
        plan = ExecutionPlan(goal=goal)
        step_id = 1
        
        for line in response.strip().split('\n'):
            line = line.strip()
            if line and (line[0].isdigit() or line.startswith('-')):
                # Extract step description
                description = line.split('.', 1)[-1].strip().lstrip('- ').strip()
                if description:
                    step = PlanStep(
                        id=step_id,
                        description=description
                    )
                    plan.steps.append(step)
                    step_id += 1
        
        return plan
    
    def execute_step(self, step: PlanStep, plan: ExecutionPlan) -> str:
        """
        Execute a single step of the plan.
        
        Args:
            step: The step to execute
            plan: The complete plan for context
            
        Returns:
            Result of executing the step
        """
        # Gather context from previous steps
        context = self._build_context(plan)
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Execute the given step and provide a detailed result.
Be specific and thorough in your execution.
If the step requires information from previous steps, use the provided context."""),
            ("human", """Goal: {goal}

Previous steps context:
{context}

Current step to execute: {step_description}

Execution result:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        try:
            result = chain.invoke({
                "goal": plan.goal,
                "context": context,
                "step_description": step.description
            })
            
            return result.strip()
            
        except Exception as e:
            raise Exception(f"Step execution failed: {str(e)}")
    
    def _build_context(self, plan: ExecutionPlan) -> str:
        """Build context string from completed steps."""
        context_parts = []
        for step in plan.steps:
            if step.status == StepStatus.COMPLETED and step.result:
                context_parts.append(f"Step {step.id}: {step.description}\nResult: {step.result}")
        
        return "\n\n".join(context_parts) if context_parts else "No previous steps completed."
    
    def should_replan(self, plan: ExecutionPlan, last_step: PlanStep) -> bool:
        """
        Determine if re-planning is needed based on execution results.
        
        Args:
            plan: Current execution plan
            last_step: Most recently executed step
            
        Returns:
            True if re-planning is recommended
        """
        if not self.enable_replanning:
            return False
        
        # Simple heuristic: replan if step failed or result suggests issues
        if last_step.status == StepStatus.FAILED:
            return True
        
        # Use LLM to evaluate if re-planning is needed
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Evaluate if the plan should be adjusted based on the execution result.
Consider if:
- The result reveals new information that changes the approach
- There are unexpected obstacles
- The remaining steps need to be modified

Respond with only 'YES' or 'NO'."""),
            ("human", """Goal: {goal}

Completed step: {step_description}
Result: {result}

Should we replan? (YES/NO):""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        try:
            response = chain.invoke({
                "goal": plan.goal,
                "step_description": last_step.description,
                "result": last_step.result
            })
            
            return "yes" in response.lower()
            
        except:
            return False
    
    def replan(self, plan: ExecutionPlan) -> ExecutionPlan:
        """
        Create a new plan based on current progress.
        
        Args:
            plan: Current execution plan
            
        Returns:
            Updated execution plan
        """
        context = self._build_context(plan)
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Given the progress so far, create an updated plan for the remaining work.
Consider what has been accomplished and what still needs to be done.

Format as:
1. [First remaining step]
2. [Second remaining step]
etc."""),
            ("human", """Goal: {goal}

Progress so far:
{context}

Updated plan for remaining work:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        response = chain.invoke({
            "goal": plan.goal,
            "context": context
        })
        
        # Create new plan with completed steps + new steps
        new_plan = ExecutionPlan(goal=plan.goal)
        
        # Keep completed steps
        for step in plan.steps:
            if step.status == StepStatus.COMPLETED:
                new_plan.steps.append(step)
        
        # Add new steps
        step_id = len(new_plan.steps) + 1
        for line in response.strip().split('\n'):
            line = line.strip()
            if line and (line[0].isdigit() or line.startswith('-')):
                description = line.split('.', 1)[-1].strip().lstrip('- ').strip()
                if description:
                    new_step = PlanStep(
                        id=step_id,
                        description=description
                    )
                    new_plan.steps.append(new_step)
                    step_id += 1
        
        return new_plan
    
    def execute_plan(self, plan: ExecutionPlan, max_steps: Optional[int] = None) -> Dict[str, Any]:
        """
        Execute the complete plan.
        
        Args:
            plan: The execution plan
            max_steps: Maximum number of steps to execute (None for all)
            
        Returns:
            Dictionary with execution results
        """
        print(f"\nGoal: {plan.goal}")
        print(f"\nPlan ({len(plan.steps)} steps):")
        for step in plan.steps:
            print(f"  {step.id}. {step.description}")
        print()
        
        executed_steps = 0
        
        while not plan.is_complete():
            if max_steps and executed_steps >= max_steps:
                print(f"\nReached maximum steps limit ({max_steps})")
                break
            
            # Get next step
            next_step = plan.get_next_step()
            if not next_step:
                print("\nNo more executable steps (dependencies not met)")
                break
            
            # Execute step
            print(f"\n{'='*60}")
            print(f"Executing Step {next_step.id}: {next_step.description}")
            print('='*60)
            
            next_step.status = StepStatus.IN_PROGRESS
            
            try:
                result = self.execute_step(next_step, plan)
                next_step.result = result
                next_step.status = StepStatus.COMPLETED
                plan.completed_steps += 1
                
                print(f"\n✓ Step {next_step.id} completed")
                print(f"Result: {result[:200]}..." if len(result) > 200 else f"Result: {result}")
                
                # Check if re-planning is needed
                if self.should_replan(plan, next_step):
                    print(f"\n⟳ Re-planning needed based on results...")
                    plan = self.replan(plan)
                    print(f"Updated plan now has {len(plan.steps)} steps")
                
            except Exception as e:
                next_step.status = StepStatus.FAILED
                next_step.error = str(e)
                print(f"\n✗ Step {next_step.id} failed: {str(e)}")
            
            executed_steps += 1
        
        # Generate final summary
        summary = self._generate_summary(plan)
        
        return {
            "goal": plan.goal,
            "total_steps": len(plan.steps),
            "completed_steps": plan.completed_steps,
            "progress": plan.get_progress(),
            "status": "completed" if plan.is_complete() else "incomplete",
            "summary": summary
        }
    
    def _generate_summary(self, plan: ExecutionPlan) -> str:
        """Generate a summary of the execution."""
        completed_results = []
        for step in plan.steps:
            if step.status == StepStatus.COMPLETED and step.result:
                completed_results.append(f"Step {step.id}: {step.result}")
        
        if not completed_results:
            return "No steps completed successfully."
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Provide a concise summary of what was accomplished."),
            ("human", """Goal: {goal}

Results:
{results}

Summary:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        try:
            summary = chain.invoke({
                "goal": plan.goal,
                "results": "\n\n".join(completed_results)
            })
            return summary.strip()
        except:
            return "Summary generation failed."
    
    def solve_with_plan_execute(self, goal: str, context: str = "") -> Dict[str, Any]:
        """
        Solve a task using plan-and-execute pattern.
        
        Args:
            goal: The goal to achieve
            context: Additional context
            
        Returns:
            Execution results
        """
        print(f"\n{'='*80}")
        print("PLAN-AND-EXECUTE: Creating plan...")
        print('='*80)
        
        # Create initial plan
        plan = self.create_plan(goal, context)
        
        # Execute the plan
        result = self.execute_plan(plan)
        
        return result


def demonstrate_plan_execute():
    """Demonstrates the Plan-and-Execute pattern."""
    
    print("=" * 80)
    print("PATTERN 006: Plan-and-Execute")
    print("=" * 80)
    print()
    print("Plan-and-Execute separates planning from execution:")
    print("1. Planning Phase: Create structured multi-step plan")
    print("2. Execution Phase: Execute each step sequentially")
    print("3. Monitoring: Track progress and results")
    print("4. Re-planning: Adjust plan based on intermediate results")
    print()
    
    # Create agent
    agent = PlanAndExecuteAgent(enable_replanning=True)
    
    # Test goals
    goals = [
        {
            "goal": "Research and write a brief report on renewable energy trends",
            "context": "Focus on solar and wind energy developments in the last 5 years"
        },
        {
            "goal": "Plan a team building event for 20 people",
            "context": "Budget: $2000, Duration: Half day, Location: Urban area"
        }
    ]
    
    for i, test_case in enumerate(goals, 1):
        print(f"\n{'='*80}")
        print(f"Example {i}")
        print('='*80)
        
        try:
            result = agent.solve_with_plan_execute(
                goal=test_case['goal'],
                context=test_case['context']
            )
            
            print(f"\n\n{'='*80}")
            print("EXECUTION SUMMARY")
            print('='*80)
            print(f"\nGoal: {result['goal']}")
            print(f"Status: {result['status'].upper()}")
            print(f"Progress: {result['progress']:.0%}")
            print(f"Steps Completed: {result['completed_steps']}/{result['total_steps']}")
            print(f"\nSummary:\n{result['summary']}")
            
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
            import traceback
            traceback.print_exc()
    
    # Summary
    print("\n\n" + "=" * 80)
    print("PLAN-AND-EXECUTE PATTERN DEMONSTRATION COMPLETE")
    print("=" * 80)
    print()
    print("Key Features Demonstrated:")
    print("1. Structured Planning: Breaking goals into actionable steps")
    print("2. Sequential Execution: Executing steps with dependency tracking")
    print("3. Context Management: Passing results between steps")
    print("4. Progress Monitoring: Tracking completion status")
    print("5. Dynamic Re-planning: Adjusting plan based on results")
    print()
    print("Advantages:")
    print("- Clear separation of planning and execution")
    print("- Better handling of complex multi-step tasks")
    print("- Adaptability through re-planning")
    print("- Progress tracking and monitoring")
    print()
    print("LangChain Components Used:")
    print("- ChatPromptTemplate: Structures planning and execution prompts")
    print("- StrOutputParser: Parses LLM outputs")
    print("- State Management: Tracks plan and execution progress")
    print("- Sequential Chains: Executes steps in order")
    print()


if __name__ == "__main__":
    # Check for API key
    if not os.getenv("OPENAI_API_KEY"):
        print("Error: OPENAI_API_KEY not found in environment variables.")
        print("Please set it in your .env file or environment.")
        exit(1)
    
    demonstrate_plan_execute()

Agentic Design Pattern: Plan And Execute

Pattern 006: Plan And Execute – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Plan And Execute pattern implementation.

Pattern Description

Pattern 006: Plan-and-Execute

Description: Separates planning and execution phases for complex multi-step tasks. First creates a high-level plan, then executes each step sequentially, with optional re-planning based on intermediate results.

Key Concepts: – Planning Phase: Create structured plan with steps – Execution Phase: Execute each step in sequence – Monitoring: Track progress and outcomes – Re-planning: Adjust plan based on execution results – State Management: Maintain context across steps

Use Cases: – Complex multi-step tasks requiring coordination – Research and information gathering workflows – Content creation with multiple stages – Problem-solving with dependencies between steps

LangChain Implementation: Uses LangGraph for state management and sequential execution with separate planner and executor chains.

File Statistics

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

Module Documentation

"""
Pattern 006: Plan-and-Execute

Description:
    Separates planning and execution phases for complex multi-step tasks.
    First creates a high-level plan, then executes each step sequentially,
    with optional re-planning based on intermediate results.

Key Concepts:
    - Planning Phase: Create structured plan with steps
    - Execution Phase: Execute each step in sequence
    - Monitoring: Track progress and outcomes
    - Re-planning: Adjust plan based on execution results
    - State Management: Maintain context across steps

Use Cases:
    - Complex multi-step tasks requiring coordination
    - Research and information gathering workflows
    - Content creation with multiple stages
    - Problem-solving with dependencies between steps

LangChain Implementation:
    Uses LangGraph for state management and sequential execution with
    separate planner and executor chains.
"""

Purpose: This module implements the Plan And Execute pattern, providing a structured approach to agent design and implementation.

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.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
from traceback

Purpose: Additional dependencies for specialized functionality.

Classes

Class: `StepStatus`

Lines 41-47

Inherits from: Enum

Purpose:

Status of an execution step.

Class: `PlanStep`

Lines 51-58

Purpose:

Represents a single step in the execution plan.

Class: `ExecutionPlan`

Lines 62-96

Purpose:

Represents the complete execution plan.

Methods:

`get_next_step(self)`

Lines 69-79

Purpose:

Get the next pending step.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D680D10>

`is_complete(self)`

Lines 81-86

Purpose:

Check if all steps are completed or skipped.

Parameters:

Returns: bool

`get_progress(self)`

Lines 88-96

Purpose:

Get completion progress as percentage.

Parameters:

Returns: float

Class: `PlanAndExecuteAgent`

Lines 99-453

Purpose:

Agent that uses plan-and-execute pattern.

Methods:

`init(self, model_name, enable_replanning)`

Lines 102-117

Purpose:

Initialize the Plan-and-Execute agent.

Args:
    model_name: Name of the OpenAI model
    enable_replanning: Whether to enable dynamic re-planning

Parameters: – model_name: Language model instance – enable_replanning: Enable replanning

`create_plan(self, goal, context)`

Lines 119-170

Purpose:

Create an execution plan for the given goal.

Args:
    goal: The goal to achieve
    context: Additional context information

Returns:
    ExecutionPlan with structured steps

Parameters: – goal: Goal – context: Context

Returns: ExecutionPlan

`execute_step(self, step, plan)`

Lines 172-212

Purpose:

Execute a single step of the plan.

Args:
    step: The step to execute
    plan: The complete plan for context

Returns:
    Result of executing the step

Parameters: – step: Step – plan: Plan

Returns: str

`_build_context(self, plan)`

Lines 214-221

Purpose:

Build context string from completed steps.

Parameters: – plan: Plan

Returns: str

`should_replan(self, plan, last_step)`

Lines 223-270

Purpose:

Determine if re-planning is needed based on execution results.

Args:
    plan: Current execution plan
    last_step: Most recently executed step

Returns:
    True if re-planning is recommended

Parameters: – plan: Plan – last_step: Last step

Returns: bool

`replan(self, plan)`

Lines 272-328

Purpose:

Create a new plan based on current progress.

Args:
    plan: Current execution plan

Returns:
    Updated execution plan

Parameters: – plan: Plan

Returns: ExecutionPlan

`execute_plan(self, plan, max_steps)`

Lines 330-399

Purpose:

Execute the complete plan.

Args:
    plan: The execution plan
    max_steps: Maximum number of steps to execute (None for all)

Returns:
    Dictionary with execution results

Parameters: – plan: Plan – max_steps: Max steps

Returns: <ast.Subscript object at 0x000001A06D608ED0>

`_generate_summary(self, plan)`

Lines 401-430

Purpose:

Generate a summary of the execution.

Parameters: – plan: Plan

Returns: str

`solve_with_plan_execute(self, goal, context)`

Lines 432-453

Purpose:

Solve a task using plan-and-execute pattern.

Args:
    goal: The goal to achieve
    context: Additional context

Returns:
    Execution results

Parameters: – goal: Goal – context: Context

Returns: <ast.Subscript object at 0x000001A06D627750>

Functions

`demonstrate_plan_execute()`

Lines 456-533

Purpose:

Demonstrates the Plan-and-Execute pattern.

Key Concepts Summary

Pattern Overview

The Plan And Execute 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 005
Pattern 007
Other Reasoning Patterns (001-011)

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 Plan And Execute 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: 006_plan_and_execute.py
Total Lines: 544
Pattern Category: AI Agent Design Pattern #006

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Agentic Design Pattern: Self Consistency


    """
Pattern 005: Self-Consistency

Description:
    Samples multiple reasoning paths for the same question and selects
    the most consistent answer through majority voting or consistency scoring.
    This approach improves reliability by generating diverse solutions and
    aggregating them.

Key Concepts:
    - Multiple Sampling: Generate several independent reasoning chains
    - Diverse Reasoning: Use temperature or prompt variations
    - Aggregation: Majority voting or consistency analysis
    - Answer Selection: Choose most frequent or highest confidence answer

Use Cases:
    - Math problems with multiple solution methods
    - Questions with ambiguous reasoning paths
    - Improving reliability of complex reasoning
    - Reducing model hallucinations

LangChain Implementation:
    Multiple chain invocations with temperature variation and result aggregation.
"""

import os
from typing import List, Dict, Any, Tuple
from collections import Counter
import re
from dotenv import load_dotenv

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

# Load environment variables
load_dotenv()


class SelfConsistencyAgent:
    """Agent that uses self-consistency for reliable reasoning."""
    
    def __init__(self, model_name: str = "gpt-3.5-turbo", num_samples: int = 5):
        """
        Initialize the Self-Consistency agent.
        
        Args:
            model_name: Name of the OpenAI model to use
            num_samples: Number of reasoning paths to sample
        """
        self.model_name = model_name
        self.num_samples = num_samples
        self.base_temperature = 0.7
        
    def _create_llm(self, temperature: float) -> ChatOpenAI:
        """Create an LLM instance with specified temperature."""
        return ChatOpenAI(
            model=self.model_name,
            temperature=temperature,
            api_key=os.getenv("OPENAI_API_KEY")
        )
    
    def generate_reasoning_paths(self, question: str, context: str = "") -> List[Dict[str, str]]:
        """
        Generate multiple diverse reasoning paths for a question.
        
        Args:
            question: The question to answer
            context: Optional context information
            
        Returns:
            List of dictionaries with 'reasoning' and 'answer' keys
        """
        # Base prompt for chain-of-thought reasoning
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Answer the question with clear step-by-step reasoning.
Show your work and explain each step.
At the end, clearly state your final answer."""),
            ("human", "{context}\n\nQuestion: {question}\n\nLet's think step by step:")
        ])
        
        reasoning_paths = []
        
        # Generate multiple samples with varying temperatures
        for i in range(self.num_samples):
            # Vary temperature for diversity
            temperature = self.base_temperature + (i - self.num_samples // 2) * 0.1
            temperature = max(0.1, min(1.0, temperature))
            
            llm = self._create_llm(temperature)
            chain = prompt | llm | StrOutputParser()
            
            try:
                response = chain.invoke({
                    "question": question,
                    "context": context if context else "No additional context."
                })
                
                # Extract answer from response
                answer = self._extract_answer(response)
                
                reasoning_paths.append({
                    "reasoning": response,
                    "answer": answer,
                    "temperature": temperature
                })
                
            except Exception as e:
                print(f"Warning: Sample {i+1} failed: {str(e)}")
                continue
        
        return reasoning_paths
    
    def _extract_answer(self, reasoning: str) -> str:
        """Extract the final answer from reasoning text."""
        # Look for common answer patterns
        patterns = [
            r"final answer is:?\s*(.+?)(?:\n|$)",
            r"answer is:?\s*(.+?)(?:\n|$)",
            r"therefore,?\s+(?:the answer is )?\s*(.+?)(?:\n|$)",
            r"thus,?\s+(.+?)(?:\n|$)",
            r"so,?\s+(?:the answer is )?\s*(.+?)(?:\n|$)",
        ]
        
        reasoning_lower = reasoning.lower()
        for pattern in patterns:
            match = re.search(pattern, reasoning_lower, re.IGNORECASE)
            if match:
                answer = match.group(1).strip()
                # Clean up the answer
                answer = re.sub(r'[*_]', '', answer)  # Remove markdown
                return answer
        
        # If no pattern matches, take last non-empty line
        lines = [line.strip() for line in reasoning.split('\n') if line.strip()]
        return lines[-1] if lines else reasoning[:100]
    
    def aggregate_by_majority_vote(self, reasoning_paths: List[Dict[str, str]]) -> Dict[str, Any]:
        """
        Aggregate answers using majority voting.
        
        Args:
            reasoning_paths: List of reasoning paths with answers
            
        Returns:
            Dictionary with final answer and aggregation statistics
        """
        answers = [path['answer'] for path in reasoning_paths]
        
        # Normalize answers for comparison (lowercase, strip punctuation)
        normalized_answers = []
        for ans in answers:
            normalized = ans.lower().strip().rstrip('.')
            normalized_answers.append(normalized)
        
        # Count occurrences
        answer_counts = Counter(normalized_answers)
        most_common_answer, count = answer_counts.most_common(1)[0]
        
        # Find original form of the answer
        original_answer = next(
            (answers[i] for i, norm in enumerate(normalized_answers) 
             if norm == most_common_answer),
            most_common_answer
        )
        
        return {
            "final_answer": original_answer,
            "confidence": count / len(answers),
            "vote_count": count,
            "total_samples": len(answers),
            "all_answers": answer_counts,
            "agreement_rate": count / len(answers)
        }
    
    def aggregate_by_consistency_score(self, reasoning_paths: List[Dict[str, str]], 
                                       question: str) -> Dict[str, Any]:
        """
        Aggregate answers using consistency scoring via LLM.
        
        Args:
            reasoning_paths: List of reasoning paths with answers
            question: Original question
            
        Returns:
            Dictionary with final answer and consistency scores
        """
        # Use LLM to evaluate consistency of each answer with others
        answers_text = "\n".join([
            f"{i+1}. {path['answer']}" 
            for i, path in enumerate(reasoning_paths)
        ])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are evaluating multiple answers to the same question.
Identify which answer is most consistent with the majority of other answers.
Consider semantic similarity, not just exact matches.

Respond with the number (1, 2, 3, etc.) of the most consistent answer."""),
            ("human", """Question: {question}

Answers:
{answers}

Most consistent answer number:""")
        ])
        
        llm = self._create_llm(0.3)  # Low temperature for consistent evaluation
        chain = prompt | llm | StrOutputParser()
        
        try:
            response = chain.invoke({
                "question": question,
                "answers": answers_text
            })
            
            # Extract number
            match = re.search(r'\d+', response)
            if match:
                index = int(match.group()) - 1
                if 0 <= index  Dict[str, Any]:
        """
        Solve a question using self-consistency.
        
        Args:
            question: The question to answer
            context: Optional context
            method: Aggregation method ('majority_vote' or 'consistency_score')
            
        Returns:
            Dictionary with solution and metadata
        """
        print(f"\nQuestion: {question}\n")
        print(f"Generating {self.num_samples} reasoning paths...\n")
        
        # Generate multiple reasoning paths
        reasoning_paths = self.generate_reasoning_paths(question, context)
        
        if not reasoning_paths:
            return {
                "question": question,
                "final_answer": "Unable to generate reasoning paths",
                "error": "All samples failed"
            }
        
        print(f"Generated {len(reasoning_paths)} paths\n")
        
        # Show individual reasoning paths
        for i, path in enumerate(reasoning_paths, 1):
            print(f"Path {i} (temp={path['temperature']:.2f}):")
            print(f"  Answer: {path['answer']}")
            print()
        
        # Aggregate results
        if method == "majority_vote":
            result = self.aggregate_by_majority_vote(reasoning_paths)
        else:
            result = self.aggregate_by_consistency_score(reasoning_paths, question)
        
        result["question"] = question
        result["reasoning_paths"] = len(reasoning_paths)
        
        return result


def demonstrate_self_consistency():
    """Demonstrates the Self-Consistency pattern."""
    
    print("=" * 80)
    print("PATTERN 005: Self-Consistency")
    print("=" * 80)
    print()
    print("Self-Consistency improves answer reliability by:")
    print("1. Generating multiple independent reasoning paths")
    print("2. Sampling with different temperatures for diversity")
    print("3. Aggregating answers via majority voting or consistency scoring")
    print()
    
    # Create agent
    agent = SelfConsistencyAgent(num_samples=5)
    
    # Test cases
    test_cases = [
        {
            "question": "If a train travels 120 miles in 2 hours, then travels 180 miles in 3 hours, what is its average speed for the entire journey?",
            "context": "",
            "method": "majority_vote"
        },
        {
            "question": "A bat and ball cost $1.10 together. The bat costs $1.00 more than the ball. How much does the ball cost?",
            "context": "",
            "method": "majority_vote"
        },
        {
            "question": "Is it ethical for AI to make medical diagnoses?",
            "context": "Consider patient safety, accuracy, and doctor-patient relationships.",
            "method": "consistency_score"
        }
    ]
    
    for i, test_case in enumerate(test_cases, 1):
        print(f"\n{'=' * 80}")
        print(f"Example {i}: {test_case['method'].replace('_', ' ').title()}")
        print('=' * 80)
        
        try:
            result = agent.solve_with_self_consistency(
                question=test_case['question'],
                context=test_case['context'],
                method=test_case['method']
            )
            
            print(f"\n{'=' * 80}")
            print("FINAL RESULT")
            print('=' * 80)
            print(f"\nFinal Answer: {result['final_answer']}")
            
            if 'confidence' in result:
                print(f"Confidence: {result['confidence']:.1%}")
                print(f"Agreement Rate: {result['agreement_rate']:.1%}")
                print(f"Vote Distribution: {dict(result['all_answers'])}")
            
            print(f"Reasoning Paths: {result['reasoning_paths']}/{agent.num_samples}")
            
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
            import traceback
            traceback.print_exc()
    
    # Summary
    print("\n\n" + "=" * 80)
    print("SELF-CONSISTENCY PATTERN DEMONSTRATION COMPLETE")
    print("=" * 80)
    print()
    print("Key Techniques Demonstrated:")
    print("1. Multiple Sampling: Generated 5 diverse reasoning paths")
    print("2. Temperature Variation: Used different temperatures for diversity")
    print("3. Majority Voting: Aggregated answers by frequency")
    print("4. Consistency Scoring: LLM-based semantic consistency evaluation")
    print("5. Confidence Metrics: Agreement rate and vote distribution")
    print()
    print("Benefits:")
    print("- Improved reliability through multiple reasoning paths")
    print("- Reduced impact of outlier/hallucinated answers")
    print("- Confidence estimation via agreement metrics")
    print("- Works with any reasoning task")
    print()
    print("LangChain Components Used:")
    print("- ChatPromptTemplate: Structures reasoning prompts")
    print("- StrOutputParser: Parses LLM outputs")
    print("- Temperature variation: Creates diverse samples")
    print("- Multiple chain invocations: Parallel reasoning paths")
    print()


if __name__ == "__main__":
    # Check for API key
    if not os.getenv("OPENAI_API_KEY"):
        print("Error: OPENAI_API_KEY not found in environment variables.")
        print("Please set it in your .env file or environment.")
        exit(1)
    
    demonstrate_self_consistency()

Agentic Design Pattern: Self Consistency

Pattern 005: Self Consistency – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Self Consistency pattern implementation.

Pattern Description

Pattern 005: Self-Consistency

Description: Samples multiple reasoning paths for the same question and selects the most consistent answer through majority voting or consistency scoring. This approach improves reliability by generating diverse solutions and aggregating them.

Key Concepts: – Multiple Sampling: Generate several independent reasoning chains – Diverse Reasoning: Use temperature or prompt variations – Aggregation: Majority voting or consistency analysis – Answer Selection: Choose most frequent or highest confidence answer

Use Cases: – Math problems with multiple solution methods – Questions with ambiguous reasoning paths – Improving reliability of complex reasoning – Reducing model hallucinations

LangChain Implementation: Multiple chain invocations with temperature variation and result aggregation.

File Statistics

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

Module Documentation

"""
Pattern 005: Self-Consistency

Description:
    Samples multiple reasoning paths for the same question and selects
    the most consistent answer through majority voting or consistency scoring.
    This approach improves reliability by generating diverse solutions and
    aggregating them.

Key Concepts:
    - Multiple Sampling: Generate several independent reasoning chains
    - Diverse Reasoning: Use temperature or prompt variations
    - Aggregation: Majority voting or consistency analysis
    - Answer Selection: Choose most frequent or highest confidence answer

Use Cases:
    - Math problems with multiple solution methods
    - Questions with ambiguous reasoning paths
    - Improving reliability of complex reasoning
    - Reducing model hallucinations

LangChain Implementation:
    Multiple chain invocations with temperature variation and result aggregation.
"""

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

Import Statements

Standard Library Imports

import collections.Counter
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import re
import typing.Any
import typing.Dict
import typing.List
import typing.Tuple

Purpose: Core Python functionality: – collections.Counter: Specialized container datatypes – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – re: Regular expression 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.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
from traceback

Purpose: Additional dependencies for specialized functionality.

Classes

Class: `SelfConsistencyAgent`

Lines 40-283

Purpose:

Agent that uses self-consistency for reliable reasoning.

Methods:

`init(self, model_name, num_samples)`

Lines 43-53

Purpose:

Initialize the Self-Consistency agent.

Args:
    model_name: Name of the OpenAI model to use
    num_samples: Number of reasoning paths to sample

Parameters: – model_name: Language model instance – num_samples: Num samples

`_create_llm(self, temperature)`

Lines 55-61

Purpose:

Create an LLM instance with specified temperature.

Parameters: – temperature: Sampling temperature (0-1)

Returns: ChatOpenAI

`generate_reasoning_paths(self, question, context)`

Lines 63-112

Purpose:

Generate multiple diverse reasoning paths for a question.

Args:
    question: The question to answer
    context: Optional context information

Returns:
    List of dictionaries with 'reasoning' and 'answer' keys

Parameters: – question: Question to answer – context: Context

Returns: <ast.Subscript object at 0x000001A06D696050>

`_extract_answer(self, reasoning)`

Lines 114-136

Purpose:

Extract the final answer from reasoning text.

Parameters: – reasoning: Reasoning

Returns: str

`aggregate_by_majority_vote(self, reasoning_paths)`

Lines 138-174

Purpose:

Aggregate answers using majority voting.

Args:
    reasoning_paths: List of reasoning paths with answers

Returns:
    Dictionary with final answer and aggregation statistics

Parameters: – reasoning_paths: Reasoning paths

Returns: <ast.Subscript object at 0x000001A06D168F10>

`aggregate_by_consistency_score(self, reasoning_paths, question)`

Lines 176-238

Purpose:

Aggregate answers using consistency scoring via LLM.

Args:
    reasoning_paths: List of reasoning paths with answers
    question: Original question

Returns:
    Dictionary with final answer and consistency scores

Parameters: – reasoning_paths: Reasoning paths – question: Question to answer

Returns: <ast.Subscript object at 0x000001A06D158310>

`solve_with_self_consistency(self, question, context, method)`

Lines 240-283

Purpose:

Solve a question using self-consistency.

Args:
    question: The question to answer
    context: Optional context
    method: Aggregation method ('majority_vote' or 'consistency_score')

Returns:
    Dictionary with solution and metadata

Parameters: – question: Question to answer – context: Context – method: Method

Returns: <ast.Subscript object at 0x000001A06D156DD0>

Functions

`demonstrate_self_consistency()`

Lines 286-373

Purpose:

Demonstrates the Self-Consistency pattern.

Key Concepts Summary

Pattern Overview

The Self Consistency 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 004
Pattern 006
Other Reasoning Patterns (001-011)

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 Self Consistency 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: 005_self_consistency.py
Total Lines: 384
Pattern Category: AI Agent Design Pattern #005

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Agentic Design Pattern: Graph Of Thoughts


    """
Pattern 004: Graph-of-Thoughts (GoT)

Description:
    Represents reasoning as a directed graph with nodes and edges.
    Unlike Tree-of-Thoughts which explores a tree structure, GoT allows for
    more flexible graph structures with cycles, merging paths, and complex
    relationships between thoughts.

Components:
    - Thought Nodes: Individual reasoning steps or ideas
    - Transformation Edges: Operations that transform one thought to another
    - Aggregation: Combining multiple thoughts into one
    - Graph Operations: Add, merge, transform, aggregate

Use Cases:
    - Complex multi-step reasoning with dependencies
    - Document analysis with cross-references
    - Problem-solving requiring non-linear thinking
    - Knowledge synthesis from multiple sources

LangChain Implementation:
    Uses LangGraph for state graph management and custom graph operations.
"""

import os
from typing import List, Dict, Any, Optional, Set
from enum import Enum
from dataclasses import dataclass, field
from collections import defaultdict
from dotenv import load_dotenv

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

# Load environment variables
load_dotenv()


class EdgeType(Enum):
    """Types of edges in the graph."""
    TRANSFORM = "transform"  # Transform one thought to another
    AGGREGATE = "aggregate"  # Combine multiple thoughts
    REFINE = "refine"  # Refine/improve a thought
    COMPARE = "compare"  # Compare two thoughts
    VALIDATE = "validate"  # Validate a thought


@dataclass
class ThoughtNode:
    """Represents a node in the graph of thoughts."""
    id: str
    content: str
    score: float = 0.0
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def __hash__(self):
        return hash(self.id)
    
    def __eq__(self, other):
        return isinstance(other, ThoughtNode) and self.id == other.id


@dataclass
class ThoughtEdge:
    """Represents an edge in the graph of thoughts."""
    source: str  # Source node ID
    target: str  # Target node ID
    edge_type: EdgeType
    operation: str  # Description of the operation
    weight: float = 1.0


class ThoughtGraph:
    """Graph structure for managing thoughts."""
    
    def __init__(self):
        self.nodes: Dict[str, ThoughtNode] = {}
        self.edges: List[ThoughtEdge] = []
        self.adjacency: Dict[str, List[str]] = defaultdict(list)
        self.reverse_adjacency: Dict[str, List[str]] = defaultdict(list)
    
    def add_node(self, node: ThoughtNode):
        """Add a node to the graph."""
        self.nodes[node.id] = node
    
    def add_edge(self, edge: ThoughtEdge):
        """Add an edge to the graph."""
        self.edges.append(edge)
        self.adjacency[edge.source].append(edge.target)
        self.reverse_adjacency[edge.target].append(edge.source)
    
    def get_node(self, node_id: str) -> Optional[ThoughtNode]:
        """Get a node by ID."""
        return self.nodes.get(node_id)
    
    def get_children(self, node_id: str) -> List[ThoughtNode]:
        """Get all child nodes."""
        return [self.nodes[child_id] for child_id in self.adjacency.get(node_id, [])]
    
    def get_parents(self, node_id: str) -> List[ThoughtNode]:
        """Get all parent nodes."""
        return [self.nodes[parent_id] for parent_id in self.reverse_adjacency.get(node_id, [])]
    
    def get_all_paths(self, start_id: str, end_id: str) -> List[List[str]]:
        """Find all paths from start to end node."""
        paths = []
        visited = set()
        
        def dfs(current: str, path: List[str]):
            if current == end_id:
                paths.append(path.copy())
                return
            
            visited.add(current)
            for neighbor in self.adjacency.get(current, []):
                if neighbor not in visited:
                    path.append(neighbor)
                    dfs(neighbor, path)
                    path.pop()
            visited.remove(current)
        
        dfs(start_id, [start_id])
        return paths


class GraphOfThoughtsAgent:
    """Agent that uses Graph-of-Thoughts reasoning."""
    
    def __init__(self, model_name: str = "gpt-3.5-turbo", temperature: float = 0.7):
        """Initialize the Graph-of-Thoughts agent."""
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=temperature,
            api_key=os.getenv("OPENAI_API_KEY")
        )
        self.graph = ThoughtGraph()
        self.node_counter = 0
    
    def generate_initial_thoughts(self, problem: str, num_thoughts: int = 3) -> List[ThoughtNode]:
        """Generate initial thoughts for the problem."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Generate {num_thoughts} diverse initial approaches or perspectives 
for solving the problem. Each should be a distinct starting point.

Format your response as:
1. [First approach]
2. [Second approach]
3. [Third approach]"""),
            ("human", "Problem: {problem}")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        response = chain.invoke({"problem": problem, "num_thoughts": num_thoughts})
        
        # Parse thoughts
        thoughts = []
        for line in response.strip().split('\n'):
            line = line.strip()
            if line and (line[0].isdigit() or line.startswith('-')):
                content = line.split('.', 1)[-1].strip().lstrip('- ').strip()
                if content:
                    node = ThoughtNode(
                        id=f"node_{self.node_counter}",
                        content=content,
                        metadata={"type": "initial", "problem": problem}
                    )
                    self.node_counter += 1
                    thoughts.append(node)
                    self.graph.add_node(node)
        
        return thoughts[:num_thoughts]
    
    def transform_thought(self, node: ThoughtNode, transformation: str) -> ThoughtNode:
        """Transform a thought using a specific operation."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Transform the given thought by: {transformation}"),
            ("human", "Original thought: {thought}\n\nTransformed thought:")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        result = chain.invoke({
            "transformation": transformation,
            "thought": node.content
        })
        
        new_node = ThoughtNode(
            id=f"node_{self.node_counter}",
            content=result.strip(),
            metadata={"type": "transformed", "parent": node.id}
        )
        self.node_counter += 1
        
        self.graph.add_node(new_node)
        self.graph.add_edge(ThoughtEdge(
            source=node.id,
            target=new_node.id,
            edge_type=EdgeType.TRANSFORM,
            operation=transformation
        ))
        
        return new_node
    
    def aggregate_thoughts(self, nodes: List[ThoughtNode], aggregation_type: str = "synthesis") -> ThoughtNode:
        """Aggregate multiple thoughts into one."""
        thoughts_text = "\n".join([f"- {node.content}" for node in nodes])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Synthesize the following thoughts into a single, coherent insight.
Combine the key ideas while maintaining logical consistency."""),
            ("human", "Thoughts to synthesize:\n{thoughts}\n\nSynthesized insight:")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        result = chain.invoke({"thoughts": thoughts_text})
        
        new_node = ThoughtNode(
            id=f"node_{self.node_counter}",
            content=result.strip(),
            metadata={"type": "aggregated", "parents": [n.id for n in nodes]}
        )
        self.node_counter += 1
        
        self.graph.add_node(new_node)
        for node in nodes:
            self.graph.add_edge(ThoughtEdge(
                source=node.id,
                target=new_node.id,
                edge_type=EdgeType.AGGREGATE,
                operation=aggregation_type
            ))
        
        return new_node
    
    def refine_thought(self, node: ThoughtNode, feedback: str) -> ThoughtNode:
        """Refine a thought based on feedback."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Refine the thought based on this feedback: {feedback}"),
            ("human", "Original thought: {thought}\n\nRefined thought:")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        result = chain.invoke({
            "feedback": feedback,
            "thought": node.content
        })
        
        new_node = ThoughtNode(
            id=f"node_{self.node_counter}",
            content=result.strip(),
            metadata={"type": "refined", "parent": node.id}
        )
        self.node_counter += 1
        
        self.graph.add_node(new_node)
        self.graph.add_edge(ThoughtEdge(
            source=node.id,
            target=new_node.id,
            edge_type=EdgeType.REFINE,
            operation="refinement"
        ))
        
        return new_node
    
    def evaluate_thought(self, node: ThoughtNode, problem: str) -> float:
        """Evaluate how well a thought addresses the problem."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Rate how well this thought addresses the problem on a scale of 0.0 to 1.0.
Consider relevance, feasibility, and completeness.
Respond with ONLY a number between 0.0 and 1.0."""),
            ("human", """Problem: {problem}

Thought: {thought}

Score (0.0-1.0):""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        try:
            response = chain.invoke({"problem": problem, "thought": node.content})
            score = float(response.strip())
            node.score = max(0.0, min(1.0, score))
            return node.score
        except:
            return 0.5
    
    def solve_with_graph(self, problem: str, max_iterations: int = 3) -> Dict[str, Any]:
        """
        Solve a problem using Graph-of-Thoughts approach.
        
        Args:
            problem: The problem to solve
            max_iterations: Maximum number of transformation iterations
            
        Returns:
            Dictionary with solution and graph information
        """
        print(f"\nSolving: {problem}\n")
        
        # Generate initial thoughts
        print("Step 1: Generate initial thoughts")
        initial_thoughts = self.generate_initial_thoughts(problem, num_thoughts=3)
        for i, node in enumerate(initial_thoughts, 1):
            print(f"  Thought {i}: {node.content[:70]}...")
        
        # Transform and refine thoughts
        print(f"\nStep 2: Transform and refine thoughts")
        transformed_nodes = []
        
        for node in initial_thoughts[:2]:  # Transform first two
            # Apply transformations
            transformation = "Make it more specific and actionable"
            transformed = self.transform_thought(node, transformation)
            print(f"  Transformed: {transformed.content[:70]}...")
            transformed_nodes.append(transformed)
        
        # Aggregate thoughts
        print(f"\nStep 3: Aggregate insights")
        if len(transformed_nodes) >= 2:
            aggregated = self.aggregate_thoughts(transformed_nodes[:2])
            print(f"  Aggregated: {aggregated.content[:70]}...")
            
            # Refine the aggregated thought
            print(f"\nStep 4: Refine final insight")
            refined = self.refine_thought(aggregated, "Make it more comprehensive and practical")
            print(f"  Refined: {refined.content[:70]}...")
            
            # Evaluate final thought
            final_score = self.evaluate_thought(refined, problem)
            print(f"\n  Final score: {final_score:.2f}")
            
            final_node = refined
        else:
            final_node = transformed_nodes[0] if transformed_nodes else initial_thoughts[0]
            final_score = self.evaluate_thought(final_node, problem)
        
        return {
            "problem": problem,
            "solution": final_node.content,
            "score": final_score,
            "graph_stats": {
                "total_nodes": len(self.graph.nodes),
                "total_edges": len(self.graph.edges),
                "node_types": self._count_node_types()
            }
        }
    
    def _count_node_types(self) -> Dict[str, int]:
        """Count nodes by type."""
        counts = defaultdict(int)
        for node in self.graph.nodes.values():
            node_type = node.metadata.get("type", "unknown")
            counts[node_type] += 1
        return dict(counts)
    
    def visualize_graph(self) -> str:
        """Generate a text visualization of the graph."""
        lines = ["Graph Structure:", "=" * 60]
        
        for node in self.graph.nodes.values():
            node_type = node.metadata.get("type", "unknown")
            lines.append(f"\n[{node.id}] ({node_type})")
            lines.append(f"  Content: {node.content[:60]}...")
            lines.append(f"  Score: {node.score:.2f}")
            
            children = self.graph.get_children(node.id)
            if children:
                lines.append(f"  Children: {', '.join([c.id for c in children])}")
        
        return "\n".join(lines)


def demonstrate_got_pattern():
    """Demonstrates the Graph-of-Thoughts pattern."""
    
    print("=" * 80)
    print("PATTERN 004: Graph-of-Thoughts (GoT)")
    print("=" * 80)
    print()
    
    # Create GoT agent
    agent = GraphOfThoughtsAgent()
    
    # Test problems
    problems = [
        "Design a sustainable urban transportation system for a growing city",
        "Create a strategy to improve team collaboration in a remote work environment"
    ]
    
    for i, problem in enumerate(problems, 1):
        print(f"\n{'=' * 80}")
        print(f"Problem {i}: {problem}")
        print('=' * 80)
        
        try:
            # Reset agent for new problem
            agent = GraphOfThoughtsAgent()
            
            result = agent.solve_with_graph(problem, max_iterations=3)
            
            print(f"\n\n{'=' * 80}")
            print("SOLUTION")
            print('=' * 80)
            print(result['solution'])
            print(f"\nFinal Score: {result['score']:.2f}")
            print(f"\nGraph Statistics:")
            print(f"  Total Nodes: {result['graph_stats']['total_nodes']}")
            print(f"  Total Edges: {result['graph_stats']['total_edges']}")
            print(f"  Node Types: {result['graph_stats']['node_types']}")
            
            # Show graph structure
            print(f"\n{agent.visualize_graph()}")
            
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
            import traceback
            traceback.print_exc()
    
    # Summary
    print("\n\n" + "=" * 80)
    print("GRAPH-OF-THOUGHTS PATTERN DEMONSTRATION COMPLETE")
    print("=" * 80)
    print()
    print("Key Features Demonstrated:")
    print("1. Graph structure for non-linear reasoning")
    print("2. Thought transformation operations")
    print("3. Aggregation of multiple thoughts")
    print("4. Refinement based on feedback")
    print("5. Flexible graph operations (transform, aggregate, refine)")
    print()
    print("Advantages over Tree-of-Thoughts:")
    print("- More flexible thought relationships")
    print("- Can merge multiple reasoning paths")
    print("- Supports cycles and cross-references")
    print("- Better for complex, interconnected problems")
    print()
    print("LangChain Components Used:")
    print("- ChatPromptTemplate: Structures transformation prompts")
    print("- StrOutputParser: Parses LLM responses")
    print("- Custom graph structure: Manages thought relationships")
    print("- Multiple edge types: Different operations between thoughts")
    print()


if __name__ == "__main__":
    # Check for API key
    if not os.getenv("OPENAI_API_KEY"):
        print("Error: OPENAI_API_KEY not found in environment variables.")
        print("Please set it in your .env file or environment.")
        exit(1)
    
    demonstrate_got_pattern()

Agentic Design Pattern: Graph Of Thoughts

Pattern 004: Graph Of Thoughts – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Graph Of Thoughts pattern implementation.

Pattern Description

Pattern 004: Graph-of-Thoughts (GoT)

Description: Represents reasoning as a directed graph with nodes and edges. Unlike Tree-of-Thoughts which explores a tree structure, GoT allows for more flexible graph structures with cycles, merging paths, and complex relationships between thoughts.

Components: – Thought Nodes: Individual reasoning steps or ideas – Transformation Edges: Operations that transform one thought to another – Aggregation: Combining multiple thoughts into one – Graph Operations: Add, merge, transform, aggregate

Use Cases: – Complex multi-step reasoning with dependencies – Document analysis with cross-references – Problem-solving requiring non-linear thinking – Knowledge synthesis from multiple sources

LangChain Implementation: Uses LangGraph for state graph management and custom graph operations.

File Statistics

Total Lines: 455
Classes: 5
Functions: 1
Constants: 0

Module Documentation

"""
Pattern 004: Graph-of-Thoughts (GoT)

Description:
    Represents reasoning as a directed graph with nodes and edges.
    Unlike Tree-of-Thoughts which explores a tree structure, GoT allows for
    more flexible graph structures with cycles, merging paths, and complex
    relationships between thoughts.

Components:
    - Thought Nodes: Individual reasoning steps or ideas
    - Transformation Edges: Operations that transform one thought to another
    - Aggregation: Combining multiple thoughts into one
    - Graph Operations: Add, merge, transform, aggregate

Use Cases:
    - Complex multi-step reasoning with dependencies
    - Document analysis with cross-references
    - Problem-solving requiring non-linear thinking
    - Knowledge synthesis from multiple sources

LangChain Implementation:
    Uses LangGraph for state graph management and custom graph operations.
"""

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

Import Statements

Standard Library Imports

import collections.defaultdict
import dataclasses.dataclass
import dataclasses.field
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
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 – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – 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.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
from traceback

Purpose: Additional dependencies for specialized functionality.

Classes

Class: `EdgeType`

Lines 41-47

Inherits from: Enum

Purpose:

Types of edges in the graph.

Class: `ThoughtNode`

Lines 51-62

Purpose:

Represents a node in the graph of thoughts.

Methods:

`hash(self)`

Lines 58-59

Purpose: Hash operation.

Parameters:

`eq(self, other)`

Lines 61-62

Purpose: Eq operation.

Parameters: – other: Other

Class: `ThoughtEdge`

Lines 66-72

Purpose:

Represents an edge in the graph of thoughts.

Class: `ThoughtGraph`

Lines 75-125

Purpose:

Graph structure for managing thoughts.

Methods:

`init(self)`

Lines 78-82

Purpose: Initialize the class instance with required parameters.

Parameters:

`add_node(self, node)`

Lines 84-86

Purpose:

Add a node to the graph.

Parameters: – node: Node

`add_edge(self, edge)`

Lines 88-92

Purpose:

Add an edge to the graph.

Parameters: – edge: Edge

`get_node(self, node_id)`

Lines 94-96

Purpose:

Get a node by ID.

Parameters: – node_id: Node id

Returns: <ast.Subscript object at 0x000001A06D624950>

`get_children(self, node_id)`

Lines 98-100

Purpose:

Get all child nodes.

Parameters: – node_id: Node id

Returns: <ast.Subscript object at 0x000001A06D6AB9D0>

`get_parents(self, node_id)`

Lines 102-104

Purpose:

Get all parent nodes.

Parameters: – node_id: Node id

Returns: <ast.Subscript object at 0x000001A06D6AAA90>

`get_all_paths(self, start_id, end_id)`

Lines 106-125

Purpose:

Find all paths from start to end node.

Parameters: – start_id: Start id – end_id: End id

Returns: <ast.Subscript object at 0x000001A06D5F7CD0>

Class: `GraphOfThoughtsAgent`

Lines 128-372

Purpose:

Agent that uses Graph-of-Thoughts reasoning.

Methods:

`init(self, model_name, temperature)`

Lines 131-139

Purpose:

Initialize the Graph-of-Thoughts agent.

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

`generate_initial_thoughts(self, problem, num_thoughts)`

Lines 141-173

Purpose:

Generate initial thoughts for the problem.

Parameters: – problem: Problem – num_thoughts: Num thoughts

Returns: <ast.Subscript object at 0x000001A06D617310>

`transform_thought(self, node, transformation)`

Lines 175-203

Purpose:

Transform a thought using a specific operation.

Parameters: – node: Node – transformation: Transformation

Returns: ThoughtNode

`aggregate_thoughts(self, nodes, aggregation_type)`

Lines 205-234

Purpose:

Aggregate multiple thoughts into one.

Parameters: – nodes: Nodes – aggregation_type: Aggregation type

Returns: ThoughtNode

`refine_thought(self, node, feedback)`

Lines 236-264

Purpose:

Refine a thought based on feedback.

Parameters: – node: Node – feedback: Feedback

Returns: ThoughtNode

`evaluate_thought(self, node, problem)`

Lines 266-287

Purpose:

Evaluate how well a thought addresses the problem.

Parameters: – node: Node – problem: Problem

Returns: float

`solve_with_graph(self, problem, max_iterations)`

Lines 289-348

Purpose:

Solve a problem using Graph-of-Thoughts approach.

Args:
    problem: The problem to solve
    max_iterations: Maximum number of transformation iterations

Returns:
    Dictionary with solution and graph information

Parameters: – problem: Problem – max_iterations: Max iterations

Returns: <ast.Subscript object at 0x000001A06D6C9C90>

`_count_node_types(self)`

Lines 350-356

Purpose:

Count nodes by type.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D6CB550>

`visualize_graph(self)`

Lines 358-372

Purpose:

Generate a text visualization of the graph.

Parameters:

Returns: str

Functions

`demonstrate_got_pattern()`

Lines 375-444

Purpose:

Demonstrates the Graph-of-Thoughts pattern.

Key Concepts Summary

Pattern Overview

The Graph Of Thoughts 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 003
Pattern 005
Other Reasoning Patterns (001-011)

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 Graph Of Thoughts 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: 004_graph_of_thoughts.py
Total Lines: 455
Pattern Category: AI Agent Design Pattern #004

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Agentic Design Pattern: Tree Of Thoughts


    """
Pattern 003: Tree-of-Thoughts (ToT)

Description:
    Explores multiple reasoning paths simultaneously in a tree structure.
    ToT allows the model to generate multiple candidate thoughts at each step,
    evaluate them, and search through the tree (BFS or DFS) to find the best solution.

Components:
    - Thought Generation: Creates multiple candidate next steps
    - State Evaluation: Scores each thought/state
    - Search Strategy: BFS, DFS, or beam search
    - Backtracking: Can revisit and try alternative paths

Use Cases:
    - Strategic planning and game playing
    - Creative writing with multiple storylines
    - Complex problem-solving with multiple approaches
    - Optimization problems

LangChain Implementation:
    Uses LangGraph for state management and custom search logic for tree exploration.
"""

import os
from typing import List, Dict, Any, Optional, Tuple
from enum import Enum
from dataclasses import dataclass, field
from collections import deque
from dotenv import load_dotenv

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

# Load environment variables
load_dotenv()


class SearchStrategy(Enum):
    """Search strategies for tree exploration."""
    BFS = "breadth_first"
    DFS = "depth_first"
    BEAM = "beam_search"


@dataclass
class ThoughtNode:
    """Represents a node in the tree of thoughts."""
    thought: str
    state: str
    depth: int
    score: float = 0.0
    parent: Optional['ThoughtNode'] = None
    children: List['ThoughtNode'] = field(default_factory=list)
    is_solution: bool = False
    
    def get_path(self) -> List[str]:
        """Returns the path from root to this node."""
        path = []
        current = self
        while current:
            path.append(current.thought)
            current = current.parent
        return list(reversed(path))


class TreeOfThoughtsAgent:
    """Agent that uses Tree-of-Thoughts reasoning."""
    
    def __init__(
        self,
        model_name: str = "gpt-3.5-turbo",
        temperature: float = 0.7,
        branching_factor: int = 3,
        max_depth: int = 4
    ):
        """
        Initialize the Tree-of-Thoughts agent.
        
        Args:
            model_name: LLM model to use
            temperature: Temperature for thought generation
            branching_factor: Number of thoughts to generate at each node
            max_depth: Maximum depth of the tree
        """
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=temperature,
            api_key=os.getenv("OPENAI_API_KEY")
        )
        self.branching_factor = branching_factor
        self.max_depth = max_depth
    
    def generate_thoughts(
        self,
        problem: str,
        current_state: str,
        num_thoughts: int = 3
    ) -> List[str]:
        """
        Generate multiple candidate thoughts/next steps.
        
        Args:
            problem: The original problem
            current_state: Current problem-solving state
            num_thoughts: Number of thoughts to generate
            
        Returns:
            List of candidate thoughts
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a creative problem solver. Generate {num_thoughts} different 
potential next steps or thoughts for solving the problem. Each thought should explore 
a different approach or direction. Be concise and specific.

Format your response as:
1. [First thought]
2. [Second thought]
3. [Third thought]"""),
            ("human", """Problem: {problem}

Current state: {current_state}

Generate {num_thoughts} diverse next steps:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        response = chain.invoke({
            "problem": problem,
            "current_state": current_state,
            "num_thoughts": num_thoughts
        })
        
        # Parse thoughts from numbered list
        thoughts = []
        for line in response.strip().split('\n'):
            line = line.strip()
            if line and (line[0].isdigit() or line.startswith('-')):
                # Remove numbering/bullets
                thought = line.split('.', 1)[-1].strip()
                thought = thought.lstrip('- ').strip()
                if thought:
                    thoughts.append(thought)
        
        return thoughts[:num_thoughts]
    
    def evaluate_thought(self, problem: str, thought: str, current_state: str) -> float:
        """
        Evaluate the quality/promise of a thought.
        
        Args:
            problem: The original problem
            thought: The thought to evaluate
            current_state: Current state description
            
        Returns:
            Score between 0 and 1
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an expert evaluator. Rate how promising this thought is 
for solving the problem on a scale of 0.0 to 1.0, where:
- 0.0-0.3: Unlikely to help or may lead to dead end
- 0.4-0.6: Moderately promising
- 0.7-0.9: Very promising
- 1.0: This is the solution or very close to it

Respond with ONLY a number between 0.0 and 1.0."""),
            ("human", """Problem: {problem}

Current state: {current_state}

Thought to evaluate: {thought}

Score (0.0-1.0):""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        try:
            response = chain.invoke({
                "problem": problem,
                "current_state": current_state,
                "thought": thought
            })
            score = float(response.strip())
            return max(0.0, min(1.0, score))  # Clamp to [0, 1]
        except:
            return 0.5  # Default score if parsing fails
    
    def is_solution(self, problem: str, current_state: str) -> bool:
        """
        Check if current state represents a solution.
        
        Args:
            problem: The original problem
            current_state: Current state description
            
        Returns:
            True if this is a valid solution
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Determine if the current state represents a complete solution 
to the problem. Respond with ONLY 'yes' or 'no'."""),
            ("human", """Problem: {problem}

Current state: {current_state}

Is this a complete solution? (yes/no):""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        response = chain.invoke({
            "problem": problem,
            "current_state": current_state
        })
        
        return response.strip().lower().startswith('yes')
    
    def search_bfs(self, problem: str, max_nodes: int = 20) -> Optional[ThoughtNode]:
        """
        Breadth-first search through the tree of thoughts.
        
        Args:
            problem: The problem to solve
            max_nodes: Maximum number of nodes to explore
            
        Returns:
            Solution node if found, None otherwise
        """
        # Initialize root
        root = ThoughtNode(
            thought="Start",
            state=f"Initial problem: {problem}",
            depth=0
        )
        
        queue = deque([root])
        nodes_explored = 0
        
        while queue and nodes_explored  0 and self.is_solution(problem, current.state):
                current.is_solution = True
                return current
            
            # Don't expand beyond max depth
            if current.depth >= self.max_depth:
                continue
            
            # Generate and evaluate child thoughts
            thoughts = self.generate_thoughts(
                problem,
                current.state,
                self.branching_factor
            )
            
            for thought in thoughts:
                new_state = f"{current.state} → {thought}"
                score = self.evaluate_thought(problem, thought, current.state)
                
                child = ThoughtNode(
                    thought=thought,
                    state=new_state,
                    depth=current.depth + 1,
                    score=score,
                    parent=current
                )
                
                current.children.append(child)
                
                # Only add promising nodes to queue (score > 0.4)
                if score > 0.4:
                    queue.append(child)
        
        # Return best leaf node if no solution found
        return self._find_best_leaf(root)
    
    def search_dfs(self, problem: str, max_nodes: int = 20) -> Optional[ThoughtNode]:
        """
        Depth-first search through the tree of thoughts.
        
        Args:
            problem: The problem to solve
            max_nodes: Maximum number of nodes to explore
            
        Returns:
            Solution node if found, None otherwise
        """
        # Initialize root
        root = ThoughtNode(
            thought="Start",
            state=f"Initial problem: {problem}",
            depth=0
        )
        
        stack = [root]
        nodes_explored = 0
        
        while stack and nodes_explored  0 and self.is_solution(problem, current.state):
                current.is_solution = True
                return current
            
            # Don't expand beyond max depth
            if current.depth >= self.max_depth:
                continue
            
            # Generate and evaluate child thoughts
            thoughts = self.generate_thoughts(
                problem,
                current.state,
                self.branching_factor
            )
            
            for thought in thoughts:
                new_state = f"{current.state} → {thought}"
                score = self.evaluate_thought(problem, thought, current.state)
                
                child = ThoughtNode(
                    thought=thought,
                    state=new_state,
                    depth=current.depth + 1,
                    score=score,
                    parent=current
                )
                
                current.children.append(child)
                
                # Only explore promising nodes (score > 0.4)
                if score > 0.4:
                    stack.append(child)
        
        return self._find_best_leaf(root)
    
    def _find_best_leaf(self, root: ThoughtNode) -> ThoughtNode:
        """Find the leaf node with the highest score."""
        best_node = root
        best_score = root.score
        
        def traverse(node):
            nonlocal best_node, best_score
            
            if not node.children and node.score > best_score:
                best_node = node
                best_score = node.score
            
            for child in node.children:
                traverse(child)
        
        traverse(root)
        return best_node
    
    def solve(
        self,
        problem: str,
        strategy: SearchStrategy = SearchStrategy.BFS,
        max_nodes: int = 20
    ) -> Dict[str, Any]:
        """
        Solve a problem using Tree-of-Thoughts.
        
        Args:
            problem: The problem to solve
            strategy: Search strategy to use
            max_nodes: Maximum nodes to explore
            
        Returns:
            Dictionary with solution path and statistics
        """
        print(f"\nSolving with {strategy.value} strategy...")
        
        if strategy == SearchStrategy.BFS:
            solution_node = self.search_bfs(problem, max_nodes)
        else:  # DFS
            solution_node = self.search_dfs(problem, max_nodes)
        
        if solution_node:
            path = solution_node.get_path()
            return {
                "problem": problem,
                "solution_found": solution_node.is_solution,
                "path": path,
                "final_score": solution_node.score,
                "depth": solution_node.depth,
                "strategy": strategy.value
            }
        
        return {
            "problem": problem,
            "solution_found": False,
            "path": [],
            "final_score": 0.0,
            "depth": 0,
            "strategy": strategy.value
        }


def demonstrate_tot_pattern():
    """Demonstrates the Tree-of-Thoughts pattern."""
    
    print("=" * 80)
    print("PATTERN 003: Tree-of-Thoughts (ToT)")
    print("=" * 80)
    print()
    
    # Create ToT agent
    agent = TreeOfThoughtsAgent(
        branching_factor=3,
        max_depth=3
    )
    
    # Test problems
    problems = [
        "Plan a surprise birthday party for a friend who loves nature and astronomy",
        "Design a simple mobile app to help people track their daily water intake"
    ]
    
    for i, problem in enumerate(problems, 1):
        print(f"\n{'=' * 80}")
        print(f"Problem {i}: {problem}")
        print('=' * 80)
        
        # Try BFS strategy
        try:
            result = agent.solve(
                problem,
                strategy=SearchStrategy.BFS,
                max_nodes=15
            )
            
            print(f"\n✓ Solution found: {result['solution_found']}")
            print(f"  Final score: {result['final_score']:.2f}")
            print(f"  Depth reached: {result['depth']}")
            print(f"\nThought path:")
            for j, thought in enumerate(result['path'], 1):
                print(f"  {j}. {thought}")
        
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
    
    # Summary
    print("\n\n" + "=" * 80)
    print("TREE-OF-THOUGHTS PATTERN DEMONSTRATION COMPLETE")
    print("=" * 80)
    print()
    print("Key Features Demonstrated:")
    print("1. Multiple thought generation at each step")
    print("2. Thought evaluation and scoring")
    print("3. Tree search strategies (BFS/DFS)")
    print("4. Path exploration and backtracking")
    print("5. Solution detection")
    print()
    print("Advantages:")
    print("- Explores multiple solution paths")
    print("- Can recover from poor initial choices")
    print("- Better solution quality through exploration")
    print("- Systematic search through problem space")
    print()
    print("LangChain Components Used:")
    print("- ChatPromptTemplate: Structures generation and evaluation prompts")
    print("- StrOutputParser: Parses LLM responses")
    print("- Custom search logic: Implements BFS/DFS")
    print("- State management: Tracks tree structure")
    print()


if __name__ == "__main__":
    # Check for API key
    if not os.getenv("OPENAI_API_KEY"):
        print("Error: OPENAI_API_KEY not found in environment variables.")
        print("Please set it in your .env file or environment.")
        exit(1)
    
    demonstrate_tot_pattern()

Agentic Design Pattern: Tree Of Thoughts

Pattern 003: Tree Of Thoughts – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Tree Of Thoughts pattern implementation.

Pattern Description

Pattern 003: Tree-of-Thoughts (ToT)

Description: Explores multiple reasoning paths simultaneously in a tree structure. ToT allows the model to generate multiple candidate thoughts at each step, evaluate them, and search through the tree (BFS or DFS) to find the best solution.

Components: – Thought Generation: Creates multiple candidate next steps – State Evaluation: Scores each thought/state – Search Strategy: BFS, DFS, or beam search – Backtracking: Can revisit and try alternative paths

Use Cases: – Strategic planning and game playing – Creative writing with multiple storylines – Complex problem-solving with multiple approaches – Optimization problems

LangChain Implementation: Uses LangGraph for state management and custom search logic for tree exploration.

File Statistics

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

Module Documentation

"""
Pattern 003: Tree-of-Thoughts (ToT)

Description:
    Explores multiple reasoning paths simultaneously in a tree structure.
    ToT allows the model to generate multiple candidate thoughts at each step,
    evaluate them, and search through the tree (BFS or DFS) to find the best solution.

Components:
    - Thought Generation: Creates multiple candidate next steps
    - State Evaluation: Scores each thought/state
    - Search Strategy: BFS, DFS, or beam search
    - Backtracking: Can revisit and try alternative paths

Use Cases:
    - Strategic planning and game playing
    - Creative writing with multiple storylines
    - Complex problem-solving with multiple approaches
    - Optimization problems

LangChain Implementation:
    Uses LangGraph for state management and custom search logic for tree exploration.
"""

Purpose: This module implements the Tree Of Thoughts pattern, providing a structured approach to agent design and implementation.

Import Statements

Standard Library Imports

import collections.deque
import dataclasses.dataclass
import dataclasses.field
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Tuple

Purpose: Core Python functionality: – collections.deque: Specialized container datatypes – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – 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.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: `SearchStrategy`

Lines 40-44

Inherits from: Enum

Purpose:

Search strategies for tree exploration.

Class: `ThoughtNode`

Lines 48-65

Purpose:

Represents a node in the tree of thoughts.

Methods:

`get_path(self)`

Lines 58-65

Purpose:

Returns the path from root to this node.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D179650>

Class: `TreeOfThoughtsAgent`

Lines 68-409

Purpose:

Agent that uses Tree-of-Thoughts reasoning.

Methods:

`init(self, model_name, temperature, branching_factor, max_depth)`

Lines 71-93

Purpose:

Initialize the Tree-of-Thoughts agent.

Args:
    model_name: LLM model to use
    temperature: Temperature for thought generation
    branching_factor: Number of thoughts to generate at each node
    max_depth: Maximum depth of the tree

Parameters: – model_name: Language model instance – temperature: Sampling temperature (0-1) – branching_factor: Branching factor – max_depth: Max depth

`generate_thoughts(self, problem, current_state, num_thoughts)`

Lines 95-147

Purpose:

Generate multiple candidate thoughts/next steps.

Args:
    problem: The original problem
    current_state: Current problem-solving state
    num_thoughts: Number of thoughts to generate

Returns:
    List of candidate thoughts

Parameters: – problem: Problem – current_state: Current state – num_thoughts: Num thoughts

Returns: <ast.Subscript object at 0x000001A06D16AFD0>

`evaluate_thought(self, problem, thought, current_state)`

Lines 149-190

Purpose:

Evaluate the quality/promise of a thought.

Args:
    problem: The original problem
    thought: The thought to evaluate
    current_state: Current state description

Returns:
    Score between 0 and 1

Parameters: – problem: Problem – thought: Thought – current_state: Current state

Returns: float

`is_solution(self, problem, current_state)`

Lines 192-220

Purpose:

Check if current state represents a solution.

Args:
    problem: The original problem
    current_state: Current state description

Returns:
    True if this is a valid solution

Parameters: – problem: Problem – current_state: Current state

Returns: bool

`search_bfs(self, problem, max_nodes)`

Lines 222-284

Purpose:

Breadth-first search through the tree of thoughts.

Args:
    problem: The problem to solve
    max_nodes: Maximum number of nodes to explore

Returns:
    Solution node if found, None otherwise

Parameters: – problem: Problem – max_nodes: Max nodes

Returns: <ast.Subscript object at 0x000001A06D15A6D0>

`search_dfs(self, problem, max_nodes)`

Lines 286-347

Purpose:

Depth-first search through the tree of thoughts.

Args:
    problem: The problem to solve
    max_nodes: Maximum number of nodes to explore

Returns:
    Solution node if found, None otherwise

Parameters: – problem: Problem – max_nodes: Max nodes

Returns: <ast.Subscript object at 0x000001A06D663ED0>

`_find_best_leaf(self, root)`

Lines 349-365

Purpose:

Find the leaf node with the highest score.

Parameters: – root: Root

Returns: ThoughtNode

`solve(self, problem, strategy, max_nodes)`

Lines 367-409

Purpose:

Solve a problem using Tree-of-Thoughts.

Args:
    problem: The problem to solve
    strategy: Search strategy to use
    max_nodes: Maximum nodes to explore

Returns:
    Dictionary with solution path and statistics

Parameters: – problem: Problem – strategy: Strategy – max_nodes: Max nodes

Returns: <ast.Subscript object at 0x000001A06D685210>

Functions

`demonstrate_tot_pattern()`

Lines 412-478

Purpose:

Demonstrates the Tree-of-Thoughts pattern.

Key Concepts Summary

Pattern Overview

The Tree Of Thoughts 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 002
Pattern 004
Other Reasoning Patterns (001-011)

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 Tree Of Thoughts 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: 003_tree_of_thoughts.py
Total Lines: 489
Pattern Category: AI Agent Design Pattern #003

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Agentic Design Pattern: Chain Of Thought


    """
Pattern 002: Chain-of-Thought (CoT)

Description:
    Agent breaks down complex problems into intermediate reasoning steps.
    CoT prompting encourages the model to explain its reasoning process
    step-by-step before arriving at a final answer.

Variants:
    - Zero-shot CoT: "Let's think step by step..."
    - Few-shot CoT: Provides examples with reasoning steps
    - Auto-CoT: Automatically generates reasoning examples

Use Cases:
    - Mathematical reasoning
    - Logical puzzles
    - Complex analysis tasks
    - Multi-step problem solving

LangChain Implementation:
    Uses structured prompts and output parsers to ensure step-by-step reasoning.
"""

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

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

# Load environment variables
load_dotenv()


# Few-shot examples demonstrating Chain-of-Thought reasoning
COT_EXAMPLES = [
    {
        "question": "Roger has 5 tennis balls. He buys 2 more cans of tennis balls. Each can has 3 tennis balls. How many tennis balls does he have now?",
        "reasoning": """Let me think step by step:
1. Roger starts with 5 tennis balls
2. He buys 2 cans of tennis balls
3. Each can contains 3 tennis balls
4. So he gets 2 × 3 = 6 new tennis balls
5. Total tennis balls = 5 (original) + 6 (new) = 11

Therefore, Roger has 11 tennis balls now.""",
        "answer": "11"
    },
    {
        "question": "If it takes 5 machines 5 minutes to make 5 widgets, how long would it take 100 machines to make 100 widgets?",
        "reasoning": """Let me think step by step:
1. 5 machines make 5 widgets in 5 minutes
2. This means each machine makes 1 widget in 5 minutes
3. The machines work in parallel (simultaneously)
4. So 100 machines working in parallel would still take 5 minutes
5. In 5 minutes, 100 machines would make 100 widgets (1 widget per machine)

Therefore, it would take 5 minutes.""",
        "answer": "5 minutes"
    }
]


class ChainOfThoughtAgent:
    """Agent that uses Chain-of-Thought reasoning."""
    
    def __init__(self, model_name: str = "gpt-3.5-turbo", temperature: float = 0):
        """Initialize the Chain-of-Thought agent."""
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=temperature,
            api_key=os.getenv("OPENAI_API_KEY")
        )
    
    def zero_shot_cot(self, question: str) -> Dict[str, str]:
        """
        Zero-shot Chain-of-Thought: Uses "Let's think step by step" prompt.
        
        Args:
            question: The question to answer
            
        Returns:
            Dictionary with reasoning and answer
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a helpful assistant that solves problems step-by-step."),
            ("human", "{question}\n\nLet's think step by step:")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        reasoning = chain.invoke({"question": question})
        
        # Extract final answer
        answer_prompt = ChatPromptTemplate.from_messages([
            ("system", "Based on the reasoning provided, give only the final answer concisely."),
            ("human", "Reasoning:\n{reasoning}\n\nFinal answer:")
        ])
        
        answer_chain = answer_prompt | self.llm | StrOutputParser()
        answer = answer_chain.invoke({"reasoning": reasoning})
        
        return {
            "question": question,
            "reasoning": reasoning,
            "answer": answer.strip()
        }
    
    def few_shot_cot(self, question: str, examples: List[Dict] = None) -> Dict[str, str]:
        """
        Few-shot Chain-of-Thought: Provides examples with reasoning steps.
        
        Args:
            question: The question to answer
            examples: List of example question-reasoning-answer triplets
            
        Returns:
            Dictionary with reasoning and answer
        """
        if examples is None:
            examples = COT_EXAMPLES
        
        # Create few-shot prompt template
        example_prompt = ChatPromptTemplate.from_messages([
            ("human", "{question}"),
            ("ai", "{reasoning}\n\nAnswer: {answer}")
        ])
        
        few_shot_prompt = FewShotChatMessagePromptTemplate(
            example_prompt=example_prompt,
            examples=examples
        )
        
        final_prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a helpful assistant that solves problems step-by-step. Show your reasoning process clearly."),
            few_shot_prompt,
            ("human", "{question}")
        ])
        
        chain = final_prompt | self.llm | StrOutputParser()
        
        response = chain.invoke({"question": question})
        
        # Parse response to separate reasoning and answer
        lines = response.strip().split('\n')
        answer_line = [line for line in lines if line.strip().startswith("Answer:")]
        
        if answer_line:
            answer = answer_line[0].replace("Answer:", "").strip()
            reasoning = response.replace(answer_line[0], "").strip()
        else:
            # If no explicit answer line, last line is the answer
            reasoning = '\n'.join(lines[:-1])
            answer = lines[-1].strip()
        
        return {
            "question": question,
            "reasoning": reasoning,
            "answer": answer
        }
    
    def auto_cot(self, question: str) -> Dict[str, str]:
        """
        Auto-CoT: Automatically generates reasoning with structured steps.
        
        Args:
            question: The question to answer
            
        Returns:
            Dictionary with reasoning and answer
        """
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a helpful assistant that solves problems systematically.
For each problem:
1. Break it down into clear steps
2. Number each step
3. Show your work clearly
4. Provide a final answer

Format your response as:
Step 1: [description]
Step 2: [description]
...
Final Answer: [answer]"""),
            ("human", "{question}")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        response = chain.invoke({"question": question})
        
        # Parse response
        lines = response.strip().split('\n')
        final_answer_line = [line for line in lines if "final answer" in line.lower()]
        
        if final_answer_line:
            answer = final_answer_line[0].split(':', 1)[-1].strip()
            reasoning = '\n'.join([line for line in lines if line not in final_answer_line])
        else:
            reasoning = response
            answer = "Unable to extract final answer"
        
        return {
            "question": question,
            "reasoning": reasoning,
            "answer": answer
        }


def demonstrate_cot_pattern():
    """Demonstrates the Chain-of-Thought pattern with various approaches."""
    
    print("=" * 80)
    print("PATTERN 002: Chain-of-Thought (CoT)")
    print("=" * 80)
    print()
    
    # Create CoT agent
    agent = ChainOfThoughtAgent()
    
    # Test questions
    test_questions = [
        "A farmer has 17 sheep and all but 9 die. How many sheep are left?",
        "If a train travels 120 km in 2 hours, how far will it travel in 5 hours at the same speed?",
        "Lisa has 10 apples. She gives 3 to her friend and buys 5 more. Her friend gives back 1 apple. How many apples does Lisa have now?"
    ]
    
    # Demonstrate Zero-shot CoT
    print("\n" + "=" * 80)
    print("ZERO-SHOT CHAIN-OF-THOUGHT")
    print("=" * 80)
    
    for i, question in enumerate(test_questions[:2], 1):
        print(f"\n{'- ' * 40}")
        print(f"Question {i}: {question}")
        print('- ' * 40)
        
        try:
            result = agent.zero_shot_cot(question)
            print(f"\nReasoning:\n{result['reasoning']}")
            print(f"\n✓ Answer: {result['answer']}")
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
    
    # Demonstrate Few-shot CoT
    print("\n\n" + "=" * 80)
    print("FEW-SHOT CHAIN-OF-THOUGHT")
    print("=" * 80)
    
    for i, question in enumerate(test_questions[1:], 1):
        print(f"\n{'- ' * 40}")
        print(f"Question {i}: {question}")
        print('- ' * 40)
        
        try:
            result = agent.few_shot_cot(question)
            print(f"\nReasoning:\n{result['reasoning']}")
            print(f"\n✓ Answer: {result['answer']}")
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
    
    # Demonstrate Auto-CoT
    print("\n\n" + "=" * 80)
    print("AUTO-COT (STRUCTURED STEPS)")
    print("=" * 80)
    
    question = test_questions[2]
    print(f"\n{'- ' * 40}")
    print(f"Question: {question}")
    print('- ' * 40)
    
    try:
        result = agent.auto_cot(question)
        print(f"\nReasoning:\n{result['reasoning']}")
        print(f"\n✓ Answer: {result['answer']}")
    except Exception as e:
        print(f"\n✗ Error: {str(e)}")
    
    # Summary
    print("\n\n" + "=" * 80)
    print("CHAIN-OF-THOUGHT PATTERN DEMONSTRATION COMPLETE")
    print("=" * 80)
    print()
    print("CoT Variants Demonstrated:")
    print("1. Zero-shot CoT: 'Let's think step by step' prompting")
    print("2. Few-shot CoT: Examples with reasoning provided")
    print("3. Auto-CoT: Structured step-by-step format")
    print()
    print("Key Benefits:")
    print("- Improved accuracy on complex tasks")
    print("- Transparent reasoning process")
    print("- Better error detection")
    print("- Enhanced interpretability")
    print()
    print("LangChain Components Used:")
    print("- ChatPromptTemplate: Structures prompts")
    print("- FewShotChatMessagePromptTemplate: Manages few-shot examples")
    print("- StrOutputParser: Parses LLM outputs")
    print("- Chain composition: Connects components")
    print()


if __name__ == "__main__":
    # Check for API key
    if not os.getenv("OPENAI_API_KEY"):
        print("Error: OPENAI_API_KEY not found in environment variables.")
        print("Please set it in your .env file or environment.")
        exit(1)
    
    demonstrate_cot_pattern()

Agentic Design Pattern: Chain Of Thought

Pattern 002: Chain Of Thought – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Chain Of Thought pattern implementation.

Pattern Description

Pattern 002: Chain-of-Thought (CoT)

Description: Agent breaks down complex problems into intermediate reasoning steps. CoT prompting encourages the model to explain its reasoning process step-by-step before arriving at a final answer.

Variants: – Zero-shot CoT: “Let’s think step by step…” – Few-shot CoT: Provides examples with reasoning steps – Auto-CoT: Automatically generates reasoning examples

Use Cases: – Mathematical reasoning – Logical puzzles – Complex analysis tasks – Multi-step problem solving

LangChain Implementation: Uses structured prompts and output parsers to ensure step-by-step reasoning.

File Statistics

Total Lines: 313
Classes: 1
Functions: 1
Constants: 1

Module Documentation

"""
Pattern 002: Chain-of-Thought (CoT)

Description:
    Agent breaks down complex problems into intermediate reasoning steps.
    CoT prompting encourages the model to explain its reasoning process
    step-by-step before arriving at a final answer.

Variants:
    - Zero-shot CoT: "Let's think step by step..."
    - Few-shot CoT: Provides examples with reasoning steps
    - Auto-CoT: Automatically generates reasoning examples

Use Cases:
    - Mathematical reasoning
    - Logical puzzles
    - Complex analysis tasks
    - Multi-step problem solving

LangChain Implementation:
    Uses structured prompts and output parsers to ensure step-by-step reasoning.
"""

Purpose: This module implements the Chain Of Thought pattern, providing a structured approach to agent design and implementation.

Import Statements

Standard Library Imports

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

Purpose: Core Python functionality: – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.prompts.FewShotChatMessagePromptTemplate: Regular expression operations – os: Environment variable access and system operations – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain.chains.LLMChain
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.

Module Constants

COT_EXAMPLES

COT_EXAMPLES = [

Line 38: Module-level constant definition.

Classes

Class: `ChainOfThoughtAgent`

Lines 66-209

Purpose:

Agent that uses Chain-of-Thought reasoning.

Methods:

`init(self, model_name, temperature)`

Lines 69-75

Purpose:

Initialize the Chain-of-Thought agent.

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

`zero_shot_cot(self, question)`

Lines 77-109

Purpose:

Zero-shot Chain-of-Thought: Uses "Let's think step by step" prompt.

Args:
    question: The question to answer

Returns:
    Dictionary with reasoning and answer

Parameters: – question: Question to answer

Returns: <ast.Subscript object at 0x000001A06D60E590>

`few_shot_cot(self, question, examples)`

Lines 111-162

Purpose:

Few-shot Chain-of-Thought: Provides examples with reasoning steps.

Args:
    question: The question to answer
    examples: List of example question-reasoning-answer triplets

Returns:
    Dictionary with reasoning and answer

Parameters: – question: Question to answer – examples: Examples

Returns: <ast.Subscript object at 0x000001A06D608810>

`auto_cot(self, question)`

Lines 164-209

Purpose:

Auto-CoT: Automatically generates reasoning with structured steps.

Args:
    question: The question to answer

Returns:
    Dictionary with reasoning and answer

Parameters: – question: Question to answer

Returns: <ast.Subscript object at 0x000001A06D5F7F50>

Functions

`demonstrate_cot_pattern()`

Lines 212-302

Purpose:

Demonstrates the Chain-of-Thought pattern with various approaches.

Key Concepts Summary

Pattern Overview

The Chain Of Thought 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 001
Pattern 003
Other Reasoning Patterns (001-011)

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 Chain Of Thought 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: 002_chain_of_thought.py
Total Lines: 313
Pattern Category: AI Agent Design Pattern #002

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Agentic Design Pattern: React


    """
Pattern 001: ReAct (Reasoning + Acting)

Description:
    Agent alternates between reasoning about the task and taking actions.
    The ReAct pattern combines reasoning traces with task-specific actions,
    allowing the agent to reason about what to do and then take actions.

Components:
    - Thought: Agent reasons about current state
    - Action: Agent takes an action (e.g., tool use)
    - Observation: Agent observes the result
    - Loop continues until task is complete

Use Cases:
    - Question answering with external knowledge
    - Task completion requiring multiple steps
    - Interactive problem solving

LangChain Implementation:
    Uses LangChain's AgentExecutor with structured tools and ReAct prompt template.
"""

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

from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import Tool
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# Load environment variables
load_dotenv()


# Define example tools
def calculator(expression: str) -> str:
    """Evaluates a mathematical expression."""
    try:
        # Safe evaluation for demo purposes
        result = eval(expression, {"__builtins__": {}}, {})
        return f"Result: {result}"
    except Exception as e:
        return f"Error: {str(e)}"


def search_tool(query: str) -> str:
    """Simulates a search tool that retrieves information."""
    # Simulated search results
    knowledge_base = {
        "capital of france": "Paris",
        "population of tokyo": "approximately 14 million",
        "height of mount everest": "8,848.86 meters (29,031.7 feet)",
        "speed of light": "299,792,458 meters per second",
        "boiling point of water": "100°C (212°F) at sea level",
    }
    
    query_lower = query.lower()
    for key, value in knowledge_base.items():
        if key in query_lower:
            return f"Search result: {value}"
    
    return f"Search result: Information about '{query}' not found in knowledge base."


def get_current_date() -> str:
    """Returns the current date."""
    from datetime import datetime
    return f"Current date: {datetime.now().strftime('%Y-%m-%d')}"


# Create tools list
tools = [
    Tool(
        name="Calculator",
        func=calculator,
        description="Useful for performing mathematical calculations. Input should be a valid mathematical expression."
    ),
    Tool(
        name="Search",
        func=search_tool,
        description="Useful for searching factual information. Input should be a search query."
    ),
    Tool(
        name="GetDate",
        func=get_current_date,
        description="Returns the current date. No input required."
    )
]


# ReAct prompt template
REACT_PROMPT = """Answer the following questions as best you can. You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {input}
Thought: {agent_scratchpad}
"""


def create_react_agent_system():
    """Creates a ReAct agent using LangChain."""
    
    # Initialize LLM
    llm = ChatOpenAI(
        model="gpt-3.5-turbo",
        temperature=0,
        api_key=os.getenv("OPENAI_API_KEY")
    )
    
    # Create prompt template
    prompt = PromptTemplate.from_template(REACT_PROMPT)
    
    # Create agent
    agent = create_react_agent(
        llm=llm,
        tools=tools,
        prompt=prompt
    )
    
    # Create agent executor
    agent_executor = AgentExecutor(
        agent=agent,
        tools=tools,
        verbose=True,
        max_iterations=10,
        handle_parsing_errors=True
    )
    
    return agent_executor


def demonstrate_react_pattern():
    """Demonstrates the ReAct pattern with various examples."""
    
    print("=" * 80)
    print("PATTERN 001: ReAct (Reasoning + Acting)")
    print("=" * 80)
    print()
    
    # Create the ReAct agent
    agent_executor = create_react_agent_system()
    
    # Test cases demonstrating ReAct pattern
    test_questions = [
        "What is the capital of France?",
        "What is 25 * 4 + 10?",
        "What is the population of Tokyo and what is 100 divided by that number in millions?",
        "What's today's date?",
    ]
    
    for i, question in enumerate(test_questions, 1):
        print(f"\n{'=' * 80}")
        print(f"Example {i}: {question}")
        print('=' * 80)
        
        try:
            result = agent_executor.invoke({"input": question})
            print(f"\n✓ Final Answer: {result['output']}")
        except Exception as e:
            print(f"\n✗ Error: {str(e)}")
        
        print()
    
    # Summary
    print("\n" + "=" * 80)
    print("REACT PATTERN DEMONSTRATION COMPLETE")
    print("=" * 80)
    print()
    print("Key Features Demonstrated:")
    print("1. Thought-Action-Observation loop")
    print("2. Tool selection and usage")
    print("3. Multi-step reasoning")
    print("4. Dynamic decision making")
    print("5. Interpretable reasoning traces")
    print()
    print("LangChain Components Used:")
    print("- AgentExecutor: Manages agent execution loop")
    print("- create_react_agent: Creates ReAct-style agent")
    print("- Tool: Wraps functions as agent tools")
    print("- PromptTemplate: Structures the ReAct prompt format")
    print("- ChatOpenAI: LLM for reasoning and action selection")
    print()


if __name__ == "__main__":
    # Check for API key
    if not os.getenv("OPENAI_API_KEY"):
        print("Error: OPENAI_API_KEY not found in environment variables.")
        print("Please set it in your .env file or environment.")
        exit(1)
    
    demonstrate_react_pattern()

Agentic Design Pattern: React

Pattern 001: React – Line-by-Line Explanation

Overview

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

Pattern Description

Pattern 001: ReAct (Reasoning + Acting)

Description: Agent alternates between reasoning about the task and taking actions. The ReAct pattern combines reasoning traces with task-specific actions, allowing the agent to reason about what to do and then take actions.

Components: – Thought: Agent reasons about current state – Action: Agent takes an action (e.g., tool use) – Observation: Agent observes the result – Loop continues until task is complete

Use Cases: – Question answering with external knowledge – Task completion requiring multiple steps – Interactive problem solving

LangChain Implementation: Uses LangChain’s AgentExecutor with structured tools and ReAct prompt template.

File Statistics

Total Lines: 209
Classes: 0
Functions: 5
Constants: 1

Module Documentation

"""
Pattern 001: ReAct (Reasoning + Acting)

Description:
    Agent alternates between reasoning about the task and taking actions.
    The ReAct pattern combines reasoning traces with task-specific actions,
    allowing the agent to reason about what to do and then take actions.

Components:
    - Thought: Agent reasons about current state
    - Action: Agent takes an action (e.g., tool use)
    - Observation: Agent observes the result
    - Loop continues until task is complete

Use Cases:
    - Question answering with external knowledge
    - Task completion requiring multiple steps
    - Interactive problem solving

LangChain Implementation:
    Uses LangChain's AgentExecutor with structured tools and ReAct prompt template.
"""

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

Import Statements

Standard Library Imports

import datetime.datetime
import langchain.agents.create_react_agent
import langchain_core.prompts.PromptTemplate
import os
import typing.List
import typing.Optional

Purpose: Core Python functionality: – datetime.datetime: Date and time handling – langchain.agents.create_react_agent: Regular expression operations – langchain_core.prompts.PromptTemplate: Regular expression operations – os: Environment variable access and system operations – 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.agents.AgentExecutor
from langchain.tools.Tool
from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows – Agents: Autonomous decision-making entities – Tools: Functions that agents can invoke

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

Module Constants

REACT_PROMPT

REACT_PROMPT = """Answer the following questions as best you can. You have access to the following tools:

Line 94: Module-level constant definition.

Functions

`calculator(expression)`

Lines 38-45

Purpose:

Evaluates a mathematical expression.

Parameters: – expression: Expression

Returns: str

`search_tool(query)`

Lines 48-64

Purpose:

Simulates a search tool that retrieves information.

Parameters: – query: Input query or question

Returns: str

`get_current_date()`

Lines 67-70

Purpose:

Returns the current date.

Returns: str

`create_react_agent_system()`

Lines 116-145

Purpose:

Creates a ReAct agent using LangChain.

`demonstrate_react_pattern()`

Lines 148-198

Purpose:

Demonstrates the ReAct pattern with various examples.

Key Concepts Summary

Pattern Overview

The React 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
Agents: Autonomous decision-making
Tools: External function calling

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 002
Other Reasoning Patterns (001-011)

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 React 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: 001_react.py
Total Lines: 209
Pattern Category: AI Agent Design Pattern #001

Generated from actual implementation code
For educational and reference purposes

November 2, 2025

Author: admin

Agentic Design Pattern: Constitutional Ai

Agentic Design Pattern: Constitutional Ai

Pattern 010: Constitutional Ai – Line-by-Line Explanation

Overview

Pattern Description

File Statistics

Module Documentation

Import Statements

Standard Library Imports

LangChain Imports

Other Imports

Classes

Class: Principle

Class: ViolationReport

Class: ConstitutionalReview

calculate_severity(self)

Class: Constitution

__init__(self, name)

add_principle(self, principle)

get_principles_text(self)

get_default()

Class: ConstitutionalAI

__init__(self, constitution, model_name, max_revisions, severity_threshold)

generate_initial(self, query, context)

critique_against_principle(self, output, query, principle)

review(self, output, query)

revise(self, query, output, review)

generate_with_constitution(self, query, context, verbose)

Functions

demonstrate_constitutional_ai()

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: Self Refine

Agentic Design Pattern: Self Refine

Pattern 009: Self Refine – Line-by-Line Explanation

Overview

Pattern Description

File Statistics

Module Documentation

Import Statements

Standard Library Imports

LangChain Imports

Other Imports

Classes

Class: IterationStatus

Class: Critique

Class: RefinementIteration

__str__(self)

Class: SelfRefineAgent

__init__(self, model_name, max_iterations, quality_threshold)

generate_initial(self, task, context)

critique(self, task, output, iteration)

Class: `Principle`

Class: `ViolationReport`

Class: `ConstitutionalReview`

`calculate_severity(self)`

Class: `Constitution`

`init(self, name)`

`add_principle(self, principle)`

`get_principles_text(self)`

`get_default()`

Class: `ConstitutionalAI`

`init(self, constitution, model_name, max_revisions, severity_threshold)`

`generate_initial(self, query, context)`

`critique_against_principle(self, output, query, principle)`

`review(self, output, query)`

`revise(self, query, output, review)`

`generate_with_constitution(self, query, context, verbose)`

`demonstrate_constitutional_ai()`

Class: `IterationStatus`

Class: `Critique`

Class: `RefinementIteration`

`str(self)`

Class: `SelfRefineAgent`

`init(self, model_name, max_iterations, quality_threshold)`

`generate_initial(self, task, context)`

`critique(self, task, output, iteration)`

`refine(self, task, current_output, critique)`

`refine_iteratively(self, task, context, verbose)`

`get_improvement_summary(self)`

`demonstrate_self_refine()`

Class: `TaskStatus`

Class: `Task`

`can_execute(self, completed_tasks)`

Class: `ExecutionPlan`

`add_task(self, task)`

`get_ready_tasks(self, completed)`

`topological_sort(self)`

Class: `LLMCompiler`

`init(self, model_name, max_parallel)`

`parse_task(self, goal)`

`analyze_dependencies(self, plan)`

`optimize_plan(self, plan)`

`execute_task(self, task, plan)`

`execute_plan(self, plan)`

`_synthesize_results(self, plan)`

`compile_and_execute(self, goal)`

`demonstrate_llm_compiler()`