
    """
Pattern 1.2: Parallel Execution Pattern
Multiple agents execute simultaneously using LangGraph.
"""
from typing import TypedDict, List
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from common import get_openai_api_key, get_model_name, log_agent_action
import asyncio


class ParallelState(TypedDict):
    """State for parallel execution pattern."""
    input: str
    agent1_output: str
    agent2_output: str
    agent3_output: str
    aggregated_output: str


def agent_parallel_1(state: ParallelState) -> ParallelState:
    """First parallel agent - technical analysis."""
    log_agent_action("agent_parallel_1", "processing", {"input": state["input"]})
    
    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"Provide a technical analysis of: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    
    state["agent1_output"] = response.content
    return state


def agent_parallel_2(state: ParallelState) -> ParallelState:
    """Second parallel agent - business analysis."""
    log_agent_action("agent_parallel_2", "processing", {"input": state["input"]})
    
    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"Provide a business perspective on: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    
    state["agent2_output"] = response.content
    return state


def agent_parallel_3(state: ParallelState) -> ParallelState:
    """Third parallel agent - user experience analysis."""
    log_agent_action("agent_parallel_3", "processing", {"input": state["input"]})
    
    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"Provide a user experience perspective on: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    
    state["agent3_output"] = response.content
    return state


def aggregator(state: ParallelState) -> ParallelState:
    """Aggregate results from parallel agents."""
    log_agent_action("aggregator", "combining results")
    
    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"""
    Synthesize these three perspectives into a comprehensive analysis:
    
    Technical: {state['agent1_output']}
    
    Business: {state['agent2_output']}
    
    UX: {state['agent3_output']}
    """
    response = llm.invoke([HumanMessage(content=prompt)])
    
    state["aggregated_output"] = response.content
    return state


def create_parallel_execution():
    """Create and return a parallel execution workflow."""
    workflow = StateGraph(ParallelState)
    
    # Add nodes
    workflow.add_node("agent1", agent_parallel_1)
    workflow.add_node("agent2", agent_parallel_2)
    workflow.add_node("agent3", agent_parallel_3)
    workflow.add_node("aggregator", aggregator)
    
    # Set entry point to all parallel agents
    workflow.set_entry_point("agent1")
    workflow.set_entry_point("agent2")
    workflow.set_entry_point("agent3")
    
    # All parallel agents feed into aggregator
    workflow.add_edge("agent1", "aggregator")
    workflow.add_edge("agent2", "aggregator")
    workflow.add_edge("agent3", "aggregator")
    workflow.add_edge("aggregator", END)
    
    return workflow.compile()


def run_parallel_execution(user_input: str):
    """Run the parallel execution pattern."""
    workflow = create_parallel_execution()
    
    initial_state = {
        "input": user_input,
        "agent1_output": "",
        "agent2_output": "",
        "agent3_output": "",
        "aggregated_output": ""
    }
    
    result = workflow.invoke(initial_state)
    return result


if __name__ == "__main__":
    # Example usage
    result = run_parallel_execution("Implementing AI chatbot for customer service")
    print("\n=== Parallel Execution Pattern ===")
    print(f"Input: {result['input']}")
    print(f"\nTechnical Analysis: {result['agent1_output'][:200]}...")
    print(f"\nBusiness Analysis: {result['agent2_output'][:200]}...")
    print(f"\nUX Analysis: {result['agent3_output'][:200]}...")
    print(f"\nAggregated Result: {result['aggregated_output']}")

Agentic Orchestration Pattern: Parallel Execution

02 Parallel Execution Pattern – Detailed Documentation

Overview

File: 01_basic_coordination/02_parallel_execution.py
Pattern Category: 01 Basic Coordination
Implementation: LangGraph/CrewAI

Pattern Description

Pattern 1.2: Parallel Execution Pattern Multiple agents execute simultaneously using LangGraph.

Code Structure

Imports

from typing import TypedDict, List
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from common import get_openai_api_key, get_model_name, log_agent_action
import asyncio

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 ParallelState(TypedDict):
    """State for parallel execution pattern."""
    input: str
    agent1_output: str
    agent2_output: str
    agent3_output: str
    aggregated_output: str


def agent_parallel_1(state: ParallelState) -> ParallelState:

ParallelState: State for parallel execution pattern.

State Fields: – input: Stores input – agent1_output: Stores agent1 output – agent2_output: Stores agent2 output – agent3_output: Stores agent3 output – aggregated_output: Stores aggregated output – def agent_parallel_1(state: Stores def agent parallel 1(state – log_agent_action("agent_parallel_1", "processing", {"input": Stores log agent action(“agent parallel 1”, “processing”, {“input” – prompt = f"Provide a technical analysis of: Stores prompt = f”Provide a technical analysis of

Functions

`agent_parallel_1()`

Line: 22
Arguments: state

First parallel agent – technical analysis.

def agent_parallel_1(state: ParallelState) -> ParallelState:
    """First parallel agent - technical analysis."""
    log_agent_action("agent_parallel_1", "processing", {"input": state["input"]})

    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"Provide a technical analysis of: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])

    state["agent1_output"] = response.content
    return state


def agent_parallel_2(state: ParallelState) -> ParallelState:

`agent_parallel_2()`

Line: 34
Arguments: state

Second parallel agent – business analysis.

def agent_parallel_2(state: ParallelState) -> ParallelState:
    """Second parallel agent - business analysis."""
    log_agent_action("agent_parallel_2", "processing", {"input": state["input"]})

    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"Provide a business perspective on: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])

    state["agent2_output"] = response.content
    return state


def agent_parallel_3(state: ParallelState) -> ParallelState:

`agent_parallel_3()`

Line: 46
Arguments: state

Third parallel agent – user experience analysis.

def agent_parallel_3(state: ParallelState) -> ParallelState:
    """Third parallel agent - user experience analysis."""
    log_agent_action("agent_parallel_3", "processing", {"input": state["input"]})

    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"Provide a user experience perspective on: {state['input']}"
    response = llm.invoke([HumanMessage(content=prompt)])

    state["agent3_output"] = response.content
    return state


def aggregator(state: ParallelState) -> ParallelState:

`aggregator()`

Line: 58
Arguments: state

Aggregate results from parallel agents.

def aggregator(state: ParallelState) -> ParallelState:
    """Aggregate results from parallel agents."""
    log_agent_action("aggregator", "combining results")

    llm = ChatOpenAI(api_key=get_openai_api_key(), model=get_model_name())
    prompt = f"""
    Synthesize these three perspectives into a comprehensive analysis:

    Technical: {state['agent1_output']}

    Business: {state['agent2_output']}

    UX: {state['agent3_output']}
    """
    response = llm.invoke([HumanMessage(content=prompt)])

    state["aggregated_output"] = response.content
    return state


def create_parallel_execution():

`create_parallel_execution()`

Line: 78
Arguments: “

Create and return a parallel execution workflow.

def create_parallel_execution():
    """Create and return a parallel execution workflow."""
    workflow = StateGraph(ParallelState)

    # Add nodes
    workflow.add_node("agent1", agent_parallel_1)
    workflow.add_node("agent2", agent_parallel_2)
    workflow.add_node("agent3", agent_parallel_3)
    workflow.add_node("aggregator", aggregator)

    # Set entry point to all parallel agents
    workflow.set_entry_point("agent1")
    workflow.set_entry_point("agent2")
    workflow.set_entry_point("agent3")

    # All parallel agents feed into aggregator
    workflow.add_edge("agent1", "aggregator")
    workflow.add_edge("agent2", "aggregator")
    workflow.add_edge("agent3", "aggregator")
    workflow.add_edge("aggregator", END)

    return workflow.compile()


def run_parallel_execution(user_input: str):

`run_parallel_execution()`

Line: 102
Arguments: user_input

Run the parallel execution pattern.

def run_parallel_execution(user_input: str):
    """Run the parallel execution pattern."""
    workflow = create_parallel_execution()

    initial_state = {
        "input": user_input,
        "agent1_output": "",
        "agent2_output": "",
        "agent3_output": "",
        "aggregated_output": ""
    }

    result = workflow.invoke(initial_state)
    return result


if __name__ == "__main__":

Line-by-Line Explanation

Line 5: Import statement

from typing import TypedDict, List

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_core.messages import HumanMessage

Imports HumanMessage for formatting prompts to the LLM.

Line 8: Import statement

from langchain_openai import ChatOpenAI

Imports ChatOpenAI for LLM interactions.

Line 9: Import statement

from common import get_openai_api_key, get_model_name, log_agent_action

Required import for the pattern implementation.

Line 10: Import statement

import asyncio

Required import for the pattern implementation.

Line 13: Class definition

class ParallelState(TypedDict):

Defines the state structure using TypedDict for type safety.

Line 14: Module docstring

    """State for parallel execution pattern."""