前面我们学过 LangGraph 的基本操作——如何添加边、添加节点、组装图，以及可视化。但有一个关键点我之前没仔细讲：节点之间究竟怎么传递消息？今天就来把这个坑填上。

0. 先上代码

代码来源: https://github.com/langchain-ai/langgraph/blob/main/examples/agent_executor/human-in-the-loop.ipynb

from langchain import hub
from langchain.agents import create_openai_functions_agent
from langchain_openai.chat_models import ChatOpenAI
from langchain_community.tools.ta vily_search import Ta vilySearchResults

tools = [Ta vilySearchResults(max_results=1)]

# Get the prompt to use - you can modify this!
prompt = hub.pull("hwchase17/openai-functions-agent")

# Choose the LLM that will drive the agent
llm = ChatOpenAI(model="gpt-3.5-turbo-1106", streaming=True)

# Construct the OpenAI Functions agent
agent_runnable = create_openai_functions_agent(llm, tools, prompt)

from typing import TypedDict, Annotated, List, Union
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
    # The input string
    input: str
    # The list of previous messages in the conversation
    chat_history: list[BaseMessage]
    # The outcome of a given call to the agent
    # Needs `None` as a valid type, since this is what this will start as
    agent_outcome: Union[AgentAction, AgentFinish, None]
    # List of actions and corresponding observations
    # Here we annotate this with `operator.add` to indicate that operations to
    # this state should be ADDED to the existing values (not overwrite it)
    intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]
    
from langchain_core.agents import AgentFinish
from langgraph.prebuilt.tool_executor import ToolExecutor

# This a helper class we ha ve that is useful for running tools
# It takes in an agent action and calls that tool and returns the result
tool_executor = ToolExecutor(tools)

# Define the agent
def run_agent(data):
    agent_outcome = agent_runnable.invoke(data)
    return {"agent_outcome": agent_outcome}

# Define the function to execute tools
def execute_tools(data):
    # Get the most recent agent_outcome - this is the key added in the `agent` above
    agent_action = data["agent_outcome"]
    response = input(f"[y/n] continue with: {agent_action}?")
    if response == "n":
        raise ValueError
    output = tool_executor.invoke(agent_action)
    return {"intermediate_steps": [(agent_action, str(output))]}

# Define logic that will be used to determine which conditional edge to go down
def should_continue(data):
    # If the agent outcome is an AgentFinish, then we return `exit` string
    # This will be used when setting up the graph to define the flow
    if isinstance(data["agent_outcome"], AgentFinish):
        return "end"
    # Otherwise, an AgentAction is returned
    # Here we return `continue` string
    # This will be used when setting up the graph to define the flow
    else:
        return "continue"

from langgraph.graph import END, StateGraph

# Define a new graph
workflow = StateGraph(AgentState)

# Define the two nodes we will cycle between
workflow.add_node("agent", run_agent)
workflow.add_node("action", execute_tools)

# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.set_entry_point("agent")

# We now add a conditional edge
workflow.add_conditional_edges(
    # First, we define the start node. We use `agent`.
    # This means these are the edges taken after the `agent` node is called.
    "agent",
    # Next, we pass in the function that will determine which node is called next.
    should_continue,
    # Finally we pass in a mapping.
    # The keys are strings, and the values are other nodes.
    # END is a special node marking that the graph should finish.
    # What will happen is we will call `should_continue`, and then the output of that
    # will be matched against the keys in this mapping.
    # Based on which one it matches, that node will then be called.
    {
        # If `tools`, then we call the tool node.
        "continue": "action",
        # Otherwise we finish.
        "end": END,
    },
)

# We now add a normal edge from `tools` to `agent`.
# This means that after `tools` is called, `agent` node is called next.
workflow.add_edge("action", "agent")

# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable
app = workflow.compile()

inputs = {"input": "北京今天的天气怎么样？", "chat_history": []}
for s in app.stream(inputs):
    print(list(s.values())[0])
    print("----")

运行结果：

1. 代码解释

1.1 总结使用LangGraph的步骤

上面这段代码其实没什么新鲜的，就是 LangGraph 的标准套路。之前入门文章里已经详细聊过：

，创建一个图：workflow = StateGraph(AgentState)

，往图里添加节点：

workflow.add_node("agent", run_agent)
workflow.add_node("action", execute_tools)

，添加边——既有条件边，也有普通边：

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "action",
        "end": END,
    },
)

workflow.add_edge("action", "agent")

，设置入口节点：workflow.set_entry_point("agent")

，编译图：app = workflow.compile()

，运行。这里用的是 stream 函数，当然也可以用 invoke。

1.2 节点定义

一共两个节点：agent 和 action，分别对应两个函数。

• run_agent 负责调用 Agent 模型，拿到执行结果。
• execute_tools 先让用户确认是否继续执行工具（人工介入），如果同意就执行工具，否则抛出异常终止。

def run_agent(data):
    agent_outcome = agent_runnable.invoke(data)
    return {"agent_outcome": agent_outcome}

def execute_tools(data):
    agent_action = data["agent_outcome"]
    response = input(f"[y/n] continue with: {agent_action}?")
    if response == "n":
        raise ValueError
    output = tool_executor.invoke(agent_action)
    return {"intermediate_steps": [(agent_action, str(output))]}

2. 节点间信息传递

下面进入今天的重头戏：节点之间到底怎么传递信息？Graph 中的 State 又是如何更新的？

前面提到过，LangGraph 的核心概念之一是

。每次执行图时都会创建一个状态对象，这个状态会在节点之间传递，每个节点执行后都会更新它。说白了，LangGraph 本质上就是一套状态机机制。而节点间的消息传递，靠的就是这个状态。

2.1 代码中的状态定义

来看代码里定义的状态：

class AgentState(TypedDict):
    input: str
    chat_history: list[BaseMessage]
    agent_outcome: Union[AgentAction, AgentFinish, None]
    intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]

只需要定义一个继承 TypedDict 的类即可。这里自定义了四个字段：input（用户输入）、chat_history（历史对话）、agent_outcome（Agent 执行后的动作 or 结束标记）、intermediate_steps（中间步骤记录）。每个节点在执行前都可以从状态中读取这些信息，执行后也可以把返回值写回状态。

注意：agent_outcome 就是 Agent 执行后返回的状态（AgentAction 或 AgentFinish），不用纠结具体是什么，把它理解成一个状态标记就行。

实际的使用流程如下：

（1）创建图时绑上状态：workflow = StateGraph(AgentState)

（2）在节点函数里就可以随意读取和更新了。以 execute_tools 为例：

def execute_tools(data):
    agent_action = data["agent_outcome"]          # 从状态中读取
    response = input(f"[y/n] continue with: {agent_action}?")
    if response == "n":
        raise ValueError
    output = tool_executor.invoke(agent_action)   # 执行工具
    return {"intermediate_steps": [(agent_action, str(output))]}  # 更新状态

执行前从状态中拿到 agent_outcome，用来提示用户是否继续。如果继续，就调用工具，然后把结果写回到 intermediate_steps 字段中。下一个节点就能读到这个新数据了。这样一来，节点间的信息就像接力棒一样传下去了。

3. 总结

今天的内容其实不复杂。先回顾了 LangGraph 的基本搭建流程，然后深入了解了状态的定义和节点间消息传递的原理。这部分算是之前入门文章的一个关键补充。总结一下自定义消息传递的操作步骤：

（1）定义 class AgentState(TypedDict)
（2）传递给图：workflow = StateGraph(AgentState)
（3）在节点里读取状态：data = data["agent_outcome"]
（4）在节点里更新状态：return {"intermediate_steps": [(agent_action, str(output))]}

顺便提一句，这个例子其实是为了演示如何在多智能体交互中让人参与进来。做法很简单：在节点里加个 input() 函数，等待用户确认就行。实际业务中可以根据这个思路扩展。