Author: admin

  • Agentic Orchestration Pattern: Context Passing Pattern

    
        """
    Pattern 08.02: Context Passing
    Shared context between agents
    """
    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    
    
    class ContextPassingState(TypedDict):
        """State for Context Passing pattern."""
        input: str
        output: str
    
    
    def agent_function(state: ContextPassingState) -> ContextPassingState:
        """Context Passing agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using Context Passing pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_context_passing():
        """Create Context Passing workflow."""
        workflow = StateGraph(ContextPassingState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
        workflow = create_context_passing()
        result = workflow.invoke({"input": "Test Context Passing", "output": ""})
        print(f"\n=== Context Passing Pattern ===\nOutput: {result['output'][:200]}...")
    
        

    Agentic Orchestration Pattern: Context Passing Pattern

    02 Context Passing Pattern – Detailed Documentation

    Overview

    File: 08_state_management/02_context_passing.py
    Pattern Category: 08 State Management
    Implementation: LangGraph/CrewAI

    Pattern Description

    Pattern 08.02: Context Passing Shared context between agents


    Code Structure

    Imports

    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    

    Purpose of imports: – TypedDict: For type-safe state definitions – StateGraph: LangGraph’s graph-based workflow orchestration – ChatOpenAI: OpenAI LLM integration – HumanMessage: LangChain message formatting

    State Definition

    class ContextPassingState(TypedDict):
        """State for Context Passing pattern."""
        input: str
        output: str
    
    
    def agent_function(state: ContextPassingState) -> ContextPassingState:
    

    ContextPassingState: State for Context Passing pattern.

    State Fields: – input: Stores input – output: Stores output – def agent_function(state: Stores def agent function(state – prompt = f"Process this using Context Passing pattern: Stores prompt = f”Process this using Context Passing pattern – def create_context_passing(): Stores def create context passing()


    Functions

    agent_function()

    Line: 17
    Arguments: state

    Context Passing agent implementation.

    def agent_function(state: ContextPassingState) -> ContextPassingState:
        """Context Passing agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using Context Passing pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_context_passing():
    

    create_context_passing()

    Line: 26
    Arguments: “

    Create Context Passing workflow.

    def create_context_passing():
        """Create Context Passing workflow."""
        workflow = StateGraph(ContextPassingState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
    

    Line-by-Line Explanation

    Line 5: Import statement

    from typing import TypedDict
    

    Imports TypedDict for creating type-safe state dictionaries.

    Line 6: Import statement

    from langgraph.graph import StateGraph, END
    

    Imports StateGraph – the core LangGraph class for building workflows.

    Line 7: Import statement

    from langchain_openai import ChatOpenAI
    

    Imports ChatOpenAI for LLM interactions.

    Line 8: Import statement

    from langchain_core.messages import HumanMessage
    

    Imports HumanMessage for formatting prompts to the LLM.

    Line 11: Class definition

    class ContextPassingState(TypedDict):
    

    Defines the state structure using TypedDict for type safety.

    Line 12: Module docstring

        """State for Context Passing pattern."""
    

    Describes the pattern and its purpose.

    Line 17: Function agent_function definition

    def agent_function(state: ContextPassingState) -> ContextPassingState:
    

    Defines the agent_function function that processes the workflow state.

    Line 18: Module docstring

        """Context Passing agent implementation."""
    

    Describes the pattern and its purpose.

    Line 26: Function create_context_passing definition

    def create_context_passing():
    

    Defines the create_context_passing function that processes the workflow state.

    Line 27: Module docstring

        """Create Context Passing workflow."""
    

    Describes the pattern and its purpose.


    Usage Example

    if __name__ == "__main__":
        workflow = create_context_passing()
        result = workflow.invoke({"input": "Test Context Passing", "output": ""})
        print(f"\n=== Context Passing Pattern ===\nOutput: {result['output'][:200]}...")
    

    Key Concepts

    1. State Management: Uses TypedDict to maintain workflow state
    2. Graph-Based Flow: LangGraph orchestrates execution flow
    3. Agent Functions: Each node processes and updates state
    4. LLM Integration: Uses ChatOpenAI for intelligent processing

    Pattern Benefits

    • ✅ Type-safe state management
    • ✅ Clear workflow visualization
    • ✅ Modular and reusable components
    • ✅ Easy to test and debug
    • ✅ Scalable architecture

    Related Patterns

    • See other patterns in the same category
    • Explore combination possibilities
    • Check the main documentation for pattern relationships
  • Agentic Orchestration Pattern: State Machine Pattern

    
        """
    Pattern 08.01: State Machine
    Explicit states and transitions
    """
    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    
    
    class StateMachineState(TypedDict):
        """State for State Machine pattern."""
        input: str
        output: str
    
    
    def agent_function(state: StateMachineState) -> StateMachineState:
        """State Machine agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using State Machine pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_state_machine():
        """Create State Machine workflow."""
        workflow = StateGraph(StateMachineState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
        workflow = create_state_machine()
        result = workflow.invoke({"input": "Test State Machine", "output": ""})
        print(f"\n=== State Machine Pattern ===\nOutput: {result['output'][:200]}...")
    
        

    Agentic Orchestration Pattern: State Machine Pattern

    01 State Machine Pattern – Detailed Documentation

    Overview

    File: 08_state_management/01_state_machine.py
    Pattern Category: 08 State Management
    Implementation: LangGraph/CrewAI

    Pattern Description

    Pattern 08.01: State Machine Explicit states and transitions


    Code Structure

    Imports

    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    

    Purpose of imports: – TypedDict: For type-safe state definitions – StateGraph: LangGraph’s graph-based workflow orchestration – ChatOpenAI: OpenAI LLM integration – HumanMessage: LangChain message formatting

    State Definition

    class StateMachineState(TypedDict):
        """State for State Machine pattern."""
        input: str
        output: str
    
    
    def agent_function(state: StateMachineState) -> StateMachineState:
    

    StateMachineState: State for State Machine pattern.

    State Fields: – input: Stores input – output: Stores output – def agent_function(state: Stores def agent function(state – prompt = f"Process this using State Machine pattern: Stores prompt = f”Process this using State Machine pattern – def create_state_machine(): Stores def create state machine()


    Functions

    agent_function()

    Line: 17
    Arguments: state

    State Machine agent implementation.

    def agent_function(state: StateMachineState) -> StateMachineState:
        """State Machine agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using State Machine pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_state_machine():
    

    create_state_machine()

    Line: 26
    Arguments: “

    Create State Machine workflow.

    def create_state_machine():
        """Create State Machine workflow."""
        workflow = StateGraph(StateMachineState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
    

    Line-by-Line Explanation

    Line 5: Import statement

    from typing import TypedDict
    

    Imports TypedDict for creating type-safe state dictionaries.

    Line 6: Import statement

    from langgraph.graph import StateGraph, END
    

    Imports StateGraph – the core LangGraph class for building workflows.

    Line 7: Import statement

    from langchain_openai import ChatOpenAI
    

    Imports ChatOpenAI for LLM interactions.

    Line 8: Import statement

    from langchain_core.messages import HumanMessage
    

    Imports HumanMessage for formatting prompts to the LLM.

    Line 11: Class definition

    class StateMachineState(TypedDict):
    

    Defines the state structure using TypedDict for type safety.

    Line 12: Module docstring

        """State for State Machine pattern."""
    

    Describes the pattern and its purpose.

    Line 17: Function agent_function definition

    def agent_function(state: StateMachineState) -> StateMachineState:
    

    Defines the agent_function function that processes the workflow state.

    Line 18: Module docstring

        """State Machine agent implementation."""
    

    Describes the pattern and its purpose.

    Line 26: Function create_state_machine definition

    def create_state_machine():
    

    Defines the create_state_machine function that processes the workflow state.

    Line 27: Module docstring

        """Create State Machine workflow."""
    

    Describes the pattern and its purpose.


    Usage Example

    if __name__ == "__main__":
        workflow = create_state_machine()
        result = workflow.invoke({"input": "Test State Machine", "output": ""})
        print(f"\n=== State Machine Pattern ===\nOutput: {result['output'][:200]}...")
    

    Key Concepts

    1. State Management: Uses TypedDict to maintain workflow state
    2. Graph-Based Flow: LangGraph orchestrates execution flow
    3. Agent Functions: Each node processes and updates state
    4. LLM Integration: Uses ChatOpenAI for intelligent processing

    Pattern Benefits

    • ✅ Type-safe state management
    • ✅ Clear workflow visualization
    • ✅ Modular and reusable components
    • ✅ Easy to test and debug
    • ✅ Scalable architecture

    Related Patterns

    • See other patterns in the same category
    • Explore combination possibilities
    • Check the main documentation for pattern relationships
  • Agentic Orchestration Pattern: Adaptive Fallback

    
        """
    Pattern 07.04: Adaptive Fallback
    Dynamic fallback chain based on context
    """
    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    
    
    class AdaptiveFallbackState(TypedDict):
        """State for Adaptive Fallback pattern."""
        input: str
        output: str
    
    
    def agent_function(state: AdaptiveFallbackState) -> AdaptiveFallbackState:
        """Adaptive Fallback agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using Adaptive Fallback pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_adaptive_fallback():
        """Create Adaptive Fallback workflow."""
        workflow = StateGraph(AdaptiveFallbackState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
        workflow = create_adaptive_fallback()
        result = workflow.invoke({"input": "Test Adaptive Fallback", "output": ""})
        print(f"\n=== Adaptive Fallback Pattern ===\nOutput: {result['output'][:200]}...")
    
        

    Agentic Orchestration Pattern: Adaptive Fallback

    04 Adaptive Fallback Pattern – Detailed Documentation

    Overview

    File: 07_adaptive/04_adaptive_fallback.py
    Pattern Category: 07 Adaptive
    Implementation: LangGraph/CrewAI

    Pattern Description

    Pattern 07.04: Adaptive Fallback Dynamic fallback chain based on context


    Code Structure

    Imports

    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    

    Purpose of imports: – TypedDict: For type-safe state definitions – StateGraph: LangGraph’s graph-based workflow orchestration – ChatOpenAI: OpenAI LLM integration – HumanMessage: LangChain message formatting

    State Definition

    class AdaptiveFallbackState(TypedDict):
        """State for Adaptive Fallback pattern."""
        input: str
        output: str
    
    
    def agent_function(state: AdaptiveFallbackState) -> AdaptiveFallbackState:
    

    AdaptiveFallbackState: State for Adaptive Fallback pattern.

    State Fields: – input: Stores input – output: Stores output – def agent_function(state: Stores def agent function(state – prompt = f"Process this using Adaptive Fallback pattern: Stores prompt = f”Process this using Adaptive Fallback pattern – def create_adaptive_fallback(): Stores def create adaptive fallback()


    Functions

    agent_function()

    Line: 17
    Arguments: state

    Adaptive Fallback agent implementation.

    def agent_function(state: AdaptiveFallbackState) -> AdaptiveFallbackState:
        """Adaptive Fallback agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using Adaptive Fallback pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_adaptive_fallback():
    

    create_adaptive_fallback()

    Line: 26
    Arguments: “

    Create Adaptive Fallback workflow.

    def create_adaptive_fallback():
        """Create Adaptive Fallback workflow."""
        workflow = StateGraph(AdaptiveFallbackState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
    

    Line-by-Line Explanation

    Line 5: Import statement

    from typing import TypedDict
    

    Imports TypedDict for creating type-safe state dictionaries.

    Line 6: Import statement

    from langgraph.graph import StateGraph, END
    

    Imports StateGraph – the core LangGraph class for building workflows.

    Line 7: Import statement

    from langchain_openai import ChatOpenAI
    

    Imports ChatOpenAI for LLM interactions.

    Line 8: Import statement

    from langchain_core.messages import HumanMessage
    

    Imports HumanMessage for formatting prompts to the LLM.

    Line 11: Class definition

    class AdaptiveFallbackState(TypedDict):
    

    Defines the state structure using TypedDict for type safety.

    Line 12: Module docstring

        """State for Adaptive Fallback pattern."""
    

    Describes the pattern and its purpose.

    Line 17: Function agent_function definition

    def agent_function(state: AdaptiveFallbackState) -> AdaptiveFallbackState:
    

    Defines the agent_function function that processes the workflow state.

    Line 18: Module docstring

        """Adaptive Fallback agent implementation."""
    

    Describes the pattern and its purpose.

    Line 26: Function create_adaptive_fallback definition

    def create_adaptive_fallback():
    

    Defines the create_adaptive_fallback function that processes the workflow state.

    Line 27: Module docstring

        """Create Adaptive Fallback workflow."""
    

    Describes the pattern and its purpose.


    Usage Example

    if __name__ == "__main__":
        workflow = create_adaptive_fallback()
        result = workflow.invoke({"input": "Test Adaptive Fallback", "output": ""})
        print(f"\n=== Adaptive Fallback Pattern ===\nOutput: {result['output'][:200]}...")
    

    Key Concepts

    1. State Management: Uses TypedDict to maintain workflow state
    2. Graph-Based Flow: LangGraph orchestrates execution flow
    3. Agent Functions: Each node processes and updates state
    4. LLM Integration: Uses ChatOpenAI for intelligent processing

    Pattern Benefits

    • ✅ Type-safe state management
    • ✅ Clear workflow visualization
    • ✅ Modular and reusable components
    • ✅ Easy to test and debug
    • ✅ Scalable architecture

    Related Patterns

    • See other patterns in the same category
    • Explore combination possibilities
    • Check the main documentation for pattern relationships
  • Agentic Orchestration Pattern: A/B Testing Pattern

    
        """
    Pattern 07.03: AB Testing
    Parallel execution of alternatives with comparison
    """
    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    
    
    class AbTestingState(TypedDict):
        """State for AB Testing pattern."""
        input: str
        output: str
    
    
    def agent_function(state: AbTestingState) -> AbTestingState:
        """AB Testing agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using AB Testing pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_ab_testing():
        """Create AB Testing workflow."""
        workflow = StateGraph(AbTestingState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
        workflow = create_ab_testing()
        result = workflow.invoke({"input": "Test AB Testing", "output": ""})
        print(f"\n=== AB Testing Pattern ===\nOutput: {result['output'][:200]}...")
    
        

    Agentic Orchestration Pattern: A/B Testing Pattern

    03 Ab Testing Pattern – Detailed Documentation

    Overview

    File: 07_adaptive/03_ab_testing.py
    Pattern Category: 07 Adaptive
    Implementation: LangGraph/CrewAI

    Pattern Description

    Pattern 07.03: AB Testing Parallel execution of alternatives with comparison


    Code Structure

    Imports

    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    

    Purpose of imports: – TypedDict: For type-safe state definitions – StateGraph: LangGraph’s graph-based workflow orchestration – ChatOpenAI: OpenAI LLM integration – HumanMessage: LangChain message formatting

    State Definition

    class AbTestingState(TypedDict):
        """State for AB Testing pattern."""
        input: str
        output: str
    
    
    def agent_function(state: AbTestingState) -> AbTestingState:
    

    AbTestingState: State for AB Testing pattern.

    State Fields: – input: Stores input – output: Stores output – def agent_function(state: Stores def agent function(state – prompt = f"Process this using AB Testing pattern: Stores prompt = f”Process this using AB Testing pattern – def create_ab_testing(): Stores def create ab testing()


    Functions

    agent_function()

    Line: 17
    Arguments: state

    AB Testing agent implementation.

    def agent_function(state: AbTestingState) -> AbTestingState:
        """AB Testing agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using AB Testing pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_ab_testing():
    

    create_ab_testing()

    Line: 26
    Arguments: “

    Create AB Testing workflow.

    def create_ab_testing():
        """Create AB Testing workflow."""
        workflow = StateGraph(AbTestingState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
    

    Line-by-Line Explanation

    Line 5: Import statement

    from typing import TypedDict
    

    Imports TypedDict for creating type-safe state dictionaries.

    Line 6: Import statement

    from langgraph.graph import StateGraph, END
    

    Imports StateGraph – the core LangGraph class for building workflows.

    Line 7: Import statement

    from langchain_openai import ChatOpenAI
    

    Imports ChatOpenAI for LLM interactions.

    Line 8: Import statement

    from langchain_core.messages import HumanMessage
    

    Imports HumanMessage for formatting prompts to the LLM.

    Line 11: Class definition

    class AbTestingState(TypedDict):
    

    Defines the state structure using TypedDict for type safety.

    Line 12: Module docstring

        """State for AB Testing pattern."""
    

    Describes the pattern and its purpose.

    Line 17: Function agent_function definition

    def agent_function(state: AbTestingState) -> AbTestingState:
    

    Defines the agent_function function that processes the workflow state.

    Line 18: Module docstring

        """AB Testing agent implementation."""
    

    Describes the pattern and its purpose.

    Line 26: Function create_ab_testing definition

    def create_ab_testing():
    

    Defines the create_ab_testing function that processes the workflow state.

    Line 27: Module docstring

        """Create AB Testing workflow."""
    

    Describes the pattern and its purpose.


    Usage Example

    if __name__ == "__main__":
        workflow = create_ab_testing()
        result = workflow.invoke({"input": "Test AB Testing", "output": ""})
        print(f"\n=== AB Testing Pattern ===\nOutput: {result['output'][:200]}...")
    

    Key Concepts

    1. State Management: Uses TypedDict to maintain workflow state
    2. Graph-Based Flow: LangGraph orchestrates execution flow
    3. Agent Functions: Each node processes and updates state
    4. LLM Integration: Uses ChatOpenAI for intelligent processing

    Pattern Benefits

    • ✅ Type-safe state management
    • ✅ Clear workflow visualization
    • ✅ Modular and reusable components
    • ✅ Easy to test and debug
    • ✅ Scalable architecture

    Related Patterns

    • See other patterns in the same category
    • Explore combination possibilities
    • Check the main documentation for pattern relationships
  • Agentic Orchestration Pattern: Learning Pattern

    
        """
    Pattern 07.02: Learning
    Agents improve from experience
    """
    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    
    
    class LearningState(TypedDict):
        """State for Learning pattern."""
        input: str
        output: str
    
    
    def agent_function(state: LearningState) -> LearningState:
        """Learning agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using Learning pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_learning():
        """Create Learning workflow."""
        workflow = StateGraph(LearningState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
        workflow = create_learning()
        result = workflow.invoke({"input": "Test Learning", "output": ""})
        print(f"\n=== Learning Pattern ===\nOutput: {result['output'][:200]}...")
    
        

    Agentic Orchestration Pattern: Learning Pattern

    02 Learning Pattern – Detailed Documentation

    Overview

    File: 07_adaptive/02_learning.py
    Pattern Category: 07 Adaptive
    Implementation: LangGraph/CrewAI

    Pattern Description

    Pattern 07.02: Learning Agents improve from experience


    Code Structure

    Imports

    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    

    Purpose of imports: – TypedDict: For type-safe state definitions – StateGraph: LangGraph’s graph-based workflow orchestration – ChatOpenAI: OpenAI LLM integration – HumanMessage: LangChain message formatting

    State Definition

    class LearningState(TypedDict):
        """State for Learning pattern."""
        input: str
        output: str
    
    
    def agent_function(state: LearningState) -> LearningState:
    

    LearningState: State for Learning pattern.

    State Fields: – input: Stores input – output: Stores output – def agent_function(state: Stores def agent function(state – prompt = f"Process this using Learning pattern: Stores prompt = f”Process this using Learning pattern – def create_learning(): Stores def create learning()


    Functions

    agent_function()

    Line: 17
    Arguments: state

    Learning agent implementation.

    def agent_function(state: LearningState) -> LearningState:
        """Learning agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using Learning pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_learning():
    

    create_learning()

    Line: 26
    Arguments: “

    Create Learning workflow.

    def create_learning():
        """Create Learning workflow."""
        workflow = StateGraph(LearningState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
    

    Line-by-Line Explanation

    Line 5: Import statement

    from typing import TypedDict
    

    Imports TypedDict for creating type-safe state dictionaries.

    Line 6: Import statement

    from langgraph.graph import StateGraph, END
    

    Imports StateGraph – the core LangGraph class for building workflows.

    Line 7: Import statement

    from langchain_openai import ChatOpenAI
    

    Imports ChatOpenAI for LLM interactions.

    Line 8: Import statement

    from langchain_core.messages import HumanMessage
    

    Imports HumanMessage for formatting prompts to the LLM.

    Line 11: Class definition

    class LearningState(TypedDict):
    

    Defines the state structure using TypedDict for type safety.

    Line 12: Module docstring

        """State for Learning pattern."""
    

    Describes the pattern and its purpose.

    Line 17: Function agent_function definition

    def agent_function(state: LearningState) -> LearningState:
    

    Defines the agent_function function that processes the workflow state.

    Line 18: Module docstring

        """Learning agent implementation."""
    

    Describes the pattern and its purpose.

    Line 26: Function create_learning definition

    def create_learning():
    

    Defines the create_learning function that processes the workflow state.

    Line 27: Module docstring

        """Create Learning workflow."""
    

    Describes the pattern and its purpose.


    Usage Example

    if __name__ == "__main__":
        workflow = create_learning()
        result = workflow.invoke({"input": "Test Learning", "output": ""})
        print(f"\n=== Learning Pattern ===\nOutput: {result['output'][:200]}...")
    

    Key Concepts

    1. State Management: Uses TypedDict to maintain workflow state
    2. Graph-Based Flow: LangGraph orchestrates execution flow
    3. Agent Functions: Each node processes and updates state
    4. LLM Integration: Uses ChatOpenAI for intelligent processing

    Pattern Benefits

    • ✅ Type-safe state management
    • ✅ Clear workflow visualization
    • ✅ Modular and reusable components
    • ✅ Easy to test and debug
    • ✅ Scalable architecture

    Related Patterns

    • See other patterns in the same category
    • Explore combination possibilities
    • Check the main documentation for pattern relationships
  • Agentic Orchestration Pattern: Dynamic Routing

    
        """
    Pattern 07.01: Dynamic Routing
    Runtime determination of execution path
    """
    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    
    
    class DynamicRoutingState(TypedDict):
        """State for Dynamic Routing pattern."""
        input: str
        output: str
    
    
    def agent_function(state: DynamicRoutingState) -> DynamicRoutingState:
        """Dynamic Routing agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using Dynamic Routing pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_dynamic_routing():
        """Create Dynamic Routing workflow."""
        workflow = StateGraph(DynamicRoutingState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
        workflow = create_dynamic_routing()
        result = workflow.invoke({"input": "Test Dynamic Routing", "output": ""})
        print(f"\n=== Dynamic Routing Pattern ===\nOutput: {result['output'][:200]}...")
    
        

    Agentic Orchestration Pattern: Dynamic Routing

    01 Dynamic Routing Pattern – Detailed Documentation

    Overview

    File: 07_adaptive/01_dynamic_routing.py
    Pattern Category: 07 Adaptive
    Implementation: LangGraph/CrewAI

    Pattern Description

    Pattern 07.01: Dynamic Routing Runtime determination of execution path


    Code Structure

    Imports

    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    

    Purpose of imports: – TypedDict: For type-safe state definitions – StateGraph: LangGraph’s graph-based workflow orchestration – ChatOpenAI: OpenAI LLM integration – HumanMessage: LangChain message formatting

    State Definition

    class DynamicRoutingState(TypedDict):
        """State for Dynamic Routing pattern."""
        input: str
        output: str
    
    
    def agent_function(state: DynamicRoutingState) -> DynamicRoutingState:
    

    DynamicRoutingState: State for Dynamic Routing pattern.

    State Fields: – input: Stores input – output: Stores output – def agent_function(state: Stores def agent function(state – prompt = f"Process this using Dynamic Routing pattern: Stores prompt = f”Process this using Dynamic Routing pattern – def create_dynamic_routing(): Stores def create dynamic routing()


    Functions

    agent_function()

    Line: 17
    Arguments: state

    Dynamic Routing agent implementation.

    def agent_function(state: DynamicRoutingState) -> DynamicRoutingState:
        """Dynamic Routing agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using Dynamic Routing pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_dynamic_routing():
    

    create_dynamic_routing()

    Line: 26
    Arguments: “

    Create Dynamic Routing workflow.

    def create_dynamic_routing():
        """Create Dynamic Routing workflow."""
        workflow = StateGraph(DynamicRoutingState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
    

    Line-by-Line Explanation

    Line 5: Import statement

    from typing import TypedDict
    

    Imports TypedDict for creating type-safe state dictionaries.

    Line 6: Import statement

    from langgraph.graph import StateGraph, END
    

    Imports StateGraph – the core LangGraph class for building workflows.

    Line 7: Import statement

    from langchain_openai import ChatOpenAI
    

    Imports ChatOpenAI for LLM interactions.

    Line 8: Import statement

    from langchain_core.messages import HumanMessage
    

    Imports HumanMessage for formatting prompts to the LLM.

    Line 11: Class definition

    class DynamicRoutingState(TypedDict):
    

    Defines the state structure using TypedDict for type safety.

    Line 12: Module docstring

        """State for Dynamic Routing pattern."""
    

    Describes the pattern and its purpose.

    Line 17: Function agent_function definition

    def agent_function(state: DynamicRoutingState) -> DynamicRoutingState:
    

    Defines the agent_function function that processes the workflow state.

    Line 18: Module docstring

        """Dynamic Routing agent implementation."""
    

    Describes the pattern and its purpose.

    Line 26: Function create_dynamic_routing definition

    def create_dynamic_routing():
    

    Defines the create_dynamic_routing function that processes the workflow state.

    Line 27: Module docstring

        """Create Dynamic Routing workflow."""
    

    Describes the pattern and its purpose.


    Usage Example

    if __name__ == "__main__":
        workflow = create_dynamic_routing()
        result = workflow.invoke({"input": "Test Dynamic Routing", "output": ""})
        print(f"\n=== Dynamic Routing Pattern ===\nOutput: {result['output'][:200]}...")
    

    Key Concepts

    1. State Management: Uses TypedDict to maintain workflow state
    2. Graph-Based Flow: LangGraph orchestrates execution flow
    3. Agent Functions: Each node processes and updates state
    4. LLM Integration: Uses ChatOpenAI for intelligent processing

    Pattern Benefits

    • ✅ Type-safe state management
    • ✅ Clear workflow visualization
    • ✅ Modular and reusable components
    • ✅ Easy to test and debug
    • ✅ Scalable architecture

    Related Patterns

    • See other patterns in the same category
    • Explore combination possibilities
    • Check the main documentation for pattern relationships
  • Agentic Orchestration Pattern: Self-Reflection Pattern

    
        """
    Pattern 06.04: Self-Reflection
    Agent evaluates its own output
    """
    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    
    
    class SelfReflectionState(TypedDict):
        """State for Self-Reflection pattern."""
        input: str
        output: str
    
    
    def agent_function(state: SelfReflectionState) -> SelfReflectionState:
        """Self-Reflection agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using Self-Reflection pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_self_reflection():
        """Create Self-Reflection workflow."""
        workflow = StateGraph(SelfReflectionState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
        workflow = create_self_reflection()
        result = workflow.invoke({"input": "Test Self-Reflection", "output": ""})
        print(f"\n=== Self-Reflection Pattern ===\nOutput: {result['output'][:200]}...")
    
        

    Agentic Orchestration Pattern: Self-Reflection Pattern

    04 Self Reflection Pattern – Detailed Documentation

    Overview

    File: 06_feedback_control/04_self_reflection.py
    Pattern Category: 06 Feedback Control
    Implementation: LangGraph/CrewAI

    Pattern Description

    Pattern 06.04: Self-Reflection Agent evaluates its own output


    Code Structure

    Imports

    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    

    Purpose of imports: – TypedDict: For type-safe state definitions – StateGraph: LangGraph’s graph-based workflow orchestration – ChatOpenAI: OpenAI LLM integration – HumanMessage: LangChain message formatting

    State Definition

    class SelfReflectionState(TypedDict):
        """State for Self-Reflection pattern."""
        input: str
        output: str
    
    
    def agent_function(state: SelfReflectionState) -> SelfReflectionState:
    

    SelfReflectionState: State for Self-Reflection pattern.

    State Fields: – input: Stores input – output: Stores output – def agent_function(state: Stores def agent function(state – prompt = f"Process this using Self-Reflection pattern: Stores prompt = f”Process this using Self-Reflection pattern – def create_self_reflection(): Stores def create self reflection()


    Functions

    agent_function()

    Line: 17
    Arguments: state

    Self-Reflection agent implementation.

    def agent_function(state: SelfReflectionState) -> SelfReflectionState:
        """Self-Reflection agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using Self-Reflection pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_self_reflection():
    

    create_self_reflection()

    Line: 26
    Arguments: “

    Create Self-Reflection workflow.

    def create_self_reflection():
        """Create Self-Reflection workflow."""
        workflow = StateGraph(SelfReflectionState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
    

    Line-by-Line Explanation

    Line 5: Import statement

    from typing import TypedDict
    

    Imports TypedDict for creating type-safe state dictionaries.

    Line 6: Import statement

    from langgraph.graph import StateGraph, END
    

    Imports StateGraph – the core LangGraph class for building workflows.

    Line 7: Import statement

    from langchain_openai import ChatOpenAI
    

    Imports ChatOpenAI for LLM interactions.

    Line 8: Import statement

    from langchain_core.messages import HumanMessage
    

    Imports HumanMessage for formatting prompts to the LLM.

    Line 11: Class definition

    class SelfReflectionState(TypedDict):
    

    Defines the state structure using TypedDict for type safety.

    Line 12: Module docstring

        """State for Self-Reflection pattern."""
    

    Describes the pattern and its purpose.

    Line 17: Function agent_function definition

    def agent_function(state: SelfReflectionState) -> SelfReflectionState:
    

    Defines the agent_function function that processes the workflow state.

    Line 18: Module docstring

        """Self-Reflection agent implementation."""
    

    Describes the pattern and its purpose.

    Line 26: Function create_self_reflection definition

    def create_self_reflection():
    

    Defines the create_self_reflection function that processes the workflow state.

    Line 27: Module docstring

        """Create Self-Reflection workflow."""
    

    Describes the pattern and its purpose.


    Usage Example

    if __name__ == "__main__":
        workflow = create_self_reflection()
        result = workflow.invoke({"input": "Test Self-Reflection", "output": ""})
        print(f"\n=== Self-Reflection Pattern ===\nOutput: {result['output'][:200]}...")
    

    Key Concepts

    1. State Management: Uses TypedDict to maintain workflow state
    2. Graph-Based Flow: LangGraph orchestrates execution flow
    3. Agent Functions: Each node processes and updates state
    4. LLM Integration: Uses ChatOpenAI for intelligent processing

    Pattern Benefits

    • ✅ Type-safe state management
    • ✅ Clear workflow visualization
    • ✅ Modular and reusable components
    • ✅ Easy to test and debug
    • ✅ Scalable architecture

    Related Patterns

    • See other patterns in the same category
    • Explore combination possibilities
    • Check the main documentation for pattern relationships
  • Agentic Orchestration Pattern: Monitor-Controller Pattern

    
        """
    Pattern 06.03: Monitor-Controller
    Monitoring agent observes, controller adjusts
    """
    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    
    
    class MonitorControllerState(TypedDict):
        """State for Monitor-Controller pattern."""
        input: str
        output: str
    
    
    def agent_function(state: MonitorControllerState) -> MonitorControllerState:
        """Monitor-Controller agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using Monitor-Controller pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_monitor_controller():
        """Create Monitor-Controller workflow."""
        workflow = StateGraph(MonitorControllerState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
        workflow = create_monitor_controller()
        result = workflow.invoke({"input": "Test Monitor-Controller", "output": ""})
        print(f"\n=== Monitor-Controller Pattern ===\nOutput: {result['output'][:200]}...")
    
        

    Agentic Orchestration Pattern: Monitor-Controller Pattern

    03 Monitor Controller Pattern – Detailed Documentation

    Overview

    File: 06_feedback_control/03_monitor_controller.py
    Pattern Category: 06 Feedback Control
    Implementation: LangGraph/CrewAI

    Pattern Description

    Pattern 06.03: Monitor-Controller Monitoring agent observes, controller adjusts


    Code Structure

    Imports

    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    

    Purpose of imports: – TypedDict: For type-safe state definitions – StateGraph: LangGraph’s graph-based workflow orchestration – ChatOpenAI: OpenAI LLM integration – HumanMessage: LangChain message formatting

    State Definition

    class MonitorControllerState(TypedDict):
        """State for Monitor-Controller pattern."""
        input: str
        output: str
    
    
    def agent_function(state: MonitorControllerState) -> MonitorControllerState:
    

    MonitorControllerState: State for Monitor-Controller pattern.

    State Fields: – input: Stores input – output: Stores output – def agent_function(state: Stores def agent function(state – prompt = f"Process this using Monitor-Controller pattern: Stores prompt = f”Process this using Monitor-Controller pattern – def create_monitor_controller(): Stores def create monitor controller()


    Functions

    agent_function()

    Line: 17
    Arguments: state

    Monitor-Controller agent implementation.

    def agent_function(state: MonitorControllerState) -> MonitorControllerState:
        """Monitor-Controller agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using Monitor-Controller pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_monitor_controller():
    

    create_monitor_controller()

    Line: 26
    Arguments: “

    Create Monitor-Controller workflow.

    def create_monitor_controller():
        """Create Monitor-Controller workflow."""
        workflow = StateGraph(MonitorControllerState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
    

    Line-by-Line Explanation

    Line 5: Import statement

    from typing import TypedDict
    

    Imports TypedDict for creating type-safe state dictionaries.

    Line 6: Import statement

    from langgraph.graph import StateGraph, END
    

    Imports StateGraph – the core LangGraph class for building workflows.

    Line 7: Import statement

    from langchain_openai import ChatOpenAI
    

    Imports ChatOpenAI for LLM interactions.

    Line 8: Import statement

    from langchain_core.messages import HumanMessage
    

    Imports HumanMessage for formatting prompts to the LLM.

    Line 11: Class definition

    class MonitorControllerState(TypedDict):
    

    Defines the state structure using TypedDict for type safety.

    Line 12: Module docstring

        """State for Monitor-Controller pattern."""
    

    Describes the pattern and its purpose.

    Line 17: Function agent_function definition

    def agent_function(state: MonitorControllerState) -> MonitorControllerState:
    

    Defines the agent_function function that processes the workflow state.

    Line 18: Module docstring

        """Monitor-Controller agent implementation."""
    

    Describes the pattern and its purpose.

    Line 26: Function create_monitor_controller definition

    def create_monitor_controller():
    

    Defines the create_monitor_controller function that processes the workflow state.

    Line 27: Module docstring

        """Create Monitor-Controller workflow."""
    

    Describes the pattern and its purpose.


    Usage Example

    if __name__ == "__main__":
        workflow = create_monitor_controller()
        result = workflow.invoke({"input": "Test Monitor-Controller", "output": ""})
        print(f"\n=== Monitor-Controller Pattern ===\nOutput: {result['output'][:200]}...")
    

    Key Concepts

    1. State Management: Uses TypedDict to maintain workflow state
    2. Graph-Based Flow: LangGraph orchestrates execution flow
    3. Agent Functions: Each node processes and updates state
    4. LLM Integration: Uses ChatOpenAI for intelligent processing

    Pattern Benefits

    • ✅ Type-safe state management
    • ✅ Clear workflow visualization
    • ✅ Modular and reusable components
    • ✅ Easy to test and debug
    • ✅ Scalable architecture

    Related Patterns

    • See other patterns in the same category
    • Explore combination possibilities
    • Check the main documentation for pattern relationships
  • Agentic Orchestration Pattern: Critic-Actor Pattern

    
        """
    Pattern 06.02: Critic-Actor
    One agent generates, another critiques
    """
    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    
    
    class CriticActorState(TypedDict):
        """State for Critic-Actor pattern."""
        input: str
        output: str
    
    
    def agent_function(state: CriticActorState) -> CriticActorState:
        """Critic-Actor agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using Critic-Actor pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_critic_actor():
        """Create Critic-Actor workflow."""
        workflow = StateGraph(CriticActorState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
        workflow = create_critic_actor()
        result = workflow.invoke({"input": "Test Critic-Actor", "output": ""})
        print(f"\n=== Critic-Actor Pattern ===\nOutput: {result['output'][:200]}...")
    
        

    Agentic Orchestration Pattern: Critic-Actor Pattern

    02 Critic Actor Pattern – Detailed Documentation

    Overview

    File: 06_feedback_control/02_critic_actor.py
    Pattern Category: 06 Feedback Control
    Implementation: LangGraph/CrewAI

    Pattern Description

    Pattern 06.02: Critic-Actor One agent generates, another critiques


    Code Structure

    Imports

    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    

    Purpose of imports: – TypedDict: For type-safe state definitions – StateGraph: LangGraph’s graph-based workflow orchestration – ChatOpenAI: OpenAI LLM integration – HumanMessage: LangChain message formatting

    State Definition

    class CriticActorState(TypedDict):
        """State for Critic-Actor pattern."""
        input: str
        output: str
    
    
    def agent_function(state: CriticActorState) -> CriticActorState:
    

    CriticActorState: State for Critic-Actor pattern.

    State Fields: – input: Stores input – output: Stores output – def agent_function(state: Stores def agent function(state – prompt = f"Process this using Critic-Actor pattern: Stores prompt = f”Process this using Critic-Actor pattern – def create_critic_actor(): Stores def create critic actor()


    Functions

    agent_function()

    Line: 17
    Arguments: state

    Critic-Actor agent implementation.

    def agent_function(state: CriticActorState) -> CriticActorState:
        """Critic-Actor agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using Critic-Actor pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_critic_actor():
    

    create_critic_actor()

    Line: 26
    Arguments: “

    Create Critic-Actor workflow.

    def create_critic_actor():
        """Create Critic-Actor workflow."""
        workflow = StateGraph(CriticActorState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
    

    Line-by-Line Explanation

    Line 5: Import statement

    from typing import TypedDict
    

    Imports TypedDict for creating type-safe state dictionaries.

    Line 6: Import statement

    from langgraph.graph import StateGraph, END
    

    Imports StateGraph – the core LangGraph class for building workflows.

    Line 7: Import statement

    from langchain_openai import ChatOpenAI
    

    Imports ChatOpenAI for LLM interactions.

    Line 8: Import statement

    from langchain_core.messages import HumanMessage
    

    Imports HumanMessage for formatting prompts to the LLM.

    Line 11: Class definition

    class CriticActorState(TypedDict):
    

    Defines the state structure using TypedDict for type safety.

    Line 12: Module docstring

        """State for Critic-Actor pattern."""
    

    Describes the pattern and its purpose.

    Line 17: Function agent_function definition

    def agent_function(state: CriticActorState) -> CriticActorState:
    

    Defines the agent_function function that processes the workflow state.

    Line 18: Module docstring

        """Critic-Actor agent implementation."""
    

    Describes the pattern and its purpose.

    Line 26: Function create_critic_actor definition

    def create_critic_actor():
    

    Defines the create_critic_actor function that processes the workflow state.

    Line 27: Module docstring

        """Create Critic-Actor workflow."""
    

    Describes the pattern and its purpose.


    Usage Example

    if __name__ == "__main__":
        workflow = create_critic_actor()
        result = workflow.invoke({"input": "Test Critic-Actor", "output": ""})
        print(f"\n=== Critic-Actor Pattern ===\nOutput: {result['output'][:200]}...")
    

    Key Concepts

    1. State Management: Uses TypedDict to maintain workflow state
    2. Graph-Based Flow: LangGraph orchestrates execution flow
    3. Agent Functions: Each node processes and updates state
    4. LLM Integration: Uses ChatOpenAI for intelligent processing

    Pattern Benefits

    • ✅ Type-safe state management
    • ✅ Clear workflow visualization
    • ✅ Modular and reusable components
    • ✅ Easy to test and debug
    • ✅ Scalable architecture

    Related Patterns

    • See other patterns in the same category
    • Explore combination possibilities
    • Check the main documentation for pattern relationships
  • Agentic Orchestration Pattern: Feedback Loop Pattern

    
        """
    Pattern 06.01: Feedback Loop
    Output fed back as input for self-correction
    """
    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    
    
    class FeedbackLoopState(TypedDict):
        """State for Feedback Loop pattern."""
        input: str
        output: str
    
    
    def agent_function(state: FeedbackLoopState) -> FeedbackLoopState:
        """Feedback Loop agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using Feedback Loop pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_feedback_loop():
        """Create Feedback Loop workflow."""
        workflow = StateGraph(FeedbackLoopState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
        workflow = create_feedback_loop()
        result = workflow.invoke({"input": "Test Feedback Loop", "output": ""})
        print(f"\n=== Feedback Loop Pattern ===\nOutput: {result['output'][:200]}...")
    
        

    Agentic Orchestration Pattern: Feedback Loop Pattern

    01 Feedback Loop Pattern – Detailed Documentation

    Overview

    File: 06_feedback_control/01_feedback_loop.py
    Pattern Category: 06 Feedback Control
    Implementation: LangGraph/CrewAI

    Pattern Description

    Pattern 06.01: Feedback Loop Output fed back as input for self-correction


    Code Structure

    Imports

    from typing import TypedDict
    from langgraph.graph import StateGraph, END
    from langchain_openai import ChatOpenAI
    from langchain_core.messages import HumanMessage
    

    Purpose of imports: – TypedDict: For type-safe state definitions – StateGraph: LangGraph’s graph-based workflow orchestration – ChatOpenAI: OpenAI LLM integration – HumanMessage: LangChain message formatting

    State Definition

    class FeedbackLoopState(TypedDict):
        """State for Feedback Loop pattern."""
        input: str
        output: str
    
    
    def agent_function(state: FeedbackLoopState) -> FeedbackLoopState:
    

    FeedbackLoopState: State for Feedback Loop pattern.

    State Fields: – input: Stores input – output: Stores output – def agent_function(state: Stores def agent function(state – prompt = f"Process this using Feedback Loop pattern: Stores prompt = f”Process this using Feedback Loop pattern – def create_feedback_loop(): Stores def create feedback loop()


    Functions

    agent_function()

    Line: 17
    Arguments: state

    Feedback Loop agent implementation.

    def agent_function(state: FeedbackLoopState) -> FeedbackLoopState:
        """Feedback Loop agent implementation."""
        llm = ChatOpenAI()
        prompt = f"Process this using Feedback Loop pattern: {state['input']}"
        response = llm.invoke([HumanMessage(content=prompt)])
        state["output"] = str(response.content)
        return state
    
    
    def create_feedback_loop():
    

    create_feedback_loop()

    Line: 26
    Arguments: “

    Create Feedback Loop workflow.

    def create_feedback_loop():
        """Create Feedback Loop workflow."""
        workflow = StateGraph(FeedbackLoopState)
        workflow.add_node("agent", agent_function)
        workflow.set_entry_point("agent")
        workflow.add_edge("agent", END)
        return workflow.compile()
    
    
    if __name__ == "__main__":
    

    Line-by-Line Explanation

    Line 5: Import statement

    from typing import TypedDict
    

    Imports TypedDict for creating type-safe state dictionaries.

    Line 6: Import statement

    from langgraph.graph import StateGraph, END
    

    Imports StateGraph – the core LangGraph class for building workflows.

    Line 7: Import statement

    from langchain_openai import ChatOpenAI
    

    Imports ChatOpenAI for LLM interactions.

    Line 8: Import statement

    from langchain_core.messages import HumanMessage
    

    Imports HumanMessage for formatting prompts to the LLM.

    Line 11: Class definition

    class FeedbackLoopState(TypedDict):
    

    Defines the state structure using TypedDict for type safety.

    Line 12: Module docstring

        """State for Feedback Loop pattern."""
    

    Describes the pattern and its purpose.

    Line 17: Function agent_function definition

    def agent_function(state: FeedbackLoopState) -> FeedbackLoopState:
    

    Defines the agent_function function that processes the workflow state.

    Line 18: Module docstring

        """Feedback Loop agent implementation."""
    

    Describes the pattern and its purpose.

    Line 26: Function create_feedback_loop definition

    def create_feedback_loop():
    

    Defines the create_feedback_loop function that processes the workflow state.

    Line 27: Module docstring

        """Create Feedback Loop workflow."""
    

    Describes the pattern and its purpose.


    Usage Example

    if __name__ == "__main__":
        workflow = create_feedback_loop()
        result = workflow.invoke({"input": "Test Feedback Loop", "output": ""})
        print(f"\n=== Feedback Loop Pattern ===\nOutput: {result['output'][:200]}...")
    

    Key Concepts

    1. State Management: Uses TypedDict to maintain workflow state
    2. Graph-Based Flow: LangGraph orchestrates execution flow
    3. Agent Functions: Each node processes and updates state
    4. LLM Integration: Uses ChatOpenAI for intelligent processing

    Pattern Benefits

    • ✅ Type-safe state management
    • ✅ Clear workflow visualization
    • ✅ Modular and reusable components
    • ✅ Easy to test and debug
    • ✅ Scalable architecture

    Related Patterns

    • See other patterns in the same category
    • Explore combination possibilities
    • Check the main documentation for pattern relationships