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

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

Agentic Design Pattern: Graph Of Thoughts

Agentic Design Pattern: Graph Of Thoughts

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

Overview

Pattern Description

File Statistics

Module Documentation

Import Statements

Standard Library Imports

LangChain Imports

Other Imports

Classes

Class: EdgeType

Class: ThoughtNode

__hash__(self)

__eq__(self, other)

Class: ThoughtEdge

Class: ThoughtGraph

__init__(self)

add_node(self, node)

add_edge(self, edge)

get_node(self, node_id)

get_children(self, node_id)

get_parents(self, node_id)

get_all_paths(self, start_id, end_id)

Class: GraphOfThoughtsAgent

__init__(self, model_name, temperature)

generate_initial_thoughts(self, problem, num_thoughts)

transform_thought(self, node, transformation)

aggregate_thoughts(self, nodes, aggregation_type)

refine_thought(self, node, feedback)

evaluate_thought(self, node, problem)

solve_with_graph(self, problem, max_iterations)

_count_node_types(self)

visualize_graph(self)

Functions

demonstrate_got_pattern()

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

Comments

Leave a Reply Cancel reply

More posts

4D AI models

Biggest AI upgrades in 2026

Tools for Architect

Quarkus in Java

Class: `EdgeType`

Class: `ThoughtNode`

`hash(self)`

`eq(self, other)`

Class: `ThoughtEdge`

Class: `ThoughtGraph`

`init(self)`

`add_node(self, node)`

`add_edge(self, edge)`

`get_node(self, node_id)`

`get_children(self, node_id)`

`get_parents(self, node_id)`

`get_all_paths(self, start_id, end_id)`

Class: `GraphOfThoughtsAgent`

`init(self, model_name, temperature)`

`generate_initial_thoughts(self, problem, num_thoughts)`

`transform_thought(self, node, transformation)`

`aggregate_thoughts(self, nodes, aggregation_type)`

`refine_thought(self, node, feedback)`

`evaluate_thought(self, node, problem)`

`solve_with_graph(self, problem, max_iterations)`

`_count_node_types(self)`

`visualize_graph(self)`

`demonstrate_got_pattern()`