从0到1搭建Multi-Agent决策系统：LangGraph完整指南

核心概念

什么是LangGraph？

在深入了解LangGraph之前，让我们先理解它所解决的核心问题。在传统的LangChain应用中，我们通常构建的是线性的、有向无环图(DAG)结构的应用。虽然这种结构对于许多应用场景已经足够，但当我们需要构建更复杂的、具有状态管理、循环和多代理协作的系统时，传统的DAG结构就显得力不从心了。

LangGraph 正是为了解决这个问题而诞生的。它是LangChain生态系统中的一个组件，专门用于构建具有状态和循环的多代理系统。与传统的LangChain不同，LangGraph允许我们构建包含循环、条件分支和持久状态的应用，这使得它非常适合构建复杂的决策系统和多代理协作系统。

什么是Multi-Agent系统？

Multi-Agent系统（多代理系统）是一种由多个相互作用的智能代理组成的计算系统。这些代理可以是自主的、半自主的或受控的，它们共同工作以解决单个代理难以或无法解决的问题。

在LLM（大语言模型）的背景下，Multi-Agent系统通常指的是由多个LLM代理组成的系统，每个代理可能有不同的专业知识、角色或目标，它们通过协作来完成复杂的任务。

LangGraph的核心概念

在开始使用LangGraph之前，我们需要理解它的几个核心概念：

1. State（状态）：LangGraph中的状态是一个字典，它在整个图的执行过程中被传递和更新。每个节点可以读取和修改状态。
2. Nodes（节点）：节点是图中的计算单元，通常是一个函数，它接收状态作为输入，并返回更新后的状态。
3. Edges（边）：边连接节点，定义了图的执行流程。边可以是无条件的（总是从一个节点流向另一个节点）或有条件的（根据状态决定下一步流向哪个节点）。
4. Conditional Edges（条件边）：这是LangGraph的一个强大特性，它允许我们根据当前状态动态决定下一步应该执行哪个节点。
5. Entry Point（入口点）：图执行的起始节点。
6. End Point（结束点）：图执行的结束节点，当到达这个节点时，图的执行就会停止。

问题背景

传统LLM应用的局限性

在LangGraph出现之前，我们通常使用LangChain来构建LLM应用。LangChain提供了一种简单的方式来将LLM与其他组件（如提示模板、工具、内存等）组合在一起。然而，传统的LangChain应用有以下几个局限性：

1. 线性执行流程：传统的LangChain应用通常是线性的，这意味着它们按照预定义的顺序执行步骤，无法轻松地处理需要循环或条件分支的复杂任务。
2. 有限的状态管理：虽然LangChain提供了内存组件来管理状态，但这些内存组件通常是简单的键值存储，难以处理复杂的状态更新和传递。
3. 难以构建多代理系统：传统的LangChain应用通常是单代理的，虽然可以通过一些技巧来实现多代理协作，但这种实现方式往往复杂且难以维护。

复杂决策任务的需求

随着LLM应用的发展，我们越来越需要构建能够处理复杂决策任务的系统。这些任务通常具有以下特点：

1. 需要多步骤推理：许多复杂任务需要LLM进行多步骤推理，每一步的结果都会影响下一步的决策。
2. 需要条件分支：根据中间结果，系统可能需要执行不同的操作。
3. 需要循环执行：某些任务可能需要反复执行某个步骤，直到满足特定条件。
4. 需要多代理协作：某些复杂任务可能需要多个具有不同专业知识的代理协作完成。

正是这些需求推动了LangGraph的发展。

问题描述

让我们通过一个具体的例子来更清楚地理解LangGraph解决的问题。假设我们要构建一个研究助手系统，它的任务是帮助用户研究某个主题。这个系统需要能够：

1. 理解用户的查询：首先，系统需要理解用户想要研究什么。
2. 规划研究步骤：根据用户的查询，系统需要规划一系列研究步骤。
3. 执行研究步骤：系统需要执行这些研究步骤，可能包括搜索网络、阅读文档、总结信息等。
4. 评估研究结果：在执行完一些步骤后，系统需要评估是否已经收集到足够的信息，或者是否需要进一步研究。
5. 生成最终报告：最后，系统需要根据收集到的信息生成一份最终的研究报告。

这是一个典型的复杂决策任务，它需要循环执行（评估结果后可能需要继续研究）、条件分支（根据评估结果决定是继续研究还是生成报告）和状态管理（保存研究过程中收集到的信息）。

如果我们尝试用传统的LangChain来构建这个系统，我们会发现它非常困难，甚至不可能。但是，使用LangGraph，我们可以相对容易地构建这样一个系统。

问题解决

现在，让我们看看如何使用LangGraph来解决上述问题。首先，我们需要设计系统的状态、节点和边。

设计状态

状态是LangGraph应用的核心，它需要包含系统在执行过程中需要的所有信息。对于我们的研究助手系统，状态可能包含以下内容：

1. 用户查询：用户最初的查询。
2. 研究计划：系统规划的研究步骤。
3. 研究笔记：系统在研究过程中收集到的信息。
4. 当前步骤：系统当前正在执行的步骤。
5. 评估结果：系统对研究结果的评估。
6. 最终报告：系统生成的最终报告。

设计节点

节点是系统中的计算单元，每个节点负责执行一个特定的任务。对于我们的研究助手系统，我们可能需要以下节点：

1. 理解查询节点：负责理解用户的查询。
2. 制定计划节点：根据用户的查询制定研究计划。
3. 执行研究节点：执行当前的研究步骤。
4. 评估结果节点：评估研究结果，决定下一步操作。
5. 生成报告节点：根据研究结果生成最终报告。

设计边

边定义了系统的执行流程。对于我们的研究助手系统，我们可能需要以下边：

1. 从"理解查询节点"到"制定计划节点"的无条件边。
2. 从"制定计划节点"到"执行研究节点"的无条件边。
3. 从"执行研究节点"到"评估结果节点"的无条件边。
4. 从"评估结果节点"到"执行研究节点"的条件边（如果需要继续研究）。
5. 从"评估结果节点"到"生成报告节点"的条件边（如果已经收集到足够的信息）。
6. 从"生成报告节点"到结束点的无条件边。

通过这样的设计，我们可以构建一个能够处理复杂决策任务的研究助手系统。

边界与外延

LangGraph的边界

虽然LangGraph非常强大，但它也有自己的边界，我们需要了解这些边界以避免不恰当的使用：

1. 不是通用的图计算框架：LangGraph是专门为构建LLM应用设计的，它不是一个通用的图计算框架。如果你需要处理大规模的图数据或进行复杂的图算法，你可能需要使用其他工具。
2. 状态的大小限制：虽然LangGraph没有明确限制状态的大小，但由于状态需要在节点之间传递，过大的状态可能会导致性能问题。
3. 执行的确定性：LangGraph的执行流程是确定的，除非你使用条件边来引入不确定性。这意味着你需要在设计图时仔细考虑所有可能的执行路径。

LangGraph的外延

LangGraph的设计使其可以很容易地与其他工具和技术集成：

1. 与LangChain的集成：LangGraph是LangChain生态系统的一部分，它可以很容易地与LangChain的其他组件（如提示模板、工具、内存等）集成。
2. 与Vector Stores的集成：你可以在LangGraph节点中使用Vector Stores来存储和检索信息。
3. 与外部工具的集成：你可以在LangGraph节点中调用外部工具，如搜索引擎、数据库、API等。
4. 与其他LLM的集成：LangGraph不局限于使用特定的LLM，你可以在节点中使用任何你喜欢的LLM。

概念结构与核心要素组成

LangGraph的核心要素

正如我们之前提到的，LangGraph有几个核心要素：State、Nodes、Edges、Conditional Edges、Entry Point和End Point。让我们更详细地了解这些要素：

State（状态）

状态是一个字典，它在整个图的执行过程中被传递和更新。每个节点可以读取和修改状态。状态的结构由你定义，你可以根据你的应用需求来设计状态的结构。

在LangGraph中，状态有两种类型：

1. 完整状态：这是状态的完整版本，包含所有的信息。
2. 增量状态：这是状态的增量版本，只包含从上一个节点到当前节点的变化。

默认情况下，LangGraph使用完整状态，但你可以通过配置来使用增量状态。

Nodes（节点）

节点是图中的计算单元，通常是一个函数，它接收状态作为输入，并返回更新后的状态。节点可以执行任何你想要的操作，如调用LLM、使用工具、更新状态等。

在LangGraph中，你可以将任何Python函数注册为节点，只要它符合以下签名：

def node_function(state: Dict[str, Any]) -> Dict[str, Any]:
    # 执行一些操作
    # 更新状态
    return updated_state

Edges（边）

边连接节点，定义了图的执行流程。边可以是无条件的（总是从一个节点流向另一个节点）或有条件的（根据状态决定下一步流向哪个节点）。

在LangGraph中，你可以使用以下方法来添加边：

1. add_edge(start_node: str, end_node: str)：添加一条从start_node到end_node的无条件边。
2. add_conditional_edges(source_node: str, condition: Callable, edge_map: Dict[str, str])：添加一条从source_node出发的条件边，condition是一个函数，它接收状态作为输入，并返回一个字符串，edge_map是一个字典，它将condition的返回值映射到目标节点。

Entry Point（入口点）

入口点是图执行的起始节点。你需要使用set_entry_point(node: str)方法来设置入口点。

End Point（结束点）

结束点是图执行的结束节点，当到达这个节点时，图的执行就会停止。你需要使用set_finish_point(node: str)方法来设置结束点。

LangGraph的概念结构

LangGraph的概念结构可以用以下方式来表示：

+----------------+       +----------------+       +----------------+
|                |       |                |       |                |
|  Entry Point   +------>+     Node       +------>+  End Point     |
|                |       |                |       |                |
+----------------+       +----------------+       +----------------+
                                ^    |
                                |    |
                           +----+----v----+
                           |              |
                           |  Conditional |
                           |     Edge     |
                           |              |
                           +--------------+

在这个结构中，图的执行从入口点开始，然后根据边的定义在节点之间流动，最后到达结束点。

概念之间的关系：概念核心属性维度对比

为了更清楚地理解LangGraph的核心概念之间的关系，让我们通过一个表格来对比它们的核心属性：

概念	定义	核心功能	是否必需	可变性	示例
State	在图执行过程中传递和更新的数据字典	存储和传递信息	是	可变	{“query”: “研究人工智能”, “notes”: [“…”]}
Node	图中的计算单元	执行操作，更新状态	是	固定（除非重新定义图）	理解查询节点，执行研究节点
Edge	连接节点，定义执行流程	控制执行流程	是	固定（除非重新定义图）	从理解查询节点到制定计划节点的边
Conditional Edge	根据状态决定执行流程的边	动态控制执行流程	否	固定（除非重新定义图），但执行结果是可变的	从评估结果节点到执行研究节点或生成报告节点的边
Entry Point	图执行的起始节点	定义图的开始	是	固定（除非重新定义图）	理解查询节点
End Point	图执行的结束节点	定义图的结束	是	固定（除非重新定义图）	生成报告节点

概念联系的ER实体关系与交互关系图

ER实体关系图

让我们用Mermaid来绘制LangGraph核心概念的ER实体关系图：

这个ER图展示了LangGraph核心概念之间的关系：

• State被Node读取和修改
• Node可以是Edge的起点或终点
• Node可以是Conditional Edge的起点
• Entry Point是一个特殊的Node
• End Point是一个特殊的Node
• Edge连接Nodes
• Conditional Edge连接到Nodes
• Conditional Edge依赖State

交互关系图

现在，让我们用Mermaid来绘制LangGraph核心概念的交互关系图：

这个交互关系图展示了LangGraph执行过程中核心概念之间的交互：

1. 用户向图输入初始状态
2. 图从入口点开始执行
3. 入口点读取初始状态，然后将其传递给第一个节点
4. 第一个节点修改状态，然后将更新后的状态传递给条件边
5. 条件边读取状态，根据条件决定下一步流向哪个节点
6. 被选择的节点修改状态，然后将更新后的状态传递给结束点
7. 结束点将最终状态返回给图
8. 图将结果返回给用户

数学模型

虽然LangGraph主要是一个工程框架，但我们可以用一些数学概念来形式化地描述它。

状态转换系统

LangGraph本质上是一个状态转换系统（State Transition System）。状态转换系统是一个数学模型，它描述了系统如何从一个状态转换到另一个状态。

一个状态转换系统可以被定义为一个四元组 $(S,S_0,A,T)$，其中：

• $S$ 是状态的集合
• $S_0\subseteq S$ 是初始状态的集合
• $A$ 是动作的集合
• $T\subseteq S\times A\times S$ 是转换关系

在LangGraph的语境中：

• $S$ 是所有可能的状态字典的集合
• $S_0$ 是用户输入的初始状态的集合
• $A$ 是节点函数的集合
• $T$ 是由边定义的转换关系

图论模型

我们也可以用图论（Graph Theory）来模型化LangGraph。一个图可以被定义为一个有序对 $G=(V,E)$，其中：

• $V$ 是顶点的集合（在LangGraph中，顶点就是节点）
• $E\subseteq V\times V$ 是边的集合

在LangGraph的语境中，我们的图通常是一个有向图（Directed Graph），因为边有方向。此外，我们的图可能包含循环（Cycles），这是LangGraph相比传统LangChain的一个重要优势。

我们可以进一步将LangGraph的图定义为一个标注图（Labeled Graph）或加权图（Weighted Graph），其中边带有标签或权重，这些标签或权重可以表示条件边的条件。

执行语义

最后，我们可以用操作语义（Operational Semantics）来形式化地描述LangGraph的执行过程。操作语义是一种描述程序行为的方法，它通过定义程序状态之间的转换规则来描述程序的执行。

对于LangGraph，我们可以定义以下转换规则：

1. 节点执行规则：如果系统处于状态 $s$，并且当前节点是 $n$，那么系统将执行节点 $n$ 的函数 $f_n$，并转换到状态 $f_n(s)$。

$$\frac{\text{current node}=n\text{state}=s}{\text{state}\Rightarrow f_n(s)}$$

2. 无条件边转换规则：如果系统刚刚执行完节点 $n_1$，并且存在一条从 $n_1$ 到 $n_2$ 的无条件边，那么当前节点将变为 $n_2$。

$$\frac{\text{last node}=n_1(n_1,n_2)\in E_{\text{unconditional}}}{\text{current node}\Rightarrow n_2}$$

3. 条件边转换规则：如果系统刚刚执行完节点 $n_1$，并且存在一条从 $n_1$ 出发的条件边，条件函数为 $c$，并且 $c(s)=l$，并且边映射将 $l$ 映射到 $n_2$，那么当前节点将变为 $n_2$。

$$\frac{\text{last node}=n_1(n_1,c,M)\in E_{\text{conditional}}c(s)=lM(l)=n_2}{\text{current node}\Rightarrow n_2}$$

4. 终止规则：如果当前节点是结束节点，那么系统将停止执行。

$$\frac{\text{current node}=n_{\text{end}}}{\text{execution stops}}$$

这些规则形式化地描述了LangGraph的执行过程。

算法流程图

现在，让我们用Mermaid来绘制LangGraph的执行算法流程图：

这个流程图展示了LangGraph的执行算法：

1. 首先，初始化状态为用户输入
2. 然后，设置当前节点为入口点
3. 检查当前节点是否是结束点：
   • 如果是，返回最终状态并结束
   • 如果不是，执行当前节点的函数，更新状态
4. 检查是否有条件边从当前节点出发：
   • 如果是，评估条件函数，获取结果标签，然后根据边映射获取目标节点
   • 如果不是，获取无条件边的目标节点
5. 设置当前节点为目标节点
6. 回到步骤3，重复执行

算法源代码

现在，让我们来看一下LangGraph的核心算法的简化Python实现。请注意，这不是LangGraph的实际源代码，而是一个简化版本，用于帮助你理解LangGraph的工作原理。

from typing import Dict, Any, Callable, Optional

class StateGraph:
    def __init__(self):
        self.nodes: Dict[str, Callable] = {}
        self.edges: Dict[str, str] = {}
        self.conditional_edges: Dict[str, tuple] = {}
        self.entry_point: Optional[str] = None
        self.end_point: Optional[str] = None
    
    def add_node(self, name: str, func: Callable):
        """添加一个节点"""
        self.nodes[name] = func
    
    def add_edge(self, start: str, end: str):
        """添加一条无条件边"""
        self.edges[start] = end
    
    def add_conditional_edge(self, source: str, condition: Callable, edge_map: Dict[str, str]):
        """添加一条条件边"""
        self.conditional_edges[source] = (condition, edge_map)
    
    def set_entry_point(self, node: str):
        """设置入口点"""
        self.entry_point = node
    
    def set_end_point(self, node: str):
        """设置结束点"""
        self.end_point = node
    
    def run(self, initial_state: Dict[str, Any]) -> Dict[str, Any]:
        """运行图"""
        if self.entry_point is None:
            raise ValueError("Entry point not set")
        if self.end_point is None:
            raise ValueError("End point not set")
        
        state = initial_state.copy()
        current_node = self.entry_point
        
        while True:
            # 检查是否到达结束点
            if current_node == self.end_point:
                break
            
            # 执行当前节点
            if current_node not in self.nodes:
                raise ValueError(f"Node {current_node} not found")
            
            node_func = self.nodes[current_node]
            state = node_func(state)
            
            # 确定下一个节点
            if current_node in self.conditional_edges:
                # 条件边
                condition, edge_map = self.conditional_edges[current_node]
                result = condition(state)
                if result not in edge_map:
                    raise ValueError(f"Condition result {result} not found in edge map")
                current_node = edge_map[result]
            elif current_node in self.edges:
                # 无条件边
                current_node = self.edges[current_node]
            else:
                raise ValueError(f"No edge found from node {current_node}")
        
        # 执行结束点
        if self.end_point in self.nodes:
            node_func = self.nodes[self.end_point]
            state = node_func(state)
        
        return state

这个简化的StateGraph类实现了LangGraph的核心功能：

• 你可以添加节点、无条件边和条件边
• 你可以设置入口点和结束点
• 你可以运行图，它会按照你定义的流程执行

让我们来看一下如何使用这个简化的StateGraph类：

# 创建一个简单的研究助手系统

# 定义节点函数
def understand_query(state):
    print("理解查询...")
    state["understood_query"] = f"研究主题: {state['query']}"
    return state

def make_plan(state):
    print("制定研究计划...")
    state["plan"] = ["步骤1: 搜索相关信息", "步骤2: 整理信息", "步骤3: 撰写报告"]
    state["current_step"] = 0
    return state

def execute_research(state):
    print(f"执行研究步骤 {state['current_step'] + 1}...")
    if "notes" not in state:
        state["notes"] = []
    state["notes"].append(f"步骤 {state['current_step'] + 1} 的结果")
    state["current_step"] += 1
    return state

def evaluate_result(state):
    print("评估研究结果...")
    if state["current_step"] >= len(state["plan"]):
        state["evaluation"] = "完成"
    else:
        state["evaluation"] = "继续"
    return state

def generate_report(state):
    print("生成最终报告...")
    state["report"] = f"关于 {state['query']} 的研究报告:\n\n"
    state["report"] += "研究结果:\n"
    for note in state["notes"]:
        state["report"] += f"- {note}\n"
    return state

def do_nothing(state):
    return state

# 创建图
graph = StateGraph()

# 添加节点
graph.add_node("understand_query", understand_query)
graph.add_node("make_plan", make_plan)
graph.add_node("execute_research", execute_research)
graph.add_node("evaluate_result", evaluate_result)
graph.add_node("generate_report", generate_report)
graph.add_node("end", do_nothing)

# 添加边
graph.add_edge("understand_query", "make_plan")
graph.add_edge("make_plan", "execute_research")
graph.add_edge("execute_research", "evaluate_result")
graph.add_edge("generate_report", "end")

# 添加条件边
def should_continue(state):
    return state["evaluation"]

graph.add_conditional_edge(
    "evaluate_result",
    should_continue,
    {
        "继续": "execute_research",
        "完成": "generate_report"
    }
)

# 设置入口点和结束点
graph.set_entry_point("understand_query")
graph.set_end_point("end")

# 运行图
initial_state = {"query": "人工智能"}
final_state = graph.run(initial_state)

print("\n最终报告:")
print(final_state["report"])

这个例子创建了一个简单的研究助手系统，它会：

1. 理解用户的查询
2. 制定研究计划
3. 循环执行研究步骤，直到所有步骤都完成
4. 生成最终报告

实际场景应用

研究助手

正如我们在上面的例子中看到的，LangGraph非常适合构建研究助手系统。这种系统需要能够规划研究步骤、执行研究、评估结果，并在必要时继续研究。

客户服务代理

LangGraph也可以用于构建复杂的客户服务代理系统。这种系统需要能够：

1. 理解客户的问题
2. 尝试解决问题
3. 评估解决方案是否有效
4. 如果无效，尝试其他解决方案
5. 如果所有解决方案都无效，将客户转接到人工客服

软件开发生命周期助手

LangGraph还可以用于构建软件开发生命周期助手系统。这种系统需要能够：

1. 理解需求
2. 生成代码
3. 测试代码
4. 评估代码质量
5. 如果需要，修改代码
6. 最后，部署代码

医疗诊断助手

最后，LangGraph可以用于构建医疗诊断助手系统。这种系统需要能够：

1. 收集患者的症状
2. 提出可能的诊断
3. 建议进一步的检查
4. 根据检查结果 refine 诊断
5. 最后，提出治疗建议

项目介绍

现在，让我们来创建一个真实的项目，使用LangGraph构建一个更复杂的研究助手系统。这个系统将能够：

1. 理解用户的查询
2. 规划研究步骤
3. 使用网络搜索工具收集信息
4. 评估收集到的信息
5. 如果需要，继续收集信息
6. 最后，生成一份详细的研究报告

项目结构

我们的项目将有以下结构：

research_assistant/
├── __init__.py
├── main.py
├── nodes/
│   ├── __init__.py
│   ├── understand_query.py
│   ├── make_plan.py
│   ├── execute_research.py
│   ├── evaluate_result.py
│   └── generate_report.py
├── tools/
│   ├── __init__.py
│   └── search.py
├── utils/
│   ├── __init__.py
│   └── state.py
└── requirements.txt

依赖

我们的项目将依赖以下库：

• langchain
• langgraph
• openai
• python-dotenv
• requests

这些依赖将列在requirements.txt文件中。

环境安装

1. 创建虚拟环境

首先，我们需要创建一个虚拟环境来隔离我们的项目依赖：

python -m venv venv

然后，激活虚拟环境：

在Windows上：

venv\Scripts\activate

在macOS或Linux上：

source venv/bin/activate

2. 安装依赖

接下来，我们需要安装我们的项目依赖。首先，创建一个requirements.txt文件：

langchain>=0.0.300
langgraph>=0.0.10
openai>=1.0.0
python-dotenv>=1.0.0
requests>=2.31.0

然后，安装这些依赖：

pip install -r requirements.txt

3. 设置环境变量

最后，我们需要设置一些环境变量。创建一个.env文件：

OPENAI_API_KEY=你的_openai_api_key
SERPAPI_API_KEY=你的_serpapi_api_key

你需要替换上述值为你自己的API密钥。你可以从OpenAI和SerpAPI的网站上获取这些API密钥。

系统功能设计

核心功能

我们的研究助手系统将有以下核心功能：

1. 查询理解：系统将能够理解用户的查询，提取关键信息。
2. 研究规划：系统将能够根据用户的查询制定一个详细的研究计划。
3. 信息收集：系统将能够使用网络搜索工具收集信息。
4. 结果评估：系统将能够评估收集到的信息，确定是否已经收集到足够的信息。
5. 报告生成：系统将能够根据收集到的信息生成一份详细的研究报告。

辅助功能

除了核心功能外，我们的系统还将有以下辅助功能：

1. 状态管理：系统将能够管理研究过程中的状态。
2. 错误处理：系统将能够处理研究过程中可能出现的错误。
3. 日志记录：系统将能够记录研究过程中的重要事件。

系统架构设计

总体架构

我们的研究助手系统将采用分层架构，分为以下几层：

1. 表示层：负责与用户交互，接收用户的查询，返回研究报告。
2. 业务逻辑层：负责实现研究助手的核心功能，包括查询理解、研究规划、信息收集、结果评估和报告生成。
3. 工具层：提供系统需要的工具，如网络搜索工具。
4. 基础设施层：提供系统需要的基础设施，如状态管理、错误处理和日志记录。

图架构

在LangGraph层面，我们的系统将有以下节点和边：

节点：

1. understand_query：理解用户的查询
2. make_plan：制定研究计划
3. execute_research：执行当前的研究步骤
4. evaluate_result：评估研究结果
5. generate_report：生成最终报告
6. end：结束节点

边：

1. 从understand_query到make_plan的无条件边
2. 从make_plan到execute_research的无条件边
3. 从execute_research到evaluate_result的无条件边
4. 从evaluate_result到execute_research的条件边（如果需要继续研究）
5. 从evaluate_result到generate_report的条件边（如果已经收集到足够的信息）
6. 从generate_report到end的无条件边

系统接口设计

状态接口

我们的系统将使用一个字典来表示状态，状态将包含以下字段：

{
    "query": str,  # 用户的原始查询
    "understood_query": str,  # 系统理解后的查询
    "plan": List[str],  # 研究计划
    "current_step": int,  # 当前执行的步骤索引
    "search_queries": List[str],  # 为当前步骤生成的搜索查询
    "search_results": List[Dict],  # 搜索结果
    "notes": List[str],  # 研究笔记
    "evaluation": str,  # 评估结果（"继续"或"完成"）
    "report": str,  # 最终报告
    "error": Optional[str]  # 错误信息（如果有）
}

节点接口

每个节点将是一个函数，它接收状态作为输入，并返回更新后的状态：

def node_function(state: Dict[str, Any]) -> Dict[str, Any]:
    # 执行一些操作
    # 更新状态
    return updated_state

主接口

我们的系统将提供一个主接口run_research_assistant，它接收用户的查询作为输入，并返回最终的研究报告：

def run_research_assistant(query: str) -> str:
    # 创建图
    # 添加节点和边
    # 运行图
    # 返回最终报告
    return report

系统核心实现源代码

现在，让我们来实现我们的研究助手系统。

1. 工具实现

首先，让我们实现网络搜索工具。创建tools/search.py文件：

import os
import requests
from typing import List, Dict
from dotenv import load_dotenv

load_dotenv()

SERPAPI_API_KEY = os.getenv("SERPAPI_API_KEY")

def search_web(query: str, num_results: int = 5) -> List[Dict]:
    """使用SerpAPI搜索网络"""
    if not SERPAPI_API_KEY:
        raise ValueError("SERPAPI_API_KEY environment variable not set")
    
    url = "https://serpapi.com/search"
    params = {
        "q": query,
        "api_key": SERPAPI_API_KEY,
        "num": num_results
    }
    
    try:
        response = requests.get(url, params=params)
        response.raise_for_status()
        data = response.json()
        
        results = []
        if "organic_results" in data:
            for item in data["organic_results"]:
                results.append({
                    "title": item.get("title", ""),
                    "link": item.get("link", ""),
                    "snippet": item.get("snippet", "")
                })
        
        return results
    except Exception as e:
        print(f"Error searching web: {e}")
        return []

2. 状态工具实现

接下来，让我们实现一些状态工具函数。创建utils/state.py文件：

from typing import Dict, Any, List

def create_initial_state(query: str) -> Dict[str, Any]:
    """创建初始状态"""
    return {
        "query": query,
        "understood_query": "",
        "plan": [],
        "current_step": 0,
        "search_queries": [],
        "search_results": [],
        "notes": [],
        "evaluation": "",
        "report": "",
        "error": None
    }

def add_note(state: Dict[str, Any], note: str) -> Dict[str, Any]:
    """添加笔记到状态"""
    updated_state = state.copy()
    updated_state["notes"].append(note)
    return updated_state

def set_error(state: Dict[str, Any], error: str) -> Dict[str, Any]:
    """设置错误信息到状态"""
    updated_state = state.copy()
    updated_state["error"] = error
    return updated_state

3. 节点实现

现在，让我们实现各个节点。我们将使用OpenAI的GPT-4模型来帮助我们实现这些节点。

首先，创建nodes/understand_query.py文件：

import os
from typing import Dict, Any
from dotenv import load_dotenv
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

llm = OpenAI(
    model_name="gpt-4",
    temperature=0,
    openai_api_key=OPENAI_API_KEY
)

def understand_query(state: Dict[str, Any]) -> Dict[str, Any]:
    """理解用户的查询"""
    updated_state = state.copy()
    
    prompt = PromptTemplate(
        input_variables=["query"],
        template="""
        请分析以下用户查询，提取关键信息，并用简洁明了的语言重述这个查询。
        
        用户查询：{query}
        
        请按以下格式输出：
        理解后的查询：[你对查询的重述]
        """
    )
    
    try:
        response = llm(prompt.format(query=state["query"]))
        # 解析响应
        if "理解后的查询：" in response:
            understood_query = response.split("理解后的查询：")[1].strip()
            updated_state["understood_query"] = understood_query
        else:
            updated_state["understood_query"] = state["query"]
        
        print(f"理解后的查询：{updated_state['understood_query']}")
    except Exception as e:
        print(f"理解查询时出错：{e}")
        updated_state["understood_query"] = state["query"]
    
    return updated_state

接下来，创建nodes/make_plan.py文件：

import os
from typing import Dict, Any
from dotenv import load_dotenv
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

llm = OpenAI(
    model_name="gpt-4",
    temperature=0,
    openai_api_key=OPENAI_API_KEY
)

def make_plan(state: Dict[str, Any]) -> Dict[str, Any]:
    """制定研究计划"""
    updated_state = state.copy()
    
    prompt = PromptTemplate(
        input_variables=["query"],
        template="""
        请为以下研究主题制定一个详细的研究计划。研究计划应该包含3-5个步骤，每个步骤都应该是具体的、可执行的。
        
        研究主题：{query}
        
        请按以下格式输出研究计划：
        步骤1：[第一步的描述]
        步骤2：[第二步的描述]
        ...
        """
    )
    
    try:
        response = llm(prompt.format(query=state["understood_query"]))
        # 解析响应
        plan = []
        lines = response.split("\n")
        for line in lines:
            line = line.strip()
            if line.startswith("步骤"):
                # 移除"步骤X："前缀
                step = line.split("：", 1)[1].strip() if "：" in line else line
                plan.append(step)
        
        updated_state["plan"] = plan
        updated_state["current_step"] = 0
        
        print("研究计划：")
        for i, step in enumerate(plan):
            print(f"  {i+1}. {step}")
    except Exception as e:
        print(f"制定研究计划时出错：{e}")
        # 创建一个默认计划
        updated_state["plan"] = ["搜索相关信息", "整理信息", "撰写报告"]
        updated_state["current_step"] = 0
    
    return updated_state

然后，创建nodes/execute_research.py文件：

import os
from typing import Dict, Any
from dotenv import load_dotenv
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from tools.search import search_web

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

llm = OpenAI(
    model_name="gpt-4",
    temperature=0,
    openai_api_key=OPENAI_API_KEY
)

def execute_research(state: Dict[str, Any]) -> Dict[str, Any]:
    """执行当前的研究步骤"""
    updated_state = state.copy()
    current_step = state["current_step"]
    plan = state["plan"]
    
    if current_step >= len(plan):
        print("所有研究步骤已完成")
        return updated_state
    
    step_description = plan[current_step]
    print(f"执行步骤 {current_step + 1}：{step_description}")
    
    # 生成搜索查询
    search_query_prompt = PromptTemplate(
        input_variables=["query", "step_description"],
        template="""
        请为以下研究步骤生成1-3个搜索查询。搜索查询应该是具体的、有助于完成这个研究步骤的。
        
        研究主题：{query}
        研究步骤：{step_description}
        
        请按以下格式输出搜索查询：
        1. [第一个搜索查询]
        2. [第二个搜索查询]
        3. [第三个搜索查询]
        """
    )
    
    try:
        search_query_response = llm(search_query_prompt.format(
            query=state["understood_query"],
            step_description=step_description
        ))
        
        # 解析搜索查询
        search_queries = []
        lines = search_query_response.split("\n")
        for line in lines:
            line = line.strip()
            if line and (line[0].isdigit() and len(line) > 1 and line[1] in [".", ")"]):
                # 移除数字前缀
                query = line.split(maxsplit=1)[1].strip() if " " in line else line
                search_queries.append(query)
        
        updated_state["search_queries"] = search_queries
        print(f"生成的搜索查询：{search_queries}")
    except Exception as e:
        print(f"生成搜索查询时出错：{e}")
        # 使用默认搜索查询
        updated_state["search_queries"] = [state["understood_query"]]
    
    # 执行搜索
    all_search_results = []
    for search_query in updated_state["search_queries"]:
        print(f"搜索：{search_query}")
        search_results = search_web(search_query)
        all_search_results.extend(search_results)
    
    updated_state["search_results"] = all_search_results
    print(f"找到 {len(all_search_results)} 个搜索结果")
    
    # 总结搜索结果
    summarize_prompt = PromptTemplate(
        input_variables=["query", "step_description", "search_results"],
        template="""
        请根据以下搜索结果，为这个研究步骤总结关键信息。总结应该是简洁的、信息丰富的。
        
        研究主题：{query}
        研究步骤：{step_description}
        
        搜索结果：
        {search_results}
        
        请按以下格式输出总结：
        总结：[你对搜索结果的总结]
        """
    )
    
    # 格式化搜索结果
    formatted_search_results = ""
    for i, result in enumerate(all_search_results):
        formatted_search_results += f"结果 {i+1}:\n"
        formatted_search_results += f"标题：{result['title']}\n"
        formatted_search_results += f"链接：{result['link']}\n"