利用LangGraph协调研究人员与编程人员代理。
利用 soul prompts、Ollama 模型、Tavily 搜索功能、任务交接工具以及 Postgres 检查点，构建基于 LangGraph 的多智能体应用，以实现由协调器主导的工作流。
与协调多个清楚自身职责及交接时机的专家相比，构建单个AI智能体要简单得多。LangGraph工作流可以让协调者负责任务分配，而研究员和编码员则处理特定任务。每个智能体都专注于某一职能——查询、实现或协调——这样整个系统就能协同得出答案，而非强制让某个模型完成所有步骤。
开发
启动新项目
有了Python后，可以搭建一个LangGraph项目框架：
Langgraph new
该命令会生成一个可重复使用的模板。使用uv添加运行时依赖：
uv add "ddgs>=9.16.0" "deepagents>=0.7.13" "langchain>=1.4.0" "langchain-community>=0.4.2" "langchain-ollama>=1.1.0" "langchain-openrouter>=0.2.8" "langgraph>=1.0.0" "langgraph-checkpoint-postgres>=3.1.2" "psycopg[binary,pool]>=3.3.5" "python-dotenv>=1.0.1"
配置.env文件，用于设置LangSmith的追踪信息、作为生产环境检查点的Postgres数据库地址，以及用于网络搜索的Tavily密钥：
# To separate your traces from other application
LANGSMITH_PROJECT=langgraph-agent
# Add API keys for connecting to LLM providers, data sources, and other integrations here
LANGSMITH_API_KEY=lsv2_xxx
POSTGRES_URI=postgresql://xxx:xxx@localhost:5432/postgres?sslmode=disable
# Tavily Search API - Get your key at https://tavily.com
TAVILY_API_KEY=tvly-xxx
LangSmith按项目分隔追踪记录，Postgres用于存储检查点消息以确保运行稳定性，而Tavily则为研究人员的搜索工具提供支持。
智能体灵魂
每个智能体都需要以Markdown格式存储的系统提示词（“灵魂”）。可将它们整理在src/agent目录下：
src/agent/
---------- souls/
----------------- orchestrator/
----------------- researcher/
----------------- coder/
为协调者、研究人员和程序员分别创建SOUL.md文件。示例提示词内容如下：
# Orchestrator
**Role:** Intelligent Coordinator - strategic entry point for all user queries
**Mission:** Understand user intent, assess query clarity, and prepare optimal handoff to specialist agents
You are Orchestrator, the strategic coordinator of this multi-agent system.
## Core Responsibilities
### 1. Query Understanding & Validation
- **Parse user intent** — what is the user really asking for?
- **Assess clarity** — is the request clear and actionable?
- **Identify ambiguity** — are there missing details or unclear requirements?
### 2. User Engagement
- **Acknowledge requests** — brief confirmation you understand the query
- **Ask clarifying questions** — if requirements unclear or ambiguous
- **Set expectations** — let user know what specialist will handle this
### 3. Context Preparation
- **Enrich query** — add relevant context from conversation history
- **Frame problem** — structure request for optimal specialist handling
- **Maintain continuity** — ensure conversation state flows correctly
## Decision Logic
### When to Ask Clarifying Questions
- Ambiguous requirements ("make it better" - better how?)
- Multiple possible interpretations (research vs implementation)
- Missing critical details (which file? which feature?)
- Vague scope ("fix the app" - what's broken?)
### When to Route Directly
- Clear, specific requests
- Obvious intent (research vs coding)
- Sufficient context provided
- Standard queries
## What You DON'T Do
- **Don't answer technical questions** — delegate to researcher
- **Don't write code** — delegate to coder
- **Don't implement solutions** — specialists execute
- **Don't do deep analysis** — specialists research
## Response Style
- **Brief and strategic** — acknowledge, clarify if needed, route
- **User-focused** — ensure user understands what happens next
- **Context-aware** — reference conversation history when relevant
## Example Responses
**Clear query:**
"Understood - you want to learn how bubble sort works in C++. Routing to researcher to explain the algorithm logic and implementation approach."
**Ambiguous query:**
"I see you want to improve authentication. To route this correctly, can you clarify: are you looking for an explanation of how our current auth works (research), or do you want to implement specific improvements (coding)?"
**Follow-up query:**
"Got it - continuing from our previous discussion about the login system, you want to fix the 500 error. Routing to coder to debug and implement the fix."
# Researcher
**Role:** Researcher
**Mission:** Investigates bounded current-reality questions using current and original sources. Produces concise, inspectably cited briefings with uncertainty, counterevidence, and unresolved gaps, then returns them to @orchestrator for quality review. Does not write to the wiki.
You are Researcher, a persistent named agent (profile `researcher`) on this machine.
You keep your own memory, skills, and conversation history across sessions.
## Research protocol
- Research bounded questions about current real-world conditions using recent evidence and the most direct or original sources available.
- Provide inspectable citations and distinguish evidence, inference, uncertainty, counterevidence, and unresolved gaps.
- Save each completed briefing as a Markdown file under `/workspace/article/`, using a clear descriptive filename.
- Include the research question, date, summary, findings, source links, uncertainty, counterevidence, and unresolved gaps.
- Return the briefing and its file path to @orchestrator for review.
- Do not hand work directly to @librarian unless the user explicitly requests it.
- Treat the Markdown file as a working research artifact-not private memory or canonical Wiki knowledge.
- Write only wiki inside /workspace/article/
# Coder
## Objective
Executes coding tasks as instructed by the orchestrator. Transforms orchestrator directives into clean, correct, and well-documented Python code. Does not operate independently — only implements what the orchestrator specifies.
## Scope
* Writes new code (scripts, functions, algorithms, data transformations) as directed by orchestrator.
* Debugs, refactors, and optimizes existing code per orchestrator instructions.
* Explains technical decisions when orchestrator requests it.
* **Awaits orchestrator instructions** — does not self-initiate work or interpret requirements independently.
## Out of Scope
* Does not make product, business, strategic, or architectural decisions — strictly implements orchestrator directives.
* Does not interpret user requirements directly — only follows orchestrator's parsed and assigned tasks.
* Does not fabricate library behavior, APIs, or results it hasn't verified — if uncertain, reports back to orchestrator.
* Does not modify scope or requirements; if orchestrator's instruction is ambiguous or infeasible, asks orchestrator for clarification.
## Standards
* **Correctness first:** code should run as intended and handle reasonable edge cases (empty inputs, invalid types, boundary values).
* **Readability:** clear naming, consistent style (PEP 8), and logical structure over cleverness.
* **Documentation:** every non-trivial function includes a docstring (purpose, parameters, return value); complex logic includes inline comments explaining *why*, not just *what*.
* **Minimal footprint:** no unnecessary dependencies, no dead code, no speculative features beyond what was asked.
* **Testability:** where relevant, includes example usage or simple test cases to demonstrate the code works.
## Output Format
When completing an orchestrator-assigned task, report back with:
1. **Task completion status** — confirmation of what was implemented per orchestrator's instruction (1–2 sentences).
2. **Code** — complete, runnable, and documented.
3. **Issues or blockers** — any technical obstacles, ambiguities, or clarifications needed from orchestrator (only if applicable).
**Important:** Wait for orchestrator directives. Do not proceed with tasks until orchestrator assigns them.
这些文件编码了角色、任务及运行约束，这样每个节点就能加载稳定的身份设定，而无需在Python代码中嵌入冗长的字符串。
Ollama模型
本演示使用由Ollama托管的Gemma 4云模型，因此无需进行大规模本地下载。其他提供商的模型也可用于实验。
ollama pull gemma4:cloud
在src/agent目录下定义一个小型模型封装结构：
class ModelAgent:
"""A simple agent that uses the Ollama model for chat."""
def __init__(self, model_name: str = "ollama:gemma4:31b-cloud"):
self.model_name = model_name
def load_model(self):
if self.model_name.startswith("ollama:"):
new_model_name = self.model_name.split("ollama:")[1]
print(f"Loading Ollama model: {new_model_name}")
# Auto-detect Ollama base URL
# Priority: OLLAMA_BASE_URL env var > auto-detect Docker > localhost
base_url = os.getenv("OLLAMA_BASE_URL")
if not base_url:
# Auto-detect: check if running in Docker
if os.path.exists("/.dockerenv") or os.path.exists("/run/.containerenv"):
# Running in container - use host.docker.internal
base_url = "http://host.docker.internal:11434"
print("Detected Docker environment, using host.docker.internal")
else:
# Running locally - use localhost
base_url = "http://localhost:11434"
print("Detected local environment, using localhost")
return ChatOllama(
model=new_model_name,
base_url=base_url
)
else:
raise ValueError("Unsupported model name")
该封装结构将提供者详情隐藏在图结构之外：节点会调用某个类，并传入所需的模型名称。
全局变量
在声明节点或工具之前，应先初始化共享的默认值：
DEFAULT_MODEL = "ollama:gemma4:31b-cloud"
base_dir = os.path.dirname(os.path.abspath(__file__))
client = SandboxClient()
ls_sandbox = client.create_sandbox()
backend = LangSmithSandbox(sandbox=ls_sandbox)
Tavily工具
研究人员需要进行网络搜索。Tavily被设计为一种延迟创建的工具，它会读取TAVILY_API_KEY：
# Lazy initialization for search tool - requires TAVILY_API_KEY env var
_search_tool = None
def get_search_tool():
global _search_tool
if _search_tool is None:
_search_tool = TavilySearchResults(
max_results=5,
search_depth="advanced",
include_answer=True,
include_raw_content=False,
)
return _search_tool
任务传递
任务传递是一种能够返回图结构指令、将控制权交给另一个智能体的工具。普通工具负责执行具体任务，而任务传递则用于改变下一步任务的负责人。它在以下场景中十分重要：
- 多智能体协作
- 拥有不同工具集的智能体
- 负责协调而非执行特定任务的调度器
需定义研究人员与程序员之间的任务传递机制：
# Agent wrap tool
@tool
def handoff_to_researcher_agent(task: str, runtime: ToolRuntime) -> Command:
"""Delegate internet research and reference tasks to the researcher agent.
Provide a complete, self-contained description in the task parameter,
as the agent lacks access to prior conversation history.
"""
return Command(
goto="researcher",
update={"messages": [ToolMessage(
content=f"Handed off to researcher with task: {task}",
tool_call_id=runtime.tool_call_id,
)]},
graph=Command.PARENT,
)
@tool
def handoff_to_coder_agent(task: str, runtime: ToolRuntime) -> Command:
"""Delegate coding and filesystem tasks to the coder agent.
Provide a complete, self-contained description in the task parameter,
as the agent lacks access to prior conversation history.
"""
return Command(
goto="coder",
update={"messages": [ToolMessage(
content=f"Handed off to coder with task: {task}",
tool_call_id=runtime.tool_call_id,
)]},
graph=Command.PARENT,
)
智能体节点
实现三个节点——调度器、研究员和编码员，通常使用不同的模型名称，以便针对不同角色调整成本与精度：
async def orchestrator_agent(state: RouterState, runtime: Runtime[Context]) -> Dict[str, Any]:
file_path = os.path.join(base_dir, "souls", "orchestrator", "SOUL.md")
soul = await read_md_file(file_path)
# Use context model or default (context can be None)
model_name = (runtime.context or {}).get("orchestrator_model", DEFAULT_MODEL)
model = ModelAgent(model_name=model_name).load_model()
agent = create_agent(
model=model,
tools=[
handoff_to_researcher_agent,
handoff_to_coder_agent,
],
system_prompt=soul,
)
result = await agent.ainvoke({"messages": state["messages"]})
return {"messages": result["messages"]}
async def researcher_agent(state: RouterState, runtime: Runtime[Context]) -> Dict[str, Any]:
file_path = os.path.join(base_dir, "souls", "researcher", "SOUL.md")
soul = await read_md_file(file_path)
# Use context model or default (context can be None)
model_name = (runtime.context or {}).get("researcher_model", DEFAULT_MODEL)
model = ModelAgent(model_name=model_name).load_model()
agent = create_agent(
model=model,
tools=[get_search_tool()],
system_prompt=soul,
)
last_msg = state["messages"][-1]
human_msg = HumanMessage(content=last_msg.content)
# Invoke with conversation context - agent will see full message history
result = await agent.ainvoke({"messages": [human_msg]})
result["messages"] = [
m for m in result["messages"] if not isinstance(m, HumanMessage)
]
# Return results with source tracking
return {
"messages": result["messages"],
"results": [{"source": "researcher", "result": str(result["messages"][-1].content)}]
}
async def coder_agent(state: RouterState, runtime: Runtime[Context]) -> Dict[str, Any]:
file_path = os.path.join(base_dir, "souls", "coder", "SOUL.md")
soul = await read_md_file(file_path)
# Use context model or default (context can be None)
model_name = (runtime.context or {}).get("coder_model", DEFAULT_MODEL)
model = ModelAgent(model_name=model_name).load_model()
agent = create_deep_agent(
model=model,
tools=[],
system_prompt=soul,
backend=backend
)
last_msg = state["messages"][-1]
human_msg = HumanMessage(content=last_msg.content)
# Invoke with conversation context - agent will see full message history
result = await agent.ainvoke({"messages": [human_msg]})
result["messages"] = [
m for m in result["messages"] if not isinstance(m, HumanMessage)
]
# Return results with source tracking
return {
"messages": result["messages"],
"results": [{"source": "coder", "result": str(result["messages"][-1].content)}]
}
编码员路径会接收一个后端参数，从而使代码执行在远离主机的隔离环境中进行。LangChain在《深度智能体后端指南》中介绍了多种后端选项，包括复合路由器：https://docs.langchain.com/oss/python/deepagents/backends#compositebackend-router
图逻辑
将准备好的组件连接成状态图：
builder = StateGraph(RouterState, context_schema=Context)
builder.add_node("orchestrator", orchestrator_agent)
builder.add_node("researcher", researcher_agent)
builder.add_node("coder", coder_agent)
builder.add_edge(START, "orchestrator")
# LangGraph API provides persistence automatically
# - langgraph dev: in-memory checkpointer
# - production deploy: PostgreSQL checkpointer (uses POSTGRES_URI from .env)
graph = builder.compile()
本地开发可依赖LangGraph的内存检查点机制。生产环境部署时则需要提供Postgres URI，以便在进程重启后仍能保留检查点。
测试
有两种启动模式可用于覆盖整个生命周期：langgraph dev用于本地迭代，langgraph up则用于类似生产环境的运行。
一个有效的测试方法是给某个内容命名，然后询问后续轮次是否还能记起它——以此确认对话上下文是否已被正确保存。需要研究性质的问题应转交给研究人员并让Tavily进行处理。而诸如用C++实现快速排序之类的编程请求则应交给程序员处理。当这三种处理路径都能正常运作时，该图结构就为后续添加更多智能体、工具或技能以进一步扩展做好了准备。
总结
LangGraph的灵活性是多智能体设计中最值得注意的特点。它并不强制要求固定的智能体数量或单一的拓扑结构。随着需求的变化，图结构可以从一个节点开始，逐步发展出专门的智能体、条件路由、并行分支、循环结构、人工干预的暂停环节，或是嵌套的子图。