
    """
Pattern 76: Graph-of-Thought Pattern

Reasoning as a graph structure with nodes (thoughts) and edges (relationships).
Allows parallel exploration, backtracking, and aggregation of multiple reasoning paths.
"""

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict


class ThoughtType(Enum):
    """Type of thought node."""
    HYPOTHESIS = "hypothesis"
    EVIDENCE = "evidence"
    REASONING = "reasoning"
    CONCLUSION = "conclusion"
    QUESTION = "question"


@dataclass
class ThoughtNode:
    """Node in the thought graph."""
    node_id: str
    content: str
    thought_type: ThoughtType
    confidence: float = 0.5  # 0.0 to 1.0
    parent_ids: List[str] = field(default_factory=list)
    child_ids: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class ThoughtEdge:
    """Edge connecting thoughts."""
    from_id: str
    to_id: str
    relationship: str  # supports, contradicts, questions, leads_to
    strength: float = 0.5  # 0.0 to 1.0


class GraphOfThoughtState(TypedDict):
    """State for graph-of-thought workflow."""
    problem: str
    nodes: List[ThoughtNode]
    edges: List[ThoughtEdge]
    current_node_id: Optional[str]
    explored_ids: Set[str]
    conclusions: List[str]


class GraphOfThoughtPattern:
    """
    Graph-of-Thought Pattern - Non-linear reasoning with graph structure.
    
    Thoughts are organized as a graph allowing:
    - Parallel exploration of multiple paths
    - Backtracking when needed
    - Aggregation of insights from different branches
    - Complex relationship modeling
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.graph = StateGraph(GraphOfThoughtState)
        self._build_graph()
        self.workflow = self.graph.compile()
    
    def _build_graph(self):
        """Build the LangGraph workflow."""
        self.graph.add_node("generate_hypotheses", self._generate_hypotheses)
        self.graph.add_node("explore_path", self._explore_path)
        self.graph.add_node("evaluate_node", self._evaluate_node)
        self.graph.add_node("aggregate_insights", self._aggregate_insights)
        
        self.graph.set_entry_point("generate_hypotheses")
        
        self.graph.add_edge("generate_hypotheses", "explore_path")
        self.graph.add_edge("explore_path", "evaluate_node")
        
        self.graph.add_conditional_edges(
            "evaluate_node",
            self._should_continue_exploration,
            {
                "explore": "explore_path",
                "aggregate": "aggregate_insights",
            }
        )
        
        self.graph.add_edge("aggregate_insights", END)
    
    def _generate_hypotheses(self, state: GraphOfThoughtState) -> GraphOfThoughtState:
        """Generate initial hypothesis nodes."""
        print(f"\n   Generating hypotheses for: {state['problem']}")
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Generate 3 diverse hypotheses or approaches to solve this problem. "
                      "Each should explore a different angle. Return one per line."),
            ("human", "{problem}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"problem": state['problem']})
        
        hypotheses = [h.strip() for h in response.content.split('\n') if h.strip()][:3]
        
        nodes = []
        for i, hypothesis in enumerate(hypotheses):
            node = ThoughtNode(
                node_id=f"hyp_{i}",
                content=hypothesis,
                thought_type=ThoughtType.HYPOTHESIS,
                confidence=0.5
            )
            nodes.append(node)
            print(f"   - Hypothesis {i+1}: {hypothesis[:60]}...")
        
        state["nodes"] = nodes
        state["edges"] = []
        state["explored_ids"] = set()
        state["conclusions"] = []
        
        return state
    
    def _explore_path(self, state: GraphOfThoughtState) -> GraphOfThoughtState:
        """Explore a thought path by expanding a node."""
        # Find unexplored node
        unexplored = [n for n in state["nodes"] 
                     if n.node_id not in state["explored_ids"]]
        
        if not unexplored:
            return state
        
        current_node = unexplored[0]
        state["current_node_id"] = current_node.node_id
        
        print(f"\n   Exploring: {current_node.content[:60]}...")
        
        # Generate reasoning from this node
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Based on this thought, generate 2 follow-up thoughts: "
                      "either supporting evidence, counterarguments, or logical next steps. "
                      "Return one per line."),
            ("human", "{thought}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"thought": current_node.content})
        
        follow_ups = [f.strip() for f in response.content.split('\n') if f.strip()][:2]
        
        for i, follow_up in enumerate(follow_ups):
            child_id = f"{current_node.node_id}_child_{i}"
            child_node = ThoughtNode(
                node_id=child_id,
                content=follow_up,
                thought_type=ThoughtType.REASONING,
                parent_ids=[current_node.node_id],
                confidence=0.6
            )
            state["nodes"].append(child_node)
            current_node.child_ids.append(child_id)
            
            # Create edge
            edge = ThoughtEdge(
                from_id=current_node.node_id,
                to_id=child_id,
                relationship="leads_to",
                strength=0.7
            )
            state["edges"].append(edge)
            
            print(f"   - Follow-up: {follow_up[:50]}...")
        
        state["explored_ids"].add(current_node.node_id)
        
        return state
    
    def _evaluate_node(self, state: GraphOfThoughtState) -> GraphOfThoughtState:
        """Evaluate current node's confidence."""
        current_id = state.get("current_node_id")
        if not current_id:
            return state
        
        current_node = next((n for n in state["nodes"] if n.node_id == current_id), None)
        if not current_node:
            return state
        
        # Simple evaluation based on depth and branching
        depth = len([p for p in current_node.parent_ids])
        children_count = len(current_node.child_ids)
        
        # Update confidence
        current_node.confidence = min(1.0, 0.5 + (children_count * 0.1) + (depth * 0.05))
        
        return state
    
    def _should_continue_exploration(self, state: GraphOfThoughtState) -> str:
        """Decide whether to continue exploring or aggregate."""
        explored = len(state["explored_ids"])
        total = len(state["nodes"])
        
        # Explore at least 5 nodes or until 80% explored
        if explored  0 and explored / total  GraphOfThoughtState:
        """Aggregate insights from the thought graph."""
        print(f"\n   Aggregating insights from {len(state['nodes'])} thoughts...")
        
        # Find leaf nodes (conclusions)
        leaf_nodes = [n for n in state["nodes"] if not n.child_ids]
        
        # Prepare summary of reasoning paths
        paths_summary = []
        for leaf in leaf_nodes[:3]:  # Top 3 paths
            path = self._trace_path(leaf, state["nodes"])
            paths_summary.append(" → ".join([n.content[:40] for n in path]))
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Given these reasoning paths, synthesize a final conclusion. "
                      "Be comprehensive and consider all perspectives."),
            ("human", "Problem: {problem}\n\nReasoning paths:\n{paths}\n\nProvide final conclusion:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "problem": state['problem'],
            "paths": "\n".join([f"{i+1}. {p}" for i, p in enumerate(paths_summary)])
        })
        
        state["conclusions"] = [response.content]
        print(f"   - Synthesized conclusion")
        
        return state
    
    def _trace_path(self, node: ThoughtNode, all_nodes: List[ThoughtNode]) -> List[ThoughtNode]:
        """Trace path from root to this node."""
        path = [node]
        current = node
        
        while current.parent_ids:
            parent_id = current.parent_ids[0]
            parent = next((n for n in all_nodes if n.node_id == parent_id), None)
            if not parent:
                break
            path.insert(0, parent)
            current = parent
        
        return path
    
    def run(self, problem: str) -> Dict[str, Any]:
        """Execute graph-of-thought reasoning."""
        print(f"\n=== Graph-of-Thought Pattern ===")
        
        initial_state: GraphOfThoughtState = {
            "problem": problem,
            "nodes": [],
            "edges": [],
            "current_node_id": None,
            "explored_ids": set(),
            "conclusions": []
        }
        
        final_state = self.workflow.invoke(initial_state)
        
        return {
            "pattern": "GraphOfThought",
            "problem": problem,
            "total_nodes": len(final_state["nodes"]),
            "total_edges": len(final_state["edges"]),
            "explored_nodes": len(final_state["explored_ids"]),
            "conclusions": final_state["conclusions"],
            "graph_structure": {
                "nodes": len(final_state["nodes"]),
                "edges": len(final_state["edges"]),
                "hypotheses": len([n for n in final_state["nodes"] 
                                  if n.thought_type == ThoughtType.HYPOTHESIS])
            }
        }


# Example usage
if __name__ == "__main__":
    print("=== Graph-of-Thought Pattern Demo ===\n")
    
    pattern = GraphOfThoughtPattern()
    
    # Complex problem requiring non-linear reasoning
    problem = """
    A company wants to reduce its carbon footprint by 50% in 2 years.
    They have manufacturing plants, offices, and a delivery fleet.
    What strategy should they pursue?
    """
    
    print("Problem:", problem.strip())
    
    result = pattern.run(problem)
    
    print(f"\n=== Results ===")
    print(f"Total thought nodes: {result['total_nodes']}")
    print(f"Explored nodes: {result['explored_nodes']}")
    print(f"Graph edges: {result['total_edges']}")
    print(f"Hypotheses generated: {result['graph_structure']['hypotheses']}")
    
    print(f"\n=== Final Conclusion ===")
    for conclusion in result['conclusions']:
        print(f"{conclusion}")
    
    print("\n✓ Graph-of-thought reasoning completed with non-linear exploration!")

Agentic Architecture Pattern: Graph Of Thought

Pattern 76: 76 Graph Of Thought – Line-by-Line Explanation

Overview

Pattern 76: Graph-of-Thought Pattern

Reasoning as a graph structure with nodes (thoughts) and edges (relationships). Allows parallel exploration, backtracking, and aggregation of multiple reasoning paths.

Code Structure Analysis

File Information

Pattern Number: 76
Pattern Name: 76 Graph Of Thought
File Path: patterns\execution_control\76_graph_of_thought.py
Total Lines: 319

Detailed Code Explanation

Module Docstring

"""
Pattern 76: Graph-of-Thought Pattern

Reasoning as a graph structure with nodes (thoughts) and edges (relationships).
Allows parallel exploration, backtracking, and aggregation of multiple reasoning paths.
"""

Explanation: This module docstring describes the purpose and key features of the 76 Graph Of Thought pattern.

Imports Section

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Set: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration – from langgraph.graph import StateGraph, END: LangGraph for workflow orchestration – from typing_extensions import TypedDict: Type hints for better code documentation

Enumerations

ThoughtType (Line 18)

class ThoughtType(Enum):
    HYPOTHESIS
    EVIDENCE
    REASONING
    CONCLUSION
    QUESTION

Purpose: Type of thought node.

Values: – HYPOTHESIS: Represents hypothesis – EVIDENCE: Represents evidence – REASONING: Represents reasoning – CONCLUSION: Represents conclusion – QUESTION: Represents question

Classes

ThoughtType (Line 18)

Description: Type of thought node.

ThoughtNode (Line 28)

Description: Node in the thought graph.

ThoughtEdge (Line 41)

Description: Edge connecting thoughts.

GraphOfThoughtState (Line 49)

Description: State for graph-of-thought workflow.

GraphOfThoughtPattern (Line 59)

Description: Graph-of-Thought Pattern – Non-linear reasoning with graph structure.

Thoughts are organized as a graph allowing: – Parallel exploration of multiple paths – Backtracking when needed – Aggregation of insights from different branches – Complex relationship modeling

Methods:

`init` (Line 70)

def __init__(self, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – model_name: LLM model to use (e.g., “gpt-4”)

`_build_graph` (Line 76)

def _build_graph(self):
    """Build the LangGraph workflow."""

Purpose: Build the LangGraph workflow.

Parameters: – self: Instance reference