在2026年，单纯调用大模型API已成过去式。真正的趋势是多智能体协作（Multi-Agent）。本文将带你使用目前生产环境最稳定、最强大的框架 LangGraph，从零构建一个能自主搜索、分析并撰写深度报告的“超级研究员”Agent系统。文末附完整代码与核心流程图。

---

为什么选择 LangGraph？

在2026年的众多Agent框架（如AutoGen, CrewAI, Microsoft AutoGen）中，LangGraph 凭借其独特的**有状态图（Stateful Graph）**架构脱颖而出，成为构建复杂生产级应用的首选。

• 可控性极强：不同于黑盒式的自动对话，LangGraph 允许你精确定义每一步的流转逻辑（循环、条件分支、人工介入）。
• 持久化状态：天然支持长流程任务的记忆保持，任务中断后可恢复。
• 人类反馈回路（Human-in-the-loop）：在关键决策点暂停，等待人类确认，这是企业级应用的安全基石。

一句话总结：如果你只是玩票，选AutoGen；如果你要上线生产环境，LangGraph 是唯一真神。

---

核心架构图：超级研究员的工作流

我们的“超级研究员”包含三个核心角色：

1. 搜索员 (Searcher)：负责联网检索最新信息。
2. 分析师 (Analyst)：负责筛选信息、去重、提炼观点。
3. 作家 (Writer)：负责整合内容，撰写最终报告。

它们通过一个循环图协作，直到信息足够丰富或达到最大迭代次数。

---

实战代码：构建你的第一个 LangGraph Agent

1. 环境准备

确保你安装了最新的 langgraph 和 langchain 生态库：

pip install langgraph langchain-openai langchain-community tavily-python

2. 定义状态与节点

我们将使用 TypedDict 来定义整个工作流的共享状态，这是 LangGraph 的核心。

import os
from typing import Annotated, TypedDict, List
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_community.tools.tavily_search import TavilySearchResults

# 配置 API Key (请在环境变量中设置)
os.environ["OPENAI_API_KEY"] = "your-openai-key"
os.environ["TAVILY_API_KEY"] = "your-tavily-key" # 推荐使用 Tavily 作为搜索工具

# 1. 定义状态结构
class AgentState(TypedDict):
    messages: Annotated[List[str], add_messages] # 消息历史
    query: str                                   # 用户查询
    research_data: List[str]                     # 收集到的数据
    iterations: int                              # 当前迭代次数

# 初始化大模型
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 初始化工具
search_tool = TavilySearchResults(max_results=3)

# 2. 定义节点函数

def search_node(state: AgentState):
    """搜索员：根据查询获取最新信息"""
    query = state["query"]
    print(f"🔍 正在搜索: {query}")
    results = search_tool.invoke(query)
    # 提取片段内容
    data = [r['content'] for r in results]
    return {"research_data": data, "iterations": state["iterations"] + 1}

def analyze_node(state: AgentState):
    """分析师：评估信息是否充足，决定下一步"""
    current_data = state["research_data"]
    query = state["query"]
    
    # 简单的启发式判断：实际项目中这里会调用 LLM 进行判断
    if len(current_data) < 6 and state["iterations"] < 3:
        print("🧠 分析师：信息不足，需要再次搜索...")
        return {"messages": ["Need more info"]} # 触发再次搜索逻辑需在图中定义边
    else:
        print("🧠 分析师：信息充足，可以撰写报告。")
        return {"messages": ["Info sufficient"]}

def write_node(state: AgentState):
    """作家：整合所有数据生成最终报告"""
    context = "\n".join(state["research_data"])
    prompt = f"基于以下调研数据，为'{state['query']}'撰写一份深度分析报告:\n{context}"
    
    response = llm.invoke(prompt)
    return {"messages": [f"Final Report:\n{response.content}"]}

# 3. 构建图谱 (Graph)

workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("searcher", search_node)
workflow.add_node("analyst", analyze_node)
workflow.add_node("writer", write_node)

# 设置入口点
workflow.set_entry_point("searcher")

# 定义边的逻辑 (条件路由)
def should_continue(state: AgentState):
    # 如果迭代次数超过3次 或 数据量足够，则去写作，否则继续搜索
    # 注意：这里为了演示简化了逻辑，实际可让 analyst 节点返回具体路由指令
    if state["iterations"] >= 2 or len(state["research_data"]) >= 6:
        return "writer"
    else:
        return "searcher"

#  searcher -> analyst -> (condition) -> searcher/writer
# 为了简化演示，我们构建一个 搜索->分析->(判断)->搜索/写作 的循环
# 在实际 LangGraph 中，通常由分析师节点直接返回下一个节点的名称，或者使用条件边

# 重新设计更清晰的流：
# Entry -> Search -> Analyze (决定去向)
# 这里我们使用条件边来实现循环

workflow.add_edge("searcher", "analyst")

# 条件边：从 analyst 决定去哪里
workflow.add_conditional_edges(
    "analyst",
    lambda state: "writer" if (state["iterations"] >= 2 or len(state["research_data"]) >= 6) else "searcher"
)

workflow.add_edge("writer", END)

# 编译应用
app = workflow.compile()

3. 运行你的 Agent

def run_agent():
    initial_state = {
        "messages": [],
        "query": "2026年量子计算在药物研发领域的最新突破",
        "research_data": [],
        "iterations": 0
    }
    
    print("🚀 超级研究员启动...\n")
    final_state = app.invoke(initial_state)
    
    print("\n✅ 最终报告摘要:")
    print(final_state["messages"][-1])

if __name__ == "__main__":
    run_agent()

---

代码解析：为什么这样写？

1. 状态即真理 (AgentState)：
   所有节点都读写同一个状态对象。这意味着 searcher 抓取的数据，writer 可以直接用，无需复杂的参数传递。Annotated[..., add_messages] 自动处理消息列表的合并，避免覆盖历史。

2. 条件边 (add_conditional_edges)：
   这是实现“智能循环”的关键。传统的脚本是线性的，而 Agent 需要根据中间结果决定是“继续干活”还是“下班交稿”。上面的 lambda 函数就是大脑，动态决定流向。

3. 模块化节点：
   每个函数（search_node, write_node）都是独立的。你可以轻松替换搜索工具（比如从 Tavily 换成 Google Custom Search），或者更换写入模型，而不影响整体架构。

---

进阶：如何加入“人类反馈”？

在生产环境中，你可能不希望 AI 直接发布报告。LangGraph 支持在任意节点后插入断点。

# 编译时开启检查点
app = workflow.compile(checkpointer=MemorySaver())

# 运行到 'analyst' 节点后暂停
config = {"configurable": {"thread_id": "123"}}
for event in app.stream(initial_state, config, stream_mode="values"):
    # 检查是否停在了 analyst 之后
    if "analyst" in event: 
        print("⏸️ 暂停：等待人类确认是否继续搜索...")
        user_input = input("继续搜索? (yes/no): ")
        if user_input.lower() == 'no':
            # 强制修改状态，跳过搜索直接去写作
            # 实际应用中可通过 update_state 实现
            pass 

---

结语

2026年的AI开发，不再是比拼谁的Prompt写得长，而是比拼谁能设计出更稳健的协作流程。

• LangGraph 让你像画流程图一样编写代码。
• 多智能体 让专业的人（模型）做专业的事。
• 人类反馈 确保了系统的可控与安全。

现在，复制上面的代码，运行属于你的第一个“超级研究员”，开启你的Agent工程化之旅吧！