langGraph state构建

<think>我们正在讨论如何构建LangGraph的状态。根据引用内容，LangGraph用于构建有状态的多智能体应用程序，其核心概念之一是状态图(StateGraph)。状态图需要定义一个状态对象，该对象会在图的各个节点之间传递和更新。 在LangGraph中，构建状态通常涉及以下步骤： 1. 定义状态模式（State Schema）：一个自定义的类，用于描述状态中包含哪些字段。每个字段表示智能体工作流中的一部分数据。 2. 初始化状态图：使用定义的状态模式创建一个StateGraph实例。 3. 添加节点：节点是处理状态的函数，每个节点可以读取和修改状态中的特定字段。 4. 定义边：包括条件边（conditional edges）和普通边，用于控制流程的走向。 下面我们详细说明如何构建状态。 ### 1. 定义状态模式 状态模式是一个类，通常使用`TypedDict`或普通的类（但需要继承自`typing.TypedDict`）来定义。在LangGraph中，状态是一个字典般的对象，其键是字符串，值可以是任意类型，但通常我们会明确指定每个字段的类型。 例如，假设我们正在构建一个简单的聊天机器人，状态可能包含用户输入和聊天历史。我们可以这样定义状态： ```python from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages class State(TypedDict): user_input: str chat_history: Annotated[List[str], add_messages] ``` 注意：这里使用了`Annotated`类型，这是为了在状态更新时使用特定的更新策略。`add_messages`是LangGraph提供的一种更新策略，用于在列表中追加消息（而不是覆盖）。LangGraph支持多种更新策略，如覆盖、追加等。 ### 2. 初始化状态图 创建状态图实例，并传入状态类作为参数： ```python from langgraph.graph import StateGraph graph = StateGraph(State) ``` ### 3. 添加节点 节点是一个函数，它接收当前状态（一个字典，包含状态模式中定义的字段）作为输入，并返回一个包含更新后状态字段的字典（或直接返回更新后的状态对象）。这个返回的字典将用于更新整个状态（根据更新策略）。 例如，我们添加一个节点来处理用户输入： ```python def process_input(state: State) -> dict: # 这里可以处理用户输入，例如调用LLM等 processed = state['user_input'].upper() # 简单示例：将输入转为大写 return {'processed_input': processed} # 注意：这里返回的字段需要与状态模式匹配？但注意，我们定义的状态模式中并没有'processed_input'，所以我们需要调整状态模式或节点行为。 # 调整状态模式：如果我们希望状态中包含处理后的输入，我们需要在状态模式中增加字段。 ``` 因此，我们需要重新考虑状态模式的设计。通常，状态模式应该包含工作流中所有节点可能读写的数据。所以，我们修改状态模式： ```python class State(TypedDict): user_input: str chat_history: Annotated[List[str], add_messages] processed_input: str # 新增字段 ``` 然后调整节点函数： ```python def process_input(state: State) -> dict: processed = state['user_input'].upper() return {'processed_input': processed} # 返回要更新的字段 ``` 然后将节点添加到图中： ```python graph.add_node("process_input", process_input) ``` ### 4. 定义边 我们需要设置起始节点和边，以及条件边（如果需要循环）。例如： ```python graph.set_entry_point("process_input") # 设置入口节点 graph.set_finish_point("process_input") # 如果只有一个节点，可以直接设置为结束点 # 如果有多个节点，我们需要添加边来连接它们，例如： # graph.add_edge('node1', 'node2') ``` ### 5. 编译图 最后，编译图以创建可执行的工作流： ```python app = graph.compile() ``` ### 6. 执行图 使用初始状态调用图： ```python initial_state = State(user_input="hello", chat_history=[], processed_input="") result = app.invoke(initial_state) ``` ### 注意事项 - 状态更新策略：在定义状态字段时，使用`Annotated`可以指定更新策略。例如，`add_messages`表示追加消息，而默认策略是覆盖（如果不用`Annotated`，则直接覆盖）。 - 节点函数返回的字典中的每个键必须与状态模式中的某个字段名对应，并且值类型要匹配。返回的字典中只需要包含需要更新的字段，不需要返回整个状态。 ### 示例：带条件边的循环 假设我们想让智能体在满足条件时循环执行。我们可以添加一个条件边。 首先，定义两个节点：一个处理节点，一个判断是否结束的节点。 ```python def should_continue(state: State) -> str: # 根据状态判断是否继续 if state['processed_input'] == "QUIT": return "end" else: return "continue" def process_node(state: State) -> dict: # 这里模拟处理，每次将processed_input加上一个感叹号，直到遇到QUIT new_processed = state['processed_input'] + "!" return {'processed_input': new_processed} # 添加节点 graph.add_node("process", process_node) graph.add_node("check_end", should_continue) # 注意：这个节点实际上是一个条件函数，但也可以作为节点添加 # 设置入口 graph.set_entry_point("process") # 添加边：从process到check_end graph.add_edge("process", "check_end") # 添加条件边：根据check_end节点的返回值决定下一步 graph.add_conditional_edges( "check_end", should_continue, # 这个函数返回字符串，表示下一个节点或结束 { "continue": "process", # 如果返回"continue"，则回到process节点 "end": END # 结束 } ) ``` 注意：条件边函数（`should_continue`）的返回值是字符串，这些字符串映射到下一个节点的名称或`END`（表示结束）。 ### 总结 构建LangGraph状态的关键步骤： 1. 设计状态模式（TypedDict），包含所有需要传递的数据字段。 2. 创建状态图实例。 3. 添加节点（处理函数），每个节点负责更新状态的一部分。 4. 设置边（包括条件边）以控制流程。 5. 编译并执行。 以上步骤可以帮助你构建LangGraph的状态图。 §§相关问题§§ 1. LangGraph状态模式中如何定义字段的更新策略？ 2. 在LangGraph中，如何实现多个智能体之间的状态共享？ 3. LangGraph的状态图与传统的有限状态机(FSM)有何区别？ 4. 在LangGraph中，如何处理状态更新时的并发冲突？</think>### LangGraph状态构建指南 在LangGraph中构建状态是实现有状态多智能体系统的核心，以下是构建步骤和关键概念： #### 1. **定义状态模式（State Schema）** 状态模式是一个自定义类，用于描述智能体工作流中的数据结构和类型： ```python from typing import TypedDict, Annotated, Sequence from langgraph.graph.message import add_messages class AgentState(TypedDict): user_query: str agent_response: Annotated[str, lambda x, y: y] # 覆盖式更新 chat_history: Annotated[Sequence[str], add_messages] # 追加式更新 tools_used: list ``` **关键特性**： - **字段类型声明**：每个字段定义数据类型（如`str`, `list`等） - **更新策略**（通过`Annotated`指定）： - `add_messages`：追加数据（适合聊天历史） - `lambda x, y: y`：覆盖更新（适合最新结果） - **状态持久化**：状态会在节点间自动传递和更新[^1] #### 2. **初始化状态图** ```python from langgraph.graph import StateGraph # 创建状态图实例 graph = StateGraph(AgentState) # 添加节点（智能体功能单元） graph.add_node("query_analyzer", analyze_query) graph.add_node("tool_executor", execute_tools) graph.add_node("response_generator", generate_response) # 设置节点关系 graph.add_edge("query_analyzer", "tool_executor") graph.add_edge("tool_executor", "response_generator") graph.set_finish_point("response_generator") # 编译可执行图 app = graph.compile() ``` #### 3. **节点函数设计** 每个节点接收当前状态，返回状态更新： ```python def execute_tools(state: AgentState) -> dict: """工具执行节点示例""" tools = select_tools(state["user_query"]) results = [tool.run() for tool in tools] return { "tools_used": tools, # 更新工具列表 "agent_response": results[0] # 更新最新结果 } ``` #### 4. **执行状态工作流** ```python # 初始化状态 initial_state = AgentState( user_query="如何计算GDP？", agent_response="", chat_history=[], tools_used=[] ) # 执行工作流 final_state = app.invoke(initial_state) print(final_state["agent_response"]) ``` #### 5. **高级功能** - **条件分支**：使用`add_conditional_edges`实现动态路由 ```python graph.add_conditional_edges( "query_analyzer", decide_next_node, # 返回"tool_executor"或"response_generator" {"need_tools": "tool_executor", "direct_answer": "response_generator"} ) ``` - **循环控制**：通过`add_edge`创建循环实现多轮交互 - **多智能体协作**：不同节点可对应不同智能体，共享状态对象 #### 最佳实践 1. **状态最小化**：只存储必要数据，避免过度复杂的状态结构 2. **更新策略**：根据字段特性选择追加（历史数据）或覆盖（最新结果） 3. **错误处理**：在节点中添加状态回滚机制 4. **状态快照**：定期持久化状态，支持断点续跑 > 示例应用场景：客户服务系统可包含`问题分类`→`数据库查询`→`结果生成`三个节点，状态对象传递用户问题、查询结果、回复内容等数据[^1]。