linghua jin This creates true collaboration: Researcher grounds in facts (RAG), critic ensures quality, writer synthesizes. In production, add more agents (e.g., fact-checker with web tool) or structured outputs for reliable routing.This pattern scales to complex tasks like research papers, code debugging teams, or enterprise workflows. If you want extensions (e.g., hierarchical teams, human-in-loop, or full tool integration), let me know! 🚀angGraph excels at building multi-agent systems by defining multiple specialized agents as nodes in a graph, with a supervisor/orchestrator managing the flow. This creates collaborative workflows where agents handle distinct roles (e.g., researcher, critic, writer) and pass state between them.A classic pattern is the Supervisor-Agent Team (inspired by LangChain/LangGraph docs and popular 2025 tutorials): A supervisor agent decomposes tasks, routes to worker agents (e.g., one for retrieval/RAG, one for analysis, one for synthesis), and decides when to finalize.Key Benefits for Multi-Agent in LangGraphState Sharing: All agents access/modify a shared state (e.g., messages, retrieved docs, research notes).
Hierarchical Control: Supervisor uses conditional routing to delegate and loop.
Scalability: Easy to add agents (e.g., fact-checker, tool specialist).
Persistence: Checkpointing for long-running research tasks.
Full Working Example: Multi-Agent Research Team (Agentic RAG Style)This example creates a team for complex queries:Researcher Agent: Handles retrieval (RAG tool).
Critic Agent: Grades/Reflects on quality.
Writer Agent: Generates final answer.
Supervisor Agent: Routes between them, decides when done.
Assumes LangGraph, LangChain, OpenAI, and a vector store (Chroma). Install: pip install langgraph langchain langchain-openai chromadb.pythonfrom typing import Literal, TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from langchain_core.tools import tool
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
import operator
# Setup retriever (mock docs)
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(collection_name="multi_rag", embedding_function=embeddings)
vectorstore.add_texts([
"LangGraph is a framework for building stateful multi-agent systems.",
"RAG improves LLM accuracy by retrieving external knowledge.",
"Multi-agent collaboration reduces hallucinations."
])
@tool
def retrieve(query: str) -> str:
"""Retrieve relevant documents."""
docs = vectorstore.similarity_search(query, k=3)
return "\n\n".join([
doc.page_content for doc in docs])
tools = [retrieve]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)
# Shared State
class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], operator.add]
next: str # Who to route to next
# Worker Agents
def researcher(state: AgentState):
messages = state["messages"]
response = llm_with_tools.invoke(messages)
return {"messages": [response]}
def critic(state: AgentState):
# Simple critique: ask LLM to evaluate relevance
last_msg = state["messages"][-1].content
critique_prompt = f"Critique this research output for relevance and completeness:\n{last_msg}\nSuggest improvements if needed."
response = llm.invoke(critique_prompt)
return {"messages": [AIMessage(content=f"Critique: {response.content}")]}
def writer(state: AgentState):
messages = state["messages"]
write_prompt = f"Synthesize a clear, final answer using all prior research:\n{messages[-3:]}" # Last few exchanges
response = llm.invoke(write_prompt)
return {"messages": [response]}
# Supervisor (routes and decides end)
members = ["researcher", "critic", "writer"]
system_prompt = (
f"You are a supervisor managing a team: {', '.join(members)}. "
"Route the task to the best agent or FINISH when complete. "
"Only route to one at a time. Use FINISH when the writer has produced a final answer."
)
options = ["FINISH"] members
def supervisor(state: AgentState):
messages = state["messages"]
supervisor_chain = llm.bind_tools([]) # No tools needed
response = supervisor_chain.invoke(
[{"role": "system", "content": system_prompt}] messages
)
# Parse next route (in real: use structured output or tool calling)
route = response.content.lower()
next_agent = "FINISH"
for member in members:
if member in route:
next_agent = member
break
return {"next": next_agent, "messages": [response]}
# Build Graph
workflow = StateGraph(AgentState)
workflow.add_node("supervisor", supervisor)
workflow.add_node("researcher", researcher)
workflow.add_node("critic", critic)
workflow.add_node("writer", writer)
# Routing from supervisor
workflow.set_entry_point("supervisor")
for member in members:
workflow.add_edge(member, "supervisor") # Always return to supervisor
# Conditional routing
def route_next(state: AgentState):
return state["next"]
workflow.add_conditional_edges(
"supervisor",
route_next,
{"researcher": "researcher", "critic": "critic", "writer": "writer", "FINISH": END}
)
# Compile with memory
memory = SqliteSaver.from_conn_string(":memory:")
graph = workflow.compile(checkpointer=memory)
# Run example
config = {"configurable": {"thread_id": "team1"}}
inputs = {"messages": [HumanMessage(content="Explain how multi-agent systems improve RAG accuracy.")]}
for event in
graph.stream(inputs, config):
for key, value in event.items():
if key != "__end__":
print(f"{key.upper()}: {value.get('messages', [-1]).content[:200]}...\n")
How It WorksUser query → Supervisor decides first agent (likely researcher).
Researcher retrieves docs → Returns to supervisor.
Supervisor routes to critic → Critique → Back.
Eventually routes to writer → Final answer → Supervisor sees "FINISH" → Ends.
Loop continues until supervisor chooses FINISH.