LangGraph 初见

使用 LangGraph 和 LangSmith 构建高级 AI 代理：一个多智能体研究系统实战

摘要

本文档详细阐述了如何利用 LangChain 生态中的 LangGraph 和 LangSmith 构建一个高级、有状态、可循环的 AI 代理系统。传统的线性链式（Chains）调用在处理需要规划、执行、反思和条件判断的复杂任务时能力有限。为了解决此问题，我们通过一个多智能体研究系统的案例，展示如何使用 LangGraph 将代理的行为建模为显式的状态图（State Graph），并利用 LangSmith 对其进行端到端的追踪、调试和评估，实现生产级的可靠性与可观测性。

1. 核心概念

在进入实战编码前，理解以下两个核心组件至关重要。

1.1 LangGraph：可控的 AI 工作流

LangGraph 是一个用于构建有状态、多智能体应用的库。它将传统代理（Agent）中隐式的、有时混沌的思维循环，转变为显式的、可控的状态机。其核心思想是将应用构建为一个图（Graph）。

• State (状态)：一个在图中所有节点间传递和更新的中心化对象。通常定义为一个 Python TypedDict 或 Pydantic BaseModel，作为系统唯一的“事实来源”（Single Source of Truth）。

• Nodes (节点)：图中的工作单元。每个节点是一个 Python 函数，接收当前状态，执行任务（如调用 LLM、执行工具），并返回对状态的更新。

• Edges (边)：定义节点之间的流转逻辑。最关键的是 Conditional Edges (条件边)，它允许我们根据当前状态，动态地决定下一个要执行的节点，从而实现循环、分支等复杂逻辑。

1.2 LangSmith：AI 应用的可观测性

LangSmith 是一个专为 LLM 应用打造的可观测性（Observability）平台，功能类似于传统后端的 DataDog 或 Jaeger。它解决了 LLM 应用调试困难、行为难以预测的痛点。

• Tracing (链路追踪)：自动捕获并可视化应用执行的每一步，包括 LLM 调用、工具使用、提示（Prompt）详情、Token 消耗等。

• Monitoring (监控)：提供对应用在生产环境中的延迟、成本、错误率和用户反馈的宏观监控。

• Evaluation (评估)：允许开发者创建数据集，对应用的不同版本进行自动化评估，以量化模型或提示变更带来的效果。

2. 系统架构与设计

我们构建的研究系统将模拟一个研究团队的工作流程，该流程被建模为一个状态图。

2.1 工作流程

1. Planner (规划师)：接收用户提出的宏观研究课题。

2. Executor (执行者)：使用工具（如搜索引擎）对具体子问题进行研究。

3. Decider (决策者)：判断研究是否完成。如果未完成，流程循环回到 Executor；如果已完成，流程进入下一步。

4. Synthesizer (综合者)：汇总所有研究资料，生成最终报告。

2.2 状态图 (Mermaid)

代码段

graph TD
    A[Planner] --> B[Executor];
    B --> C{Should Continue?};
    C -- Yes --> B;
    C -- No --> D[Synthesizer];
    D --> E[END];

3. Step-by-Step 实现

以下是完整的代码实现步骤。

步骤 1: 环境设置

安装所有必要的依赖库，并配置环境变量。

Python

# 安装依赖库
!pip install -qU langchain langgraph langchain_openai tavily-python langsmith

# 导入并设置环境变量
import os
# os.environ["OPENAI_API_KEY"] = "sk-..."
# os.environ["LANGSMITH_API_KEY"] = "ls__..."
# os.environ["TAVILY_API_KEY"] = "tvly-..."
# os.environ["OPENAI_API_BASE"] = "YOUR_API_BASE_URL" # 如有需要

# 开启 LangSmith 追踪
os.environ["LANGCHAIN_TRACING_V2"] = "true"
# 为项目命名
os.environ["LANGCHAIN_PROJECT"] = "Multi-Agent Research System"

步骤 2: 定义状态与工具

定义图的中心化 State 对象和代理需要使用的 Tool。

Python

from typing import TypedDict, List
from langchain_community.tools.tavily_search import TavilySearchResults

# 定义将在整个图中流转的状态对象
class ResearchState(TypedDict):
    """
    表示研究任务状态的 TypedDict。
    """
    topic: str
    plan: List[str]
    researched_questions: List[str]
    research_data: List[dict]
    report: str

# 实例化工具。Tavily 是一个为 AI 优化的搜索引擎。
# max_results 控制每次搜索返回的结果数量。
tavily_tool = TavilySearchResults(max_results=3)
tools = [tavily_tool]

步骤 3: 实现图节点

为系统中的每个角色（Planner, Executor, Synthesizer）创建对应的节点函数。

Python

import re
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 统一定义 LLM 实例，便于管理和复用
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 节点 1: Planner
planner_prompt = ChatPromptTemplate.from_template(
    """你是一个专业的研究规划师。
你的任务是根据给定的研究主题，生成一个由3到5个具体、可搜索的子问题组成的列表，这个列表将作为后续研究的计划。
请确保问题是独立的，并且能够通过网络搜索找到答案。

研究主题: {topic}

请以无序列表（使用-或*）的格式返回你的计划。"""
)

def planner_node(state: ResearchState) -> dict:
    """接收用户主题，生成研究计划。"""
    print("--- 节点: 规划师 ---")
    topic = state["topic"]
    planner_chain = planner_prompt | llm | StrOutputParser()
    plan_string = planner_chain.invoke({"topic": topic})
    plan = [item.strip() for item in re.findall(r'[-*]\s*(.*)', plan_string)]
    print(f"研究计划生成完毕: {plan}")
    return {"plan": plan}

# 节点 2: Executor
def executor_node(state: ResearchState) -> dict:
    """执行研究计划中的下一个问题。"""
    print("--- 节点: 执行者 ---")
    plan = state["plan"]
    researched_questions = state["researched_questions"]
    question_to_research = plan[len(researched_questions)]

    print(f"正在研究问题: {question_to_research}")
    research_result = tavily_tool.invoke({"query": question_to_research})

    newly_researched = researched_questions + [question_to_research]
    new_data = state["research_data"] + [{"question": question_to_research, "answer": research_result}]
    print("研究完成，数据已更新。")
    return {"researched_questions": newly_researched, "research_data": new_data}

# 节点 3: Synthesizer
synthesizer_prompt = ChatPromptTemplate.from_template(
    """你是一位专业的研究报告撰写员。
你的任务是根据下面的研究主题和收集到的数据，撰写一份全面、流畅、结构清晰的研究报告。
报告应该很好地综合所有信息，而不仅仅是罗列数据。

研究主题: {topic}
研究数据:\n---\n{research_data}\n---

请生成最终的研究报告。"""
)

def synthesizer_node(state: ResearchState) -> dict:
    """综合所有研究数据，生成最终报告。"""
    print("--- 节点: 综合者 ---")
    topic = state["topic"]
    research_data_str = "\n\n".join(
        [f"问题: {item['question']}\n答案: {item['answer']}" for item in state["research_data"]]
    )
    synthesizer_chain = synthesizer_prompt | llm | StrOutputParser()
    report = synthesizer_chain.invoke({"topic": topic, "research_data": research_data_str})
    print("报告生成完毕。")
    return {"report": report}

步骤 4: 组装与编译图

将节点连接起来，定义工作流的逻辑，特别是实现循环的条件边。

Python

from langgraph.graph import StateGraph, END

# 定义决策函数
def should_continue(state: ResearchState) -> str:
    """决策节点：判断是否需要继续研究。"""
    print("--- 节点: 决策 ---")
    if len(state["plan"]) == len(state["researched_questions"]):
        print("决策: 所有问题研究完毕，准备生成报告。")
        return "end"
    else:
        print("决策: 计划尚未完成，继续研究。")
        return "continue"

# 实例化图并绑定 State 对象
graph = StateGraph(ResearchState)

# 添加节点
graph.add_node("planner", planner_node)
graph.add_node("executor", executor_node)
graph.add_node("synthesizer", synthesizer_node)

# 设置入口点
graph.set_entry_point("planner")

# 添加边
graph.add_edge("planner", "executor")
graph.add_conditional_edges(
    "executor",
    should_continue,
    {
        "continue": "executor",
        "end": "synthesizer"
    }
)
graph.add_edge("synthesizer", END)

# 编译图，生成可执行的应用
app = graph.compile()

步骤 5: 运行与观测

使用 app.stream() 运行应用，并展示如何在 LangSmith 中进行分析。

5.1 运行代码

Python

# 定义研究主题和初始状态
topic = "分析一下2025年AI芯片市场的主要趋势、关键参与者和面临的挑战"
initial_state = {
    "topic": topic,
    "plan": [], "researched_questions": [], "research_data": [], "report": ""
}

# 流式运行并打印日志
final_state = {}
print(f"🚀 开始研究主题: {topic}\n")
for chunk in app.stream(initial_state, {"recursion_limit": 10}):
    node_name = list(chunk.keys())[0]
    print(f"--- 节点 '{node_name}' 已执行 ---")
    final_state = chunk[node_name]

print("\n\n✅ 研究流程结束！")
print("\n\n--- 最终研究报告 ---")
print(final_state.get("report", "报告生成失败。"))

5.2 运行日志示例

Plaintext

--- 节点: 规划师 ---
研究计划生成完毕: [...]
--- 节点 'planner' 已执行 ---
--- 节点: 执行者 ---
正在研究问题: ...
--- 节点: 决策 ---
决策: 计划尚未完成，继续研究。
--- 节点 'executor' 已执行 ---
... (循环执行) ...
--- 节点: 决策 ---
决策: 所有问题研究完毕，准备生成报告。
--- 节点 'executor' 已执行 ---
--- 节点: 综合者 ---
报告生成完毕。
--- 节点 'synthesizer' 已执行 ---
✅ 研究流程结束！

5.3 在 LangSmith 中分析

在 LangSmith 平台对应的项目中，可以找到本次运行的完整 Trace。通过 Trace 视图，可以获得以下关键洞见：

• 可视化流程: 直观地看到图的执行路径，包括循环和分支，与设计的状态图完全对应。

• 输入/输出详情: 点击任一 LLM 节点（如 planner），可以审查其完整的输入 Prompt、模型参数、原始输出、Token 消耗和延迟。

• 工具调用分析: 展开 executor 节点，可以清晰地看到其内部对 TavilySearchResults 工具的调用，包括查询的子问题和工具返回的原始数据。

• 逻辑决策透明化: should_continue 决策步骤清晰可见，使得复杂的控制流逻辑不再是黑盒。

4. 结论

通过本案例，我们成功构建并演示了一个基于 LangGraph 的高级 AI 代理。其核心优势在于将代理的内部工作流显式化、结构化，通过状态图的方式极大地增强了系统的可控性、可预测性和可调试性。结合 LangSmith 平台提供的强大可观测能力，开发者可以构建、调试并维护真正达到生产级别的复杂 AI 应用。

5. 未来探索方向

• 错误处理: 为图的节点增加异常捕获和重试逻辑。

• 工具集扩展: 引入更多工具（如代码解释器、数据库查询工具），并让代理动态选择使用。

• 自我修正循环: 增加一个“审稿”（Critique）节点，对 Synthesizer 生成的报告进行评估和反馈，形成一个内部的自我优化循环。