LangGraph — Multi-Agent Patterns¶

from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage


class ScriptedModel(GenericFakeChatModel):
    """Returns pre-defined messages in order; ignores bound tools."""

    def bind_tools(self, tools, **kwargs):
        return self


def scripted(*messages: AIMessage) -> ScriptedModel:
    return ScriptedModel(messages=iter(messages))


def tool_call(name: str, args: dict, call_id: str) -> AIMessage:
    return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": call_id}])

from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool


@tool
def get_order_status(order_id: str) -> str:
    """Return the delivery status of an order."""
    return f"{order_id}: shipped"


support_agent = create_agent(
    scripted(
        tool_call("get_order_status", {"order_id": "ord-42"}, "call_1"),
        AIMessage("Order ord-42 has shipped."),
    ),
    tools=[get_order_status],
    system_prompt="You are a support agent. Use tools, never guess order data.",
    name="support",
)

result = support_agent.invoke({"messages": [HumanMessage("Where is ord-42?")]})
print([type(m).__name__ for m in result["messages"]])
# ['HumanMessage', 'AIMessage', 'ToolMessage', 'AIMessage']
print(list(support_agent.get_graph().nodes))    # ['__start__', 'model', 'tools', '__end__']

from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

guarded_agent = create_agent(
    scripted(
        tool_call("get_order_status", {"order_id": "ord-42"}, "call_1"),
        AIMessage("Order ord-42 has shipped."),
    ),
    tools=[get_order_status],
    middleware=[HumanInTheLoopMiddleware(interrupt_on={"get_order_status": True})],
    checkpointer=InMemorySaver(),               # required for interrupts
)

config = {"configurable": {"thread_id": "support-1"}}
paused = guarded_agent.invoke({"messages": [HumanMessage("Where is ord-42?")]}, config)
request = paused["__interrupt__"][0].value
print(request["action_requests"][0]["name"], request["review_configs"][0]["allowed_decisions"])
# get_order_status ['approve', 'edit', 'reject', 'respond']

done = guarded_agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config)
print(done["messages"][-1].content)          # Order ord-42 has shipped.

from langchain_core.tools import tool

research_agent = create_agent(
    scripted(AIMessage("Top flaky test: test_checkout_timeout (12% failure rate).")),
    tools=[],
    system_prompt="You analyse CI history.",
    name="research",
)


@tool
def ask_research(question: str) -> str:
    """Ask the research agent about CI history and flaky tests."""
    result = research_agent.invoke({"messages": [HumanMessage(question)]})
    return result["messages"][-1].content          # only the final answer goes back


supervisor = create_agent(
    scripted(
        tool_call("ask_research", {"question": "Which test is the flakiest?"}, "call_r1"),
        AIMessage("The flakiest test is test_checkout_timeout — quarantine it and open a ticket."),
    ),
    tools=[ask_research],
    system_prompt="You coordinate specialists. Delegate, then summarise.",
    name="supervisor",
)

answer = supervisor.invoke({"messages": [HumanMessage("What should we fix first in CI?")]})
print(answer["messages"][-1].content)

from typing import Literal, TypedDict

from pydantic import BaseModel
from langchain_core.runnables import RunnableLambda
from langgraph.graph import START, StateGraph
from langgraph.types import Command


class Route(BaseModel):
    next: Literal["ui_tester", "api_tester", "done"]


class QAState(TypedDict):
    request: str
    done: list[str]


def build_graph(router):
    """router: Runnable[QAState -> Route], e.g. model.with_structured_output(Route)."""

    def supervisor(state: QAState) -> Command[Literal["ui_tester", "api_tester", "__end__"]]:
        decision: Route = router.invoke(state)
        return Command(goto="__end__" if decision.next == "done" else decision.next)

    def ui_tester(state: QAState) -> Command[Literal["supervisor"]]:
        return Command(update={"done": state["done"] + ["ui"]}, goto="supervisor")

    def api_tester(state: QAState) -> Command[Literal["supervisor"]]:
        return Command(update={"done": state["done"] + ["api"]}, goto="supervisor")

    builder = StateGraph(QAState)
    builder.add_node("supervisor", supervisor)
    builder.add_node("ui_tester", ui_tester)
    builder.add_node("api_tester", api_tester)
    builder.add_edge(START, "supervisor")
    return builder.compile()


def stub_router(state: QAState) -> Route:        # deterministic stand-in for the LLM
    if "api" not in state["done"]:
        return Route(next="api_tester")
    if "ui" not in state["done"]:
        return Route(next="ui_tester")
    return Route(next="done")


graph = build_graph(RunnableLambda(stub_router))
print(graph.invoke({"request": "Regression for release 1.4", "done": []}, {"recursion_limit": 20}))
# {'request': 'Regression for release 1.4', 'done': ['api', 'ui']}

from langchain.tools import ToolRuntime
from langchain_core.messages import ToolMessage
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.types import Command


@tool
def transfer_to_billing(runtime: ToolRuntime) -> Command:
    """Hand the conversation to the billing agent."""
    note = ToolMessage("Transferred to billing", tool_call_id=runtime.tool_call_id)
    return Command(
        goto="billing",
        graph=Command.PARENT,
        update={"messages": runtime.state["messages"] + [note]},   # pass the history along
    )


triage = create_agent(scripted(tool_call("transfer_to_billing", {}, "h1")),
                      tools=[transfer_to_billing], name="triage")
billing = create_agent(scripted(AIMessage("Refund issued for invoice INV-9.")),
                       tools=[], name="billing")

builder = StateGraph(MessagesState)
builder.add_node("triage", triage, destinations=("billing",))   # destinations: for rendering
builder.add_node("billing", billing)
builder.add_edge(START, "triage")
desk = builder.compile()

result = desk.invoke({"messages": [HumanMessage("I was charged twice")]})
print([f"{type(m).__name__}: {m.content}" for m in result["messages"]])
# ['HumanMessage: I was charged twice', 'AIMessage: ', 'ToolMessage: Transferred to billing',
#  'AIMessage: Refund issued for invoice INV-9.']