LangGraph Multi-Agent Workflows: Production Patterns for Complex Stateful Orchestration

import os
import operator
from typing import Annotated, TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import interrupt, Command
from langgraph.prebuilt import ToolNode


# --- State definition ---
class ResearchState(TypedDict):
    query: str
    search_results: Annotated[list[str], operator.add]  # parallel branches append
    synthesis: str
    human_feedback: str
    final_report: str
    approved: bool


# --- Tools ---
@tool
async def web_search(query: str) -> str:
    """Search the web for recent information on the given query."""
    # Stub: replace with real search API (Tavily, Serper, etc.)
    return f"[Web search results for '{query}': Found 5 relevant articles]"


@tool
async def knowledge_base_search(query: str) -> str:
    """Search the internal enterprise knowledge base."""
    return f"[KB search results for '{query}': Found 3 internal documents]"


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


# --- Nodes ---
async def web_search_agent(state: ResearchState) -> dict:
    """Parallel branch 1: web search agent."""
    agent = llm.bind_tools([web_search])
    response = await agent.ainvoke([
        SystemMessage(content="You are a research agent. Use web_search to find relevant information."),
        HumanMessage(content=f"Research this topic: {state['query']}"),
    ])
    # In production: execute tool calls, collect results
    return {"search_results": [f"Web findings: {response.content or 'Tool calls initiated'}"]}


async def kb_search_agent(state: ResearchState) -> dict:
    """Parallel branch 2: internal knowledge base agent."""
    agent = llm.bind_tools([knowledge_base_search])
    response = await agent.ainvoke([
        SystemMessage(content="You are a research agent. Search the knowledge base for relevant internal information."),
        HumanMessage(content=f"Find internal information about: {state['query']}"),
    ])
    return {"search_results": [f"KB findings: {response.content or 'Tool calls initiated'}"]}


async def synthesis_agent(state: ResearchState) -> dict:
    """Fan-in: synthesize results from both search agents."""
    combined = "\n\n".join(state["search_results"])
    response = await llm.ainvoke([
        SystemMessage(content="You are a synthesis agent. Combine research findings into a coherent summary."),
        HumanMessage(content=f"Synthesize these research results:\n\n{combined}\n\nOriginal query: {state['query']}"),
    ])
    return {"synthesis": response.content}


async def human_review_node(state: ResearchState) -> dict:
    """Interrupt the workflow for human review before finalizing."""
    # interrupt() pauses execution and surfaces the synthesis to a human reviewer.
    # Requires a persistent checkpointer (PostgresSaver) to work correctly.
    human_feedback = interrupt({
        "message": "Please review the synthesis before the final report is generated.",
        "synthesis": state["synthesis"],
    })
    # Execution resumes here after graph.invoke(Command(resume=feedback))
    return {"human_feedback": human_feedback, "approved": True}


async def report_generation_node(state: ResearchState) -> dict:
    """Generate the final report incorporating human feedback."""
    feedback_context = f"\n\nHuman reviewer feedback: {state['human_feedback']}" if state.get("human_feedback") else ""
    response = await llm.ainvoke([
        SystemMessage(content="You are a report writer. Generate a polished final report."),
        HumanMessage(content=f"Synthesis:\n{state['synthesis']}{feedback_context}\n\nGenerate the final report."),
    ])
    return {"final_report": response.content}


def route_after_synthesis(state: ResearchState) -> str:
    """Always route through human review for this workflow."""
    return "human_review"


# --- Graph assembly ---
def build_research_graph(checkpointer) -> object:
    builder = StateGraph(ResearchState)

    # Add nodes
    builder.add_node("web_search", web_search_agent)
    builder.add_node("kb_search", kb_search_agent)
    builder.add_node("synthesis", synthesis_agent)
    builder.add_node("human_review", human_review_node)
    builder.add_node("report", report_generation_node)

    # Parallel fan-out from START
    builder.add_edge(START, "web_search")
    builder.add_edge(START, "kb_search")

    # Fan-in: both search agents must complete before synthesis
    builder.add_edge("web_search", "synthesis")
    builder.add_edge("kb_search", "synthesis")

    # Synthesis -> human review -> report -> END
    builder.add_conditional_edges("synthesis", route_after_synthesis, {"human_review": "human_review"})
    builder.add_edge("human_review", "report")
    builder.add_edge("report", END)

    return builder.compile(checkpointer=checkpointer, interrupt_before=[])


# --- Usage with PostgresSaver ---
async def run_research_workflow(query: str, thread_id: str):
    db_uri = os.environ["DATABASE_URL"]  # postgresql://user:pass@host/db
    with PostgresSaver.from_conn_string(db_uri) as checkpointer:
        checkpointer.setup()  # creates checkpoint tables if not exists
        graph = build_research_graph(checkpointer)
        config = {"configurable": {"thread_id": thread_id}}

        # Phase 1: run until interrupt
        async for event in graph.astream_events(
            {"query": query, "search_results": [], "approved": False},
            config=config,
            version="v2",
        ):
            if event["event"] == "on_chat_model_stream":
                chunk = event["data"]["chunk"]
                if chunk.content:
                    print(chunk.content, end="", flush=True)

        # Phase 2: simulate human approval (in production: collect from UI)
        human_response = "Looks good. Please emphasize the cost implications in the final report."
        final_state = await graph.ainvoke(
            Command(resume=human_response),
            config=config,
        )
        return final_state["final_report"]