Agent Orchestration Patterns

from typing import Annotated, List, Optional
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from pydantic import BaseModel, Field

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# ── State ──
class PlanExecuteState(TypedDict):
    messages:      Annotated[list[BaseMessage], add_messages]
    plan:          List[str]          # ordered list of steps
    current_step:  int
    past_results:  List[str]
    final_answer:  Optional[str]

# ── Planner — generates initial step list ──
class Plan(BaseModel):
    steps: List[str] = Field(description="Ordered list of steps to complete the task")

planner_prompt = ChatPromptTemplate.from_messages([
    ("system", "Create a clear, step-by-step plan to complete the user's task. "
               "Each step should be one specific, actionable instruction."),
    ("human", "{task}"),
])

def planner_node(state: PlanExecuteState) -> dict:
    task = state["messages"][-1].content
    structured_llm = llm.with_structured_output(Plan)
    plan = (planner_prompt | structured_llm).invoke({"task": task})
    return {"plan": plan.steps, "current_step": 0, "past_results": []}

# ── Executor — carries out one step at a time ──
executor_prompt = ChatPromptTemplate.from_messages([
    ("system", "Execute the given step using available information.\n"
               "Previous results: {past_results}"),
    ("human", "Step to execute: {step}"),
])

def executor_node(state: PlanExecuteState) -> dict:
    step = state["plan"][state["current_step"]]
    past = "\n".join(state["past_results"])
    result = (executor_prompt | llm).invoke({"step": step, "past_results": past})
    return {
        "past_results": state["past_results"] + [f"Step {state['current_step']+1}: {result.content}"],
        "current_step": state["current_step"] + 1,
    }

# ── Router — done or continue? ──
def should_continue(state: PlanExecuteState) -> str:
    if state["current_step"] >= len(state["plan"]):
        return "finish"
    return "execute"

def finish_node(state: PlanExecuteState) -> dict:
    summary = "\n".join(state["past_results"])
    final = llm.invoke(f"Summarise the following completed steps into a final answer:\n{summary}")
    return {"final_answer": final.content}

# ── Build graph ──
builder = StateGraph(PlanExecuteState)
builder.add_node("planner",  planner_node)
builder.add_node("executor", executor_node)
builder.add_node("finish",   finish_node)
builder.add_edge(START, "planner")
builder.add_edge("planner", "executor")
builder.add_conditional_edges("executor", should_continue, {"execute": "executor", "finish": "finish"})
builder.add_edge("finish", END)
plan_execute_graph = builder.compile()

result = plan_execute_graph.invoke({
    "messages": [HumanMessage("Research the top 3 Python web frameworks and compare their performance.")],
    "plan": [], "current_step": 0, "past_results": [], "final_answer": None,
})
print(result["final_answer"])

from typing_extensions import TypedDict
from typing import Annotated, Optional
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
MAX_ITERATIONS = 3

class ReflectionState(TypedDict):
    messages:   Annotated[list[BaseMessage], add_messages]
    draft:      str
    critique:   Optional[str]
    iteration:  int

def generate_node(state: ReflectionState) -> dict:
    """Generate or revise based on critique."""
    system = SystemMessage(content="You are an expert technical writer.")
    if state.get("critique"):
        # Revise based on critique
        msgs = [system] + state["messages"] + [
            HumanMessage(f"Revise your draft based on this critique:\n{state['critique']}")
        ]
    else:
        msgs = [system] + state["messages"]
    response = llm.invoke(msgs)
    return {"draft": response.content, "iteration": state.get("iteration", 0) + 1}

def reflect_node(state: ReflectionState) -> dict:
    """Critique the current draft."""
    critique_prompt = [
        SystemMessage(content="You are a harsh but fair editor."),
        HumanMessage(f"Critique this draft. Be specific about weaknesses:\n\n{state['draft']}")
    ]
    critique = llm.invoke(critique_prompt)
    return {"critique": critique.content}

def should_reflect(state: ReflectionState) -> str:
    if state["iteration"] >= MAX_ITERATIONS:
        return "done"
    return "reflect"

builder = StateGraph(ReflectionState)
builder.add_node("generate", generate_node)
builder.add_node("reflect",  reflect_node)
builder.add_edge(START, "generate")
builder.add_conditional_edges("generate", should_reflect, {"reflect": "reflect", "done": END})
builder.add_edge("reflect", "generate")
reflection_graph = builder.compile()

result = reflection_graph.invoke({
    "messages": [HumanMessage("Write a 200-word executive summary of LangGraph for a CTO.")],
    "draft": "", "critique": None, "iteration": 0,
})
print(f"After {result['iteration']} iterations:\n{result['draft']}")

from typing import Annotated, Literal
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from pydantic import BaseModel

llm = ChatOpenAI(model="gpt-4o", temperature=0)

WORKERS = ["researcher", "coder", "analyst", "FINISH"]

class RouteDecision(BaseModel):
    next: Literal["researcher", "coder", "analyst", "FINISH"]
    reason: str

class SupervisorState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    next:     str

# ── Supervisor node ──
SUPERVISOR_PROMPT = """You are a supervisor coordinating a team of three agents:
- researcher: finds information and facts
- coder: writes and reviews Python code
- analyst: analyses data and draws conclusions

Given the conversation, decide which agent should act next, or FINISH if the task is complete."""

def supervisor_node(state: SupervisorState) -> dict:
    structured = llm.with_structured_output(RouteDecision)
    messages = [SystemMessage(content=SUPERVISOR_PROMPT)] + state["messages"]
    decision = structured.invoke(messages)
    return {"next": decision.next}

# ── Worker factory ──
def make_worker(role: str, instructions: str):
    def worker_node(state: SupervisorState) -> dict:
        response = llm.invoke([
            SystemMessage(content=instructions),
            *state["messages"]
        ])
        # Tag the message with who sent it
        response.name = role
        return {"messages": [response]}
    return worker_node

# ── Build graph ──
builder = StateGraph(SupervisorState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", make_worker("researcher",
    "You are a research specialist. Find and summarise relevant facts."))
builder.add_node("coder", make_worker("coder",
    "You are a Python expert. Write clean, well-commented code."))
builder.add_node("analyst", make_worker("analyst",
    "You are a data analyst. Interpret results and draw conclusions."))

builder.add_edge(START, "supervisor")

# Route based on supervisor's decision
def route(state: SupervisorState) -> str:
    return END if state["next"] == "FINISH" else state["next"]

for worker in ["researcher", "coder", "analyst"]:
    builder.add_conditional_edges("supervisor", route)
    builder.add_edge(worker, "supervisor")   # always return to supervisor

supervisor_graph = builder.compile()

result = supervisor_graph.invoke({
    "messages": [HumanMessage(
        "Research the Fibonacci sequence, write a Python function to compute it, "
        "and analyse its time complexity."
    )],
    "next": "",
})
print(result["messages"][-1].content)

from typing import Annotated, List
from typing_extensions import TypedDict
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
import operator

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class OverallState(TypedDict):
    documents:   List[str]           # input: list of document texts
    summaries:   Annotated[List[str], operator.add]  # reducer: append
    final_summary: str

class ChunkState(TypedDict):
    document: str   # single document for this worker

# ── Map: summarise one document ──
def summarise_document(state: ChunkState) -> dict:
    result = llm.invoke(
        f"Summarise this document in 2 sentences:\n\n{state['document']}"
    )
    return {"summaries": [result.content]}

# ── Fan-out: send each document to a separate summarise_document node ──
def fan_out(state: OverallState) -> List[Send]:
    return [Send("summarise_document", {"document": doc})
            for doc in state["documents"]]

# ── Reduce: combine all summaries ──
def combine_summaries(state: OverallState) -> dict:
    all_summaries = "\n\n".join(
        f"Doc {i+1}: {s}" for i, s in enumerate(state["summaries"])
    )
    final = llm.invoke(
        f"Create one coherent executive summary from these {len(state['summaries'])} summaries:\n\n{all_summaries}"
    )
    return {"final_summary": final.content}

# ── Build ──
builder = StateGraph(OverallState)
builder.add_node("summarise_document", summarise_document)
builder.add_node("combine_summaries",  combine_summaries)

builder.add_conditional_edges(START, fan_out, ["summarise_document"])
builder.add_edge("summarise_document", "combine_summaries")
builder.add_edge("combine_summaries", END)

map_reduce_graph = builder.compile()

# Test with 5 documents processed in parallel
docs = [f"Document {i}: This is content about topic {i}..." for i in range(5)]
result = map_reduce_graph.invoke({"documents": docs, "summaries": []})
print(result["final_summary"])

from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

llm = ChatOpenAI(model="gpt-4o", temperature=0.4)

class DebateState(TypedDict):
    messages:  Annotated[list[BaseMessage], add_messages]
    topic:     str
    pro_arg:   str
    con_arg:   str
    verdict:   str

def advocate_pro(state: DebateState) -> dict:
    response = llm.invoke([
        SystemMessage("You argue STRONGLY IN FAVOUR of the topic. Be concise, specific, and compelling."),
        HumanMessage(f"Topic: {state['topic']}")
    ])
    return {"pro_arg": response.content}

def advocate_con(state: DebateState) -> dict:
    response = llm.invoke([
        SystemMessage("You argue STRONGLY AGAINST the topic. Be concise, specific, and compelling."),
        HumanMessage(f"Topic: {state['topic']}")
    ])
    return {"con_arg": response.content}

def arbiter(state: DebateState) -> dict:
    verdict = llm.invoke([
        SystemMessage("You are an impartial arbiter. Weigh both arguments fairly and give a balanced verdict."),
        HumanMessage(
            f"Topic: {state['topic']}\n\n"
            f"PRO argument:\n{state['pro_arg']}\n\n"
            f"CON argument:\n{state['con_arg']}\n\n"
            "Deliver your verdict, noting the strongest points from each side."
        )
    ])
    return {"verdict": verdict.content}

builder = StateGraph(DebateState)
builder.add_node("advocate_pro", advocate_pro)
builder.add_node("advocate_con", advocate_con)
builder.add_node("arbiter",      arbiter)
builder.add_edge(START, "advocate_pro")
builder.add_edge(START, "advocate_con")
builder.add_edge(["advocate_pro", "advocate_con"], "arbiter")  # wait for both
builder.add_edge("arbiter", END)
debate_graph = builder.compile()

result = debate_graph.invoke({
    "topic": "Should enterprises adopt LLMs in customer-facing workflows?",
    "messages": [], "pro_arg": "", "con_arg": "", "verdict": ""
})
print(result["verdict"])