A multi-agent system coordinates specialized AI agents to handle a request that crosses several domains. A shopping question might require product search, an answer grounded in return policies, and analysis of an uploaded photo. Keeping those jobs separate avoids loading one agent with every tool and policy; an orchestrator decides which specialists to call and combines their results. section 8.0 section 8.1.1
A practical workflow separates responsibilities:
- Analyze intent and route the request to search, policy Q&A, vision processing, or a combination. section 8.1.7
- Let search return structured products, while policy Q&A retrieves business documents before generating an answer. section 8.1.4
- Preserve intermediate work in shared state, then synthesize one response rather than exposing disconnected agent outputs. section 8.1.7
LangGraph makes the coordination explicit: state holds conversation data and agent results; nodes perform work; edges and conditional edges determine the route. Its error handling supports fallbacks when an agent fails. This improves failure containment, but a coherent final answer still depends on the quality of each agent’s results. section 8.1.2 section 8.1.7
# shopbot_langgraph.py
from typing import Dict, Any, List, Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from search_agent import SearchAgent
from qa_agent import QAAgent
import json
import asyncio
class ShopBotState(TypedDict): #1
messages: Annotated[List[BaseMessage], add_messages]
user_query: str
search_results: List[Dict[str, Any]]
qa_response: str
needs_search: bool
needs_qa: bool
final_response: str
intent_analysis: Dict[str, Any]
class ShopBotLangGraph:
def __init__(self):
self.search_agent = SearchAgent()
self.qa_agent = QAAgent()
self.intent_llm = ChatOpenAI(model="gpt-5.4-mini”) #2
self.synthesis_llm = ChatOpenAI(model="gpt-5.4") #2
self.graph = self._build_graph()
def _build_graph(self) -> StateGraph: #3
workflow = StateGraph(ShopBotState)
workflow.add_node("analyze_intent", self._analyze_intent)
workflow.add_node("search_products", self._search_products)
workflow.add_node("answer_questions", self._answer_questions)
workflow.add_node("synthesize_response", self._synthesize_response)
workflow.set_entry_point("analyze_intent")
workflow.add_conditional_edges(
"analyze_intent",
self._route_after_analysis,
{
"search_only": "search_products",
"qa_only": "answer_questions",
"both_needed": "search_products",
"unclear": "search_products"
}
)
workflow.add_conditional_edges(
"search_products",
self._check_qa_needed,
{
"needs_qa": "answer_questions",
"synthesis": "synthesize_response"
}
)
workflow.add_edge("answer_questions", "synthesize_response")
workflow.add_edge("synthesize_response", END)
return workflow.compile()
async def _analyze_intent(self, state: ShopBotState) -> ShopBotState: #4
intent_result = await analyze_intent(state["user_query"])
state["needs_search"] = intent_result["search_needed"]
state["needs_qa"] = intent_result["qa_needed"]
state["intent_analysis"] = {
"reasoning": intent_result["reasoning"],
"confidence": intent_result.get("confidence", 0.5)
}
return state
async def _search_products(self, state: ShopBotState) -> ShopBotState: #5
try:
search_response = await
self.search_agent.process_query(state["user_query"])
state["search_results"] = search_response.products_mentioned
or []
if any(phrase in search_response.content.lower() for phrase in
["care", "return", "policy", "warranty", "sizing"]):
state["needs_qa"] = True
except Exception as e:
state["search_results"] = []
return state
async def _answer_questions(self, state: ShopBotState) -> ShopBotState: #6
try:
qa_response = await
self.qa_agent.process_query(state["user_query"])
state["qa_response"] = qa_response.content
except Exception as e:
state["qa_response"] = "I apologize, but I
couldn't retrieve that information."
return state
def _check_qa_needed(self, state: ShopBotState) -> str: #7
return "needs_qa" if state["needs_qa"] else "synthesis"
async def _synthesize_response(self, state: ShopBotState)
-> ShopBotState: #8
search_info = f"Found {len(state['search_results'])}
products" if state["search_results"] else "No products found"
qa_info = state["qa_response"] if state["qa_response"]
else "No additional information"
synthesis_prompt = ChatPromptTemplate.from_messages([
("system", """You are a helpful shopping assistant. Combine
search results and Q&A information into a single, coherent
response.
Guidelines:
- Start with the most relevant information
- Present product results and policy information logically
- Use a conversational, helpful tone
- Don't mention internal agents or processes
- Focus on being useful to the customer"""),
("human", f"""Original Query: {state['user_query']}
Search Results: {search_info}
Q&A Information: {qa_info}
Create a unified, helpful response.""")
])
synthesis_response = await
self.synthesis_llm.ainvoke(synthesis_prompt.format_messages())
state["final_response"] = synthesis_response.content
return state
async def chat(self, user_input: str) -> str: #9
initial_state = ShopBotState(
messages=[HumanMessage(content=user_input)],
user_query=user_input,
search_results=[],
qa_response="",
needs_search=False,
needs_qa=False,
final_response="",
intent_analysis={}
)
final_state = await self.graph.ainvoke(initial_state)
return final_state["final_response"]
The Python implementation defines a shared ShopBotState with messages, intent analysis, routing flags, specialist results, and a final response. section 8.1.4
One request, several specialists
For “Find a winter-camping jacket under $200, and explain the warranty,” the coordinator routes the product constraints to search and the warranty question to policy Q&A. It then combines the product results and policy-grounded answer into one reply. section 8.1.1 section 8.1.4
The payoff is modularity: a policy change need not alter product search, and a new capability can be added as another agent. The orchestration choice has a cost. LangGraph requires explicit state and routing code in exchange for fine-grained control; CrewAI offers role-based collaboration with less code and less control. section 8.1.7 section 8.3.1
