Agentic AI with LangChain — Part 7: Building Multi-Agent

This walkthrough rebuilds the path from raw materials to a working system for: Agentic AI with LangChain — Part 7: Building Multi-Agent Systems with LangGraph. The focus is operable steps, explicit checks, and code that you can drop into a repo without guessing intent. For the Overview stage, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.

Why Multi-Agents Matter

When working through the Why Multi-Agents Matter stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Checkpoint after expensive steps. Resume should not re-bill the same LLM call when an operator retries a later node.

Technical Concepts Used in This Article

When working through the Technical Concepts Used in stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Checkpoint after expensive steps. Resume should not re-bill the same LLM call when an operator retries a later node.

Project Overview: A Financial Assistant Agent

When working through the Project Overview A Financial stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Checkpoint after expensive steps. Resume should not re-bill the same LLM call when an operator retries a later node. When working through the Project Overview A Financial stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.

1. System Configuration

The 1 System Configuration stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.

OPENAI_API_KEY="<Your OpenAI API Key>"
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="<Your LangSmith API Key"
LANGSMITH_PROJECT="agent-ai"

class BaseConfig(BaseSettings):
    OPENAI_API_KEY: Optional[str]

model_config = SettingsConfigDict(env_file=".env", extra="ignore")

2. Project Structure

The 2 Project Structure stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.

project/
├── tools/
│   ├── __init__.py
│   ├── financial_search.py # fetch stock prices for companies
│   ├── python_REPL.py      # async wrapper of PythonREPL()
|   └── wikipedia_search.py # fetch financial summary of companies
│
├── graphs/            # implement subagents and multi-agents
│   ├── agent.py       # implement research and analyst in a single agent
│   ├── sub_agents.py  # implement research and analyst sub agents
│   ├── multi_agent_supervisor.py # implement supervisor multi-agents
│   └── multi_agent_swarm.py # implement swamp multi-agents
│
├── config.py           # pydantic BaseConfig
│── utils.py            # utility for printing and executing agents
├── .env                # environment variable definition
├── .gitignore
└── requirements.txt     # package requirements

3. Tools of the Financial Assistant Agent

The 3 Tools of the stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Expose tools with narrow schemas and explicit side-effect labels. Hosts need to know which calls mutate state before they auto-approve. The 3 Tools of the stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.

3.1. Stock Data Search Tool

For the 3 1 Stock Data stage, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Authenticate at the gateway and re-authorize at the data plane. A bearer token alone is not a tenancy boundary.

# define LangChain tool to retrieve stock data
@tool
async def stock_data_tool(
        company_ticker: Annotated[str, "The ticker symbol of the company to retrieve their stock performance data."],
        num_days: Annotated[int, "The number of business days of stock data required to respond to the user query."]
) -> str:

        # fetch stock data asynchronously from yfinance
        # check full code for _fetch_stock_sync function
        df = await asyncio.to_thread(_fetch_stock_sync, company_ticker, f"{num_days}d")

        return (f"Successfully executed the stock performance data retrieval tool to retrieve "
                f"the last *{num_days} days* of data for company **{company_ticker}**:\n\n{df.to_markdown()}")

3.2. Python Code Execution Tool

For the 3 2 Python Code stage, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Separate client construction from the message loop so providers can be swapped without rewriting the conversation state machine.

@tool
async def python_repl_tool(
        code: Annotated[str, "The python code to execute to generate your chart."],
) -> str:

    # wrap LLM generated code and execute it by python.REPL()
    # repl = PythonREPL().
    clean_code = textwrap.dedent(code).strip()
    result = await asyncio.to_thread(repl.run, clean_code)

    return (
        f"Successfully executed the Python REPL tool.\n\n"
        f"Python code executed:\n```python\n{code}\n```\n\n"
        f"Code output:\n```\n{result}```"
    )

3.3. Wikipedia Search Tool

For the 3 3 Wikipedia Search stage, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Authenticate at the gateway and re-authorize at the data plane. A bearer token alone is not a tenancy boundary. For the 3 3 Wikipedia Search stage, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.

@tool()
async def wikipedia_tool(
        query: Annotated[str, "The Wikipedia search to execute to find key summary information."],
) -> str:
    """Use this to search Wikipedia for factual information asynchronously."""
    headers = {"User-Agent": "MyLangChainAgent/1.0 (contact@example.com)"}

    async with httpx.AsyncClient(headers=headers, follow_redirects=True) as client:

        # Step 1: Search Wikipedia for matching pages
        search_url = "https://en.wikipedia.org/w/api.php"
        search_params = {
            "action": "query",
            "list": "search",
            "srsearch": query,
            "format": "json"
        }
        response = await client.get(search_url, params=search_params)
        search_data = response.json()
        results = search_data.get("query", {}).get("search", [])
        title = results[0]["title"]

        # Step 2: Fetch the page extract/summary
        summary_url = f"https://en.wikipedia.org/api/rest_v1/page/summary/{title}"
        summary_res = await client.get(summary_url)
        summary_data = summary_res.json()
        summary = summary_data.get("extract", "No summary available.")

        return f"Successfully executed:\nWikipedia summary: {summary}"

4. Swarm Multi-Agents

When working through the 4 Swarm Multi-Agents stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Checkpoint after expensive steps. Resume should not re-bill the same LLM call when an operator retries a later node.

4.1. Architecture of Swarm Multi-Agents

When working through the 4 1 Architecture of stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Checkpoint after expensive steps. Resume should not re-bill the same LLM call when an operator retries a later node.

4.2. Swarm Multi-agents Implementation

When working through the 4 2 Swarm Multi-agents stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Checkpoint after expensive steps. Resume should not re-bill the same LLM call when an operator retries a later node. When working through the 4 2 Swarm Multi-agents stage, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.

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

# Create a handoff tool for researcher -> analyst
transfer_to_analyst = create_handoff_tool(
    agent_name="analyst",
    description="Transfer user to the analyst assistant, who can create visualizations of provided data.",
)

# create research agent as a subagent
research_agent = create_agent(
    llm,
    tools=[wikipedia_tool, stock_data_tool, transfer_to_analyst],
    system_prompt="You provide summaries from Wikipedia, and can query load raw, numerical stock performance data.",
    name="researcher"
)

# define the checkpointer
checkpointer = InMemorySaver()

# create swarm agents
swarm_agent = create_swarm(
    agents=[research_agent, analyst_agent],
    default_active_agent="researcher"
).compile(checkpointer=checkpointer)

5. Supervisor Multi-Agents

The 5 Supervisor Multi-Agents stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.

5.1. Architecture of Supervisor Multi-Agents

The 5 1 Architecture of stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.

5.2. Supervisor Multi-agents Implementation

The 5 2 Supervisor Multi-agents stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts. The 5 2 Supervisor Multi-agents stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.

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

# create a checkpointer
checkpointer = InMemorySaver()

# create research subagent
research_agent = create_agent(
    llm,
    tools=[wikipedia_tool, stock_data_tool],
    system_prompt="You provide summaries from Wikipedia, and can query load raw, numerical stock performance data.",
    name="researcher"
)

# Create the supervisor multi-agent graph and compile it
supervisor = create_supervisor(
    model=llm,
    agents=[research_agent, analyst_agent],
    prompt=(
        "You are a supervisor managing two agents:\n"
        "- a research agent. Assign research and data collection tasks to this agent\n"
        "- an analyst agent. Assign the creation of visualizations via code to this agent\n"
        "Assign work to one agent at a time, do not call agents in parallel.\n"
        "Do not do any work yourself."
    ),
    add_handoff_back_messages=True,
    output_mode="last_message"
).compile(checkpointer=checkpointer)

6. Closing Summary

For the 6 Closing Summary stage, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Put human approval on edges that spend money or change production data. Compile-time wiring does not equal business completeness.

Operational checklist

The Operational checklist stage works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope.

Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.

Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.

Add a smoke test that exercises the critical path in CI with fixtures, not live paid APIs, whenever budgets allow.

Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.

Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.