Skip to main content

Multi-Agent Systems

Adapted from: Chapter 8, "Multi-agent systems" — rewritten and condensed for teaching, with the source's orchestration bugs fixed and its tests made deterministic.

A customer texts a photo of their hiking boots and writes, "Do you have something similar but waterproof?" Seconds later: "What's your return policy if they don't fit?" Answering takes image understanding, product search, policy retrieval, and one coherent reply. A single LLM juggling all of that gets mediocre at everything. Its context fills with irrelevant material, its behavior becomes inconsistent, and every new capability means rewriting the whole system.

The fix is a multi-agent architecture: specialized agents, each good at one job, coordinated by an orchestrator. In this lesson you'll build ShopBot on top of the tools from Lesson 7, orchestrate it with LangGraph, and test it the way multi-agent systems actually fail: at the seams between agents.

What you'll be able to do after this lesson:

  • Explain the monolithic agent problem and when specialization pays off.
  • Model a workflow as a graph: shared state, nodes, edges, and conditional routing.
  • Build a search agent, a RAG policy agent, a vision adapter, and a synthesis step, then wire them together in LangGraph.
  • Make agents degrade gracefully and keep per-request state isolated.
  • Write deterministic workflow tests instead of brittle checks on LLM output.
  • Compare LangGraph with CrewAI, and choose between them.

Why Multi-Agent?​

Consider this request: "I need a warm jacket for winter camping under $200, and what's your warranty on outdoor gear?" It requires three kinds of intelligence:

  • Product search: warm jackets under $200.
  • Policy knowledge: the warranty terms for outdoor gear.
  • Coordination: merging both into one helpful answer.

One agent doing everything needs the whole product database, every policy, product-category expertise, and coordination logic in a single prompt. That's the monolithic agent problem. Split the work instead:

AgentExpert at
SearchAgentTurning fuzzy requests into product queries and results
PolicyAgent (QA)Returns, shipping, warranties, care, and sizing, grounded by RAG
VisionAgentTurning an image into a text description that search can use
CoordinatorUnderstanding intent, routing to agents, and synthesizing the final reply

Specialization brings several benefits:

  • Modularity: test and improve each agent on its own.
  • Extensibility: add a RecommendationAgent or an InventoryAgent without touching the others.
  • Resilience: one failing agent doesn't take down the rest.
  • Maintainability: a new return policy touches only the PolicyAgent.
  • Cost control: give each agent the right model. A small model can handle intent classification and parameter extraction, and a stronger model can handle synthesis and nuanced policy answers.

Multi-agent systems also cost something: more LLM calls, more latency, and more places to fail. Use them when the work genuinely splits into distinct skills. A single agent with good tools (Lesson 6) is often enough.

LangGraph: Workflows as Graphs​

LangGraph builds stateful agent workflows as graphs:

  • State: a typed, shared data structure that every node reads and updates.
  • Nodes: agents or processing steps. Each one is a function that takes the state and returns updates to it.
  • Edges: fixed transitions from one node to the next.
  • Conditional edges: a router function inspects the state and picks the next node.
  • Graph: the compiled workflow. It supports cycles, such as retrying or looping back for clarification, plus checkpointing and human-in-the-loop steps.
flowchart LR
  START([start]) --> AI[analyze_intent]
  AI -- "search / both / unclear" --> S[search_products]
  AI -- "qa only" --> Q[answer_questions]
  S -- "needs QA" --> Q
  S -- "no QA" --> SY[synthesize_response]
  Q --> SY --> E([end])

Intent analysis with structured output​

Keyword matching is brittle. Instead, have a small model fill in a schema describing the request, using function calling or structured output, and keep a safe fallback:

from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI

class Intent(BaseModel):
    search_needed: bool = Field(description="User wants products found or compared")
    qa_needed: bool = Field(description="User asks about policies, care, sizing, shipping, warranty")
    reasoning: str

intent_llm = ChatOpenAI(model=ROUTER_MODEL).with_structured_output(Intent)   # small, cheap model

async def analyze_intent(query: str) -> Intent:
    try:
        return await intent_llm.ainvoke(f"Classify this customer request: {query}")
    except Exception:
        return Intent(search_needed=True, qa_needed=False, reasoning="fallback")

with_structured_output binds the schema for you. If you call tools directly, bind them first with llm.bind_tools(tools, tool_choice=...). Passing tools= straight to ainvoke won't bind them.

Conditional routing​

def route_after_analysis(state: "ShopBotState") -> str:
    if state["needs_search"] and state["needs_qa"]:
        return "both_needed"
    if state["needs_search"]:
        return "search_only"
    if state["needs_qa"]:
        return "qa_only"
    return "unclear"

The router is a plain, deterministic function of the state. That makes it easy to unit-test, which matters below.

Practice: build the graph engine​

To see what LangGraph does under the hood, implement a tiny graph runner. Graph.run(state, max_steps) should:

  • Start at the entry node. At each step, call the node's function with the current state, and merge the dict it returns into the state. Nodes return updates, not whole states.
  • Record each visited node in a trace.
  • Choose the next node:
    • If the node has conditional edges, call the router and look its result up in the mapping. An unknown result raises ValueError(f'router returned unknown route: {key}').
    • Otherwise follow its plain edge. A node with no outgoing edge raises ValueError(f'no edge out of {node}').
  • Stop and return (state, trace) when the next node is END.
  • Raise RuntimeError('step limit reached') after max_steps nodes, which guards against runaway cycles.

The wrapper wires up a ShopBot with fake agents and runs one query per input line. The special inputs @badroute and @cycle test the error paths.

Because routing is ordinary code over state, you can test every path with fake agents, with no LLM involved. Also note the unclear route. It goes straight to a clarifying question instead of guessing a search, which is better than the source's default of sending unclear requests to search.

Building ShopBot​

The policy agent's knowledge base (RAG)​

Customers ask about returns, warranties, care, and sizing as well as products. Ground those answers in your documents (Lesson 2):

from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

TOPIC_KEYWORDS = {
    "return_policy": ["return", "refund", "exchange"],
    "shipping": ["shipping", "delivery"],
    "product_care": ["material", "care", "wash"],
    "sizing": ["sizing", "fit", "size"],
}

class BusinessKnowledgeRAG:
    def __init__(self, path: str = "business_knowledge.txt"):
        docs = TextLoader(path).load()
        chunks = RecursiveCharacterTextSplitter(               # policies are dense: smaller chunks
            chunk_size=800, chunk_overlap=100,
            separators=["\n\n", "\n", ".", "!", "?", " "]).split_documents(docs)
        for i, c in enumerate(chunks):
            c.metadata.update({"chunk_id": i, "source": path, "topic": self._topic(c.page_content)})
        self.store = FAISS.from_documents(chunks, OpenAIEmbeddings())

    @staticmethod
    def _topic(text: str) -> str:
        lowered = text.lower()
        return next((t for t, kws in TOPIC_KEYWORDS.items() if any(k in lowered for k in kws)), "general")

    def context(self, query: str, max_chars: int = 2000) -> str:
        expanded = query + {"return": " refund exchange", "shipping": " delivery time cost",
                            "care": " cleaning maintenance"}.get(
            next((k for k in ("return", "shipping", "care") if k in query.lower()), ""), "")
        parts, used = [], 0
        for d in self.store.similarity_search(expanded, k=3):
            if used + len(d.page_content) > max_chars:
                break
            parts.append(d.page_content)
            used += len(d.page_content)
        return "\n\n".join(parts)

Two techniques here:

  • Topic tags on each chunk enable filtering later (Lesson 3).
  • Query expansion adds synonyms: a question about "returns" also searches for refund and exchange.

A shared response type​

from dataclasses import dataclass, field

@dataclass
class AgentResponse:
    agent_name: str
    content: str
    products: list[dict] = field(default_factory=list)   # never a shared mutable default
    ok: bool = True

SearchAgent​

The search agent extracts structured parameters with a small model, runs the catalog search (the word-matching search from Lesson 7, or that MCP server itself), and falls back to a plain search if extraction fails:

from typing import Optional
from pydantic import BaseModel

class SearchParams(BaseModel):
    query: str
    max_price: Optional[float] = None

class SearchAgent:
    name = "SearchAgent"

    def __init__(self, catalog_search):
        self.search = catalog_search                           # e.g. Lesson 7's search_products
        self.extract = ChatOpenAI(model=ROUTER_MODEL).with_structured_output(SearchParams)

    async def process_query(self, text: str) -> AgentResponse:
        try:
            p = await self.extract.ainvoke(f"Extract the product search and any price limit: {text}")
            products = self.search(p.query, p.max_price)
        except Exception:                                      # extraction failed: still search
            products = self.search(text, None)
        return AgentResponse(self.name, f"{len(products)} matching products", products)
Bugs in the source's agents
  • Silent extraction failure. The source reads response.tool_calls[0]["function"]["arguments"]. LangChain's tool_calls entries keep the arguments under "args", so that lookup raises on every call, and a bare except: quietly falls back. Price limits were never actually extracted.
  • Template injection. It builds ChatPromptTemplate messages with f-strings. Any { or } in the retrieved context breaks formatting. Use template variables and pass the values to format_messages(...).
  • Missing helpers. _route_after_analysis and _assess_confidence are called but never defined.

PolicyAgent (QA)​

from langchain_core.prompts import ChatPromptTemplate

POLICY_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "You are a customer service representative. Answer ONLY from the context about "
               "returns, shipping, warranties, care, and sizing. Cite the policy you use. "
               "If the context doesn't cover it, say so and suggest contacting support."),
    ("human", "Context:\n{context}\n\nCustomer question: {question}"),   # variables, not f-strings
])

class PolicyAgent:
    name = "PolicyAgent"

    def __init__(self, rag: BusinessKnowledgeRAG):
        self.rag = rag
        self.llm = ChatOpenAI(model=SYNTHESIS_MODEL)       # stronger model for nuanced policy answers

    async def process_query(self, question: str) -> AgentResponse:
        messages = POLICY_PROMPT.format_messages(context=self.rag.context(question), question=question)
        return AgentResponse(self.name, (await self.llm.ainvoke(messages)).content)

VisionAgent​

Customers sometimes show rather than describe: "Something like this, but waterproof and under $120." A vision model (any multimodal LLM, or a captioning model) turns the image into a short description, such as "brown leather hiking boots." VisionAgent then combines that caption with the user's text and reuses SearchAgent, with no new search logic:

class VisionAgent:
    name = "VisionAgent"

    def __init__(self, search_agent: SearchAgent):
        self.search_agent = search_agent

    async def process(self, image_caption: str, user_text: str = "") -> AgentResponse:
        return await self.search_agent.process_query(f"{image_caption} {user_text}".strip())

The captioning step runs first, in your app code or a preprocessing node, and VisionAgent receives the caption as input.

Orchestrating with LangGraph​

from typing import Annotated, Any
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph import END, StateGraph
from langgraph.graph.message import add_messages

class ShopBotState(TypedDict, total=False):
    messages: Annotated[list[BaseMessage], add_messages]
    user_query: str
    image_caption: str
    needs_search: bool
    needs_qa: bool
    search_results: list[dict[str, Any]]
    qa_response: str
    notices: list[str]          # user-safe notes about anything that failed
    final_response: str

SYNTH_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful shopping assistant. Combine the product results and policy "
               "information into one friendly reply. Mention specific products and prices when "
               "present. Never mention internal agents or errors; use the notices as given."),
    ("human", "Customer: {query}\nProducts: {products}\nPolicy info: {policy}\nNotices: {notices}"),
])

class ShopBot:
    def __init__(self, search: SearchAgent, policy: PolicyAgent, vision: VisionAgent):
        self.search, self.policy, self.vision = search, policy, vision
        self.synth_llm = ChatOpenAI(model=SYNTHESIS_MODEL)
        self.graph = self._build()

    def _build(self):
        g = StateGraph(ShopBotState)
        g.add_node("analyze_intent", self._analyze)
        g.add_node("search_products", self._search)
        g.add_node("answer_questions", self._answer)
        g.add_node("synthesize_response", self._synthesize)
        g.set_entry_point("analyze_intent")
        g.add_conditional_edges("analyze_intent", route_after_analysis, {
            "search_only": "search_products", "both_needed": "search_products",
            "qa_only": "answer_questions", "unclear": "synthesize_response"})
        g.add_conditional_edges("search_products",
                                lambda s: "needs_qa" if s.get("needs_qa") else "synthesis",
                                {"needs_qa": "answer_questions", "synthesis": "synthesize_response"})
        g.add_edge("answer_questions", "synthesize_response")
        g.add_edge("synthesize_response", END)
        return g.compile()

    async def _analyze(self, s: ShopBotState) -> dict:
        intent = await analyze_intent(s["user_query"])
        return {"needs_search": intent.search_needed or bool(s.get("image_caption")),
                "needs_qa": intent.qa_needed}

    async def _search(self, s: ShopBotState) -> dict:          # handles text and image queries
        try:
            if s.get("image_caption"):
                r = await self.vision.process(s["image_caption"], s["user_query"])
            else:
                r = await self.search.process_query(s["user_query"])
            return {"search_results": r.products}
        except Exception:
            return {"search_results": [],
                    "notices": s.get("notices", []) + ["Product search is unavailable right now."]}

    async def _answer(self, s: ShopBotState) -> dict:
        try:
            return {"qa_response": (await self.policy.process_query(s["user_query"])).content}
        except Exception:
            return {"qa_response": "",
                    "notices": s.get("notices", []) + ["Policy details are unavailable; "
                                                       "please contact support."]}

    async def _synthesize(self, s: ShopBotState) -> dict:
        products = "; ".join(f"{p['title']} (${p['price']:.2f})" for p in s.get("search_results", [])[:3])
        reply = await self.synth_llm.ainvoke(SYNTH_PROMPT.format_messages(
            query=s["user_query"], products=products or "none", policy=s.get("qa_response") or "none",
            notices="; ".join(s.get("notices", [])) or "none"))
        return {"final_response": reply.content}

    async def chat(self, text: str, image_caption: str = "") -> str:
        state = {"messages": [HumanMessage(content=text)], "user_query": text,      # fresh per request
                 "image_caption": image_caption, "search_results": [], "notices": []}
        return (await self.graph.ainvoke(state))["final_response"]

What changed from the source, and why:

  • Nodes return partial updates ({"search_results": …}) instead of mutating and returning the whole state. That's LangGraph's model, and it keeps each node's effect explicit.
  • Synthesis receives product titles and prices. The source passed only "Found 3 products," so the final answer couldn't name a single product.
  • Vision is wired in. The source defined a _vision_search node but never added it to the graph. Here the search node uses the caption whenever one is present.
  • Failures become user-safe notices, never stack traces. Synthesis is told to use them as written.
  • Fresh state for every request, so nothing leaks between conversations. There's an exercise on this below.
flowchart LR
  IN["Text and/or image"] --> R["LangGraph router<br/>intent analysis"]
  R --> SA["Search (text)<br/>or Vision → Search"]
  R --> QA["PolicyAgent<br/>RAG answers"]
  SA --> SY["Synthesis<br/>one customer reply"]
  QA --> SY

Running it​

import asyncio

async def demo(bot: ShopBot):
    for q in ["I need waterproof hiking gear under $150",
              "What's your return policy for outdoor gear?",
              "I want a down jacket but need to know how to care for it",
              "Show me trail running shoes and explain your shipping policy"]:
        print(f"\n> {q}\n{await bot.chat(q)}")
    print(await bot.chat("Do you have something like this but waterproof under $120?",
                         image_caption="brown leather hiking boots with laces"))

Practice: degrade gracefully​

When one agent fails, the customer should still get everything that worked, plus a plain-language note, and never the raw exception. Implement two functions:

  • run_agents(agents, query): agents is a list of (name, fn) pairs. Call each fn(query). Collect the successful outputs in a dict, and collect the names of agents that raised in a list. Return (outputs, failed).
  • synthesize(outputs, failed): join the successful outputs in agent order with ' '. For each failed agent, append FALLBACKS[name]. If nothing succeeded, return "Sorry, I can't help with that right now. Please try again shortly."

The wrapper defines a product agent and a policy agent. Input: query|failing_agents, where the failing agents are a comma-separated list or blank.

Check: orchestration​

Quiz

Graphs, state, and agents

1 / 3

In LangGraph, what should a node function return?

Pick an answer to see if you are right.

Testing Multi-Agent Workflows​

Test the workflows, not just the agents. Bugs live at the seams: when agents hand off, when state updates, and when one component fails.

flowchart LR
  B["Blended queries<br/>vision + search + QA"] --> T{Workflow tests}
  A["Ambiguous intent<br/>routing + fallbacks"] --> T
  F["Agent failure<br/>graceful degradation"] --> T
  S["State flow<br/>details preserved,<br/>no contamination"] --> T
  R["Regression<br/>a test for every bug"] --> T

The source writes these tests as assertions on the LLM's final prose, for example assert "$150" in response.lower() or assert "error" not in response.lower(). Tests like that are flaky: the model may write "under 150 dollars," legitimately use the word "error," or phrase a correct answer differently. They're also slow and costly, since every run calls models.

Test in layers instead:

  1. Deterministic workflow tests, which run on every commit. Replace the LLM-backed pieces with fakes and assert on state and routing: which nodes ran, what each wrote, and what reached synthesis.
  2. LLM evaluations, which run on a schedule. Score real end-to-end answers with rubrics or an LLM judge on a curated dataset, as in Lesson 2, and track the pass rates.
  3. Regression cases: every bug you fix becomes a test.

Here are deterministic versions of the source's five test categories, using fakes:

import pytest

class FakeSearch:
    def __init__(self, results=None, fail=False):
        self.results, self.fail, self.calls = results or [], fail, []
    async def process_query(self, text):
        self.calls.append(text)
        if self.fail:
            raise TimeoutError("Product database timeout")
        return AgentResponse("SearchAgent", "", self.results)

class FakePolicy:
    def __init__(self, answer="30-day returns", fail=False):
        self.answer, self.fail, self.calls = answer, fail, []
    async def process_query(self, q):
        self.calls.append(q)
        if self.fail:
            raise ConnectionError("Knowledge base unavailable")
        return AgentResponse("PolicyAgent", self.answer)

@pytest.fixture
def bot(monkeypatch):
    async def fake_intent(q):                       # deterministic intent for tests
        ql = q.lower()
        return Intent(search_needed=any(w in ql for w in ("boot", "jacket", "shoe")),
                      qa_needed=any(w in ql for w in ("return", "policy", "fit", "shipping")),
                      reasoning="test")
    monkeypatch.setattr("shopbot.analyze_intent", fake_intent)   # module where ShopBot lives
    search = FakeSearch([{"title": "Trail Boot", "price": 95.0}])
    b = ShopBot(search, FakePolicy(), VisionAgent(search))
    b._synthesize = lambda s: {"final_response": s}    # capture what synthesis receives
    b.graph = b._build()
    return b

# 1. Blended query: vision + search + policy all contribute
async def test_blended(bot):
    state = (await bot.graph.ainvoke({"user_query": "like these boots, return policy?",
                                      "image_caption": "brown leather boots",
                                      "search_results": [], "notices": []}))["final_response"]
    assert state["search_results"] and state["qa_response"]
    assert bot.search.calls == ["brown leather boots like these boots, return policy?"]

# 2. Ambiguous intent routes to a clarifying answer, not a blind search
async def test_unclear_routes_to_synthesis(bot):
    state = (await bot.graph.ainvoke({"user_query": "Can you help me?", "notices": []}))["final_response"]
    assert bot.search.calls == [] and bot.policy.calls == []

# 3. Agent failure: partial results preserved, user-safe notice added
async def test_policy_failure(bot):
    bot.policy.fail = True
    state = (await bot.graph.ainvoke({"user_query": "jackets and returns",
                                      "search_results": [], "notices": []}))["final_response"]
    assert state["search_results"] and "contact support" in state["notices"][0]

# 4. State flow: the user's constraints reach the search agent intact
async def test_constraints_reach_search(bot):
    await bot.graph.ainvoke({"user_query": "size 10 men's waterproof boots under $120", "notices": []})
    assert "under $120" in bot.search.calls[0] and "size 10" in bot.search.calls[0]

(Run them with pytest-asyncio or a similar async test plugin.) The fifth category, no contamination between conversations, comes from a classic bug that's worth practicing on its own.

Practice: isolate state per request​

A shared mutable default is a classic way for one customer's data to leak into another's reply. The DEFAULT_STATE template below holds lists. Copying it shallowly (dict(DEFAULT_STATE)), or using it directly, means every request appends to the same list objects.

Implement new_state(query) so that it returns a fully independent state built from DEFAULT_STATE, with user_query set. Mutating the result must never change DEFAULT_STATE or another request's state. copy.deepcopy is one way to do it.

The wrapper runs two conversations in a row through a fake search that appends to state['search_results']. It prints each reply, then checks that the template is still pristine. Input: one query per line.

With a shallow copy, the second test would show two jackets on the second request, and the template's intent.search would be stuck at True. That's exactly what the source's "red jacket" contamination test is trying to catch, now caught deterministically.

Regression habit: when a bug ships, write the test that would have caught it, then fix it. Rerun the whole suite every time you add an agent, model, or tool.

Check: testing​

Quiz

Testing at the seams

1 / 2

Why is `assert "$150" in response` a weak test for a multi-agent system?

Pick an answer to see if you are right.

An Alternative: CrewAI​

CrewAI takes a different view. Where LangGraph has you draw the flowchart (state, nodes, edges), CrewAI has you staff a team: agents get roles, goals, and backstories, tasks get descriptions and expected outputs, and the framework coordinates them.

pip install crewai
from crewai import Agent, Crew, Process, Task
from crewai.tools import tool        # newer CrewAI; older versions exported it from crewai_tools

@tool("Search Products")
def search_products(query: str, max_price: float | None = None) -> str:
    """Search the product catalog for in-stock items matching the query (USD max_price optional)."""
    results = catalog_search(query, max_price)        # reuse Lesson 7's word-matching search
    return "\n".join(f"- {p['title']}: ${p['price']:.2f}" for p in results[:3]) or "No products found."

@tool("Lookup Policy")
def lookup_policy(topic: str) -> str:
    """Look up business policies about returns, shipping, warranty, or sizing."""
    policies = {"return": "30-day returns; items unworn with tags.",
                "shipping": "Free shipping over $50; standard delivery 3-5 days.",
                "warranty": "1-year warranty on outdoor gear against manufacturing defects.",
                "sizing": "See our size guide; we recommend your usual size."}
    return next((v for k, v in policies.items() if k in topic.lower()),
                "Please contact customer service for detailed policy information.")

product_agent = Agent(role="Product Specialist",
                      goal="Find the right products for each customer's needs",
                      backstory="An outdoor-gear expert who knows the catalog inside out.",
                      tools=[search_products], llm=ROUTER_MODEL)
service_agent = Agent(role="Customer Service Representative",
                      goal="Give accurate policy information",
                      backstory="A friendly rep who knows every company policy.",
                      tools=[lookup_policy], llm=ROUTER_MODEL)
coordinator = Agent(role="Response Coordinator",
                    goal="Combine findings into one clear, friendly reply",
                    backstory="Writes seamless customer replies without mentioning internal processes.",
                    llm=SYNTHESIS_MODEL)

def handle_customer_query(query: str) -> str:
    search_task = Task(description=f"If this asks about products, search for them: {query}",
                       expected_output="Product recommendations, or a note that none were requested",
                       agent=product_agent)
    policy_task = Task(description=f"If this needs policy information, look it up: {query}",
                       expected_output="Relevant policy details, or a note that none were needed",
                       agent=service_agent)
    reply_task = Task(description=f"Write the customer reply to: {query}. Be specific; acknowledge gaps.",
                      expected_output="A complete, helpful customer reply",
                      agent=coordinator, context=[search_task, policy_task])   # depends on both
    crew = Crew(agents=[product_agent, service_agent, coordinator],
                tasks=[search_task, policy_task, reply_task],
                process=Process.sequential)            # or Process.hierarchical with a manager agent
    return crew.kickoff().raw

(Recent CrewAI versions take model names directly as llm= strings, or a CrewAI LLM object, rather than LangChain chat models. Check your version's documentation.)

LangGraph or CrewAI?​

LangGraphCrewAI
Mental modelEngineer drawing a state machineManager staffing a team
ControlExplicit state, routing, and cyclesRole-based, with framework-managed coordination
Best forComplex branching, retries, human approval steps, shared infrastructureFast prototypes, mostly linear task flows
Trade-offMore code, steeper learning curveLess visibility and fine-grained control

Other options include AutoGen (Microsoft; agents that converse with each other), Agency Swarm (autonomy-focused), and Haystack (pipelines, strong for RAG and search).

The framework matters less than the patterns: intent analysis, specialization, explicit state, failure handling, and synthesis. Master those and you can build with any tool. Start simple, and add agents only when a single one really struggles.

Check: frameworks​

Quiz

Choosing a framework

1 / 2

Your workflow retries a search with relaxed filters when results are empty, then asks the user to confirm before placing an order. Which framework fits best?

Pick an answer to see if you are right.

Key Takeaways​

  • One agent doing everything becomes mediocre at everything. Specialized agents (search, policy, vision) with a coordinator give modularity, extensibility, resilience, and per-agent model choice, at the cost of more calls and more seams.
  • LangGraph models workflows as state + nodes + (conditional) edges. Nodes return partial updates, routers are plain functions, and graphs can cycle. Guard cycles with step limits.
  • Use structured output for intent and parameter extraction (with_structured_output / bind_tools), read tool_calls[...]["args"], and use template variables rather than f-strings in prompts.
  • Synthesis needs the details (titles, prices, policy text), not counts. Wire every agent into the graph, including vision.
  • Turn failures into user-safe notices, keep the partial results, and build fresh state for every request.
  • Test the seams deterministically with fakes (routing, state flow, failure injection, isolation), evaluate the prose with judges on a schedule, and turn every bug into a regression test.
  • CrewAI trades control for speed with role-based crews. Choose by workflow complexity, and remember that the patterns outlast any framework.

Previous lesson: Tool Integration and MCP.

Next lesson: Evaluation and Performance — measure hallucinations, speed up and cut costs with streaming, batching, caching, and routing, and evaluate agent trajectories.

References​

  1. LangGraph documentation
  2. LangChain: structured output
  3. CrewAI documentation
  4. Microsoft AutoGen
  5. Haystack documentation