LangGraph 系列 8：AgentState 状态详解与自定义状态

langchain langgraph agentstate tool
langgraph
发布日期:   2026-04-19
文章字数:   4.4k
阅读时长:   18 分
阅读次数:  

上一篇介绍了智能体的 Runtime Context。Context 适合保存用户 ID、用户名和权限等调用开始时已经确定的数据，在一次运行中通常按只读方式使用。

但是 Agent 运行起来以后，还会不断产生新的内容。例如用户消息、模型生成的工具调用、工具执行结果，以及工具从外部系统得到的数据。这些数据会随着执行过程变化，因此不能全部放在静态 Context 中。

LangChain 使用 AgentState 管理这类运行时状态。本篇从默认消息状态开始，再为 State 增加自定义字段，并让两个 Tool 通过 State 传递用户名。

1. 为什么 Agent 需要 State

普通模型调用通常只有一次输入和一次输出：

用户问题 -> 模型回答

具有工具调用能力的 Agent 会经历多个步骤：

用户问题
-> 模型决定调用工具
-> 工具执行
-> 模型读取工具结果
-> 生成最终回答

后面的步骤需要知道前面发生了什么。如果工具执行阶段看不到模型刚刚生成的工具调用，便不知道应该执行哪个 Tool；如果第二次模型调用看不到 ToolMessage，也无法根据工具结果整理答案。

因此，Agent 需要一个贯穿本次执行过程的状态对象。它的主要职责有两个：

  1. 保存 Agent 执行过程中产生的消息。
  2. 保存应用自己定义的可变字段。

AgentState 是状态的结构，不是数据库。State 中有数据，不代表这些数据已经持久化。

2. 默认 State 如何保存消息

一次完整的工具调用通常会生成四条消息：

  1. 用户输入形成 HumanMessage。
  2. 模型生成包含 tool_calls 的 AIMessage。
  3. 工具执行结果形成 ToolMessage。
  4. 模型根据工具结果生成最终 AIMessage。

下图中的每个 State 卡片都是 stream_mode=”values” 返回的完整快照，不是某个节点单独返回的局部更新。后一个快照保留了前面的消息，并通过 add_messages 合并本步骤产生的新消息。

AgentState 消息与完整状态快照

这些消息保存在 AgentState.messages 中。Agent 每执行完一个节点，都会提交一部分状态更新；messages 字段通过 add_messages 合并新消息，不需要业务代码每次重新构造完整消息列表。

2.1 项目锁定版本中的 AgentState

本文使用 langchain==1.2.13。这个版本应从 langchain.agents 导入 AgentState：

from langchain.agents import AgentState

当前定义的主要字段可以简化理解为：

class AgentState(TypedDict):
    messages: list[AnyMessage]
    jump_to: NotRequired[str | None]
    structured_response: NotRequired[object]
  • messages 是最重要的字段，保存对话和工具调用消息。
  • jump_to 是框架内部控制执行跳转使用的字段，普通 Agent 代码通常不需要操作。
  • structured_response 在 Agent 使用结构化输出时保存解析后的结果。

课程代码中使用过下面的导入路径：

from langgraph.prebuilt.chat_agent_executor import AgentState

这个兼容入口已经弃用，而且旧定义中还包含 remaining_steps。因此不能把旧源码中的字段直接写成 langchain==1.2.13 的固定字段。本文统一使用新的导入路径，并以项目实际安装版本为准。

3. 观察默认 AgentState 的变化过程

示例代码放在：

llm_learning/langgraph/p08_agent_state/

本文继续使用 .venv_langgraph，不需要安装新的数据库或向量组件：

cd source/_posts/llm_learning
source .venv_langgraph/bin/activate
python -m pip install -r langgraph/p08_agent_state/requirements.txt

使用 LangGraph 系列 6 已经创建的模型环境启动 Qwen3：

source .venv_tool_server/bin/activate

"$VIRTUAL_ENV/bin/python" -m mlx_lm server \
  --model model \
  --host 127.0.0.1 \
  --port 18080 \
  --prompt-cache-size 0 \
  --chat-template-args '{"enable_thinking": false}'

先确认模型接口正常：

curl --noproxy '*' http://127.0.0.1:18080/v1/models

3.1 完整代码

01_inspect_default_agent_state.py：

"""观察默认 AgentState 中的消息如何随 Agent 执行过程逐步增加。"""

import httpx
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_core.messages import AIMessage, ToolMessage
from langchain_openai import ChatOpenAI


@tool
def calculator(a: int, b: int) -> int:
    """计算两个整数的加法。"""

    return a + b


model = ChatOpenAI(
    model="default_model",
    base_url="http://127.0.0.1:18080/v1",
    api_key="not-needed",
    temperature=0,
    max_tokens=128,
    # 禁止本机请求读取系统代理，避免 VPN 影响 127.0.0.1。
    http_client=httpx.Client(trust_env=False),
)


agent = create_agent(
    model=model,
    tools=[calculator],
    system_prompt="你是一个计算助手。遇到加法问题时必须调用 calculator 工具。",
)


# values 模式会在每个执行步骤后返回一份完整状态快照。
states = agent.stream(
    {"messages": [{"role": "user", "content": "请计算 12 加 8。"}]},
    stream_mode="values",
)

for step, state in enumerate(states, start=1):
    messages = state["messages"]
    latest_message = messages[-1]

    print(f"步骤 {step}：消息数量={len(messages)}，最新类型={type(latest_message).__name__}")

    if isinstance(latest_message, AIMessage) and latest_message.tool_calls:
        tool_call = latest_message.tool_calls[0]
        print(f"  工具调用：{tool_call['name']}，参数={tool_call['args']}")
    elif isinstance(latest_message, ToolMessage):
        print(f"  工具结果：{latest_message.content}")
    elif isinstance(latest_message, AIMessage):
        print(f"  最终回答：{latest_message.content}")

运行：

python langgraph/p08_agent_state/01_inspect_default_agent_state.py

真实输出如下：

步骤 1：消息数量=1，最新类型=HumanMessage
步骤 2：消息数量=2，最新类型=AIMessage
  工具调用：calculator，参数={'a': 12, 'b': 8}
步骤 3：消息数量=3，最新类型=ToolMessage
  工具结果：20
步骤 4：消息数量=4，最新类型=AIMessage
  最终回答：12 加 8 的结果是 20。

stream_mode=”values” 返回每个步骤执行后的完整 State，所以可以直接看到 messages 从 1 条增加到 4 条。这不是四次独立调用，而是同一次 Agent 运行的四个状态快照。

4. 自定义 AgentState

默认 State 已经能够保存消息。如果还要保存用户名、订单号、当前任务阶段或工具生成的中间数据，就需要扩展 AgentState。

本文定义一个 CustomState：

class CustomState(AgentState):
    user_name: NotRequired[str]

这里使用 NotRequired，因为 Agent 刚开始运行时还没有把用户名写入 State。用户名会在第一个 Tool 执行后出现。

创建 Agent 时，需要通过 state_schema 注册它：

agent = create_agent(
    model=model,
    tools=[load_user_name, greet_user],
    context_schema=UserContext,
    state_schema=CustomState,
)

如果只定义 CustomState，却没有传给 state_schema，Agent 不会按这份自定义 Schema 管理字段。

5. Context 与 State 如何配合

上一篇已经区分了静态 Context 与可变 State。本例把两者连接起来：

  1. 调用方通过 Context 传入用户名。
  2. load_user_name 从 runtime.context 读取用户名。
  3. Tool 返回 Command(update=…)，把用户名写入 State。
  4. greet_user 从 runtime.state 读取用户名。
  5. Qwen3 根据工具结果生成最终回答。

图中保留了两次 Tool 调用前后的模型决策。Command(update=…) 只把字段和 ToolMessage 提交给 State，并不会直接执行 greet_user；Qwen3 读取第一条工具结果后，才决定发起第二次工具调用。

Runtime Context、Command 与自定义 AgentState 更新流程

这里特意不让第二个 Tool 再次读取 Context。这样可以明确验证：第一个 Tool 写入的 State，确实能够被后续执行步骤读取。

6. 使用 Command 更新 State

普通 Tool 返回字符串或字典时，返回值会转换成工具结果交给模型，但不会自动写入任意自定义 State 字段。

需要修改 State 时，Tool 应返回 Command：

return Command(
    update={
        "user_name": user_name,
        "messages": [
            ToolMessage(
                content=f"已经把用户名 {user_name} 写入 AgentState。",
                tool_call_id=runtime.tool_call_id,
            )
        ],
    }
)

这段代码同时更新两个字段：

  • user_name 保存应用需要的自定义状态。
  • messages 保存模型需要看到的工具执行结果。

6.1 为什么还要返回 ToolMessage

模型生成工具调用时，会产生一个工具调用 ID。工具执行后的 ToolMessage 必须使用相同 ID，模型才能知道这条结果对应哪一次调用。

项目锁定版本可以直接通过 runtime.tool_call_id 取得该值，不需要再单独声明 InjectedToolCallId 参数。

如果 Tool 返回 Command 却不补充对应的 ToolMessage，消息历史可能出现只有工具请求、没有工具结果的无效结构，后续模型调用可能失败。

7. Tool 如何读取 State

当前代码使用 ToolRuntime：

@tool
def greet_user(runtime: ToolRuntime[UserContext, CustomState]) -> str:
    user_name = runtime.state.get("user_name")
    return f"祝你学习顺利，{user_name}！"

ToolRuntime 由框架自动注入，不会出现在模型可见的 Tool Schema 中。它同时提供 state、context、config 和 tool_call_id 等运行信息。

旧代码也可能使用 InjectedState：

from typing import Annotated
from langgraph.prebuilt import InjectedState


@tool
def greet_user(state: Annotated[CustomState, InjectedState]) -> str:
    return f"祝你学习顺利，{state['user_name']}！"

这种写法仍然可以用于只注入 State 的场景。本文使用 ToolRuntime，因为同一个 Tool 还需要访问 Context 和工具调用 ID，入口更加统一。

8. 完整的自定义 State 示例

02_custom_state_and_command.py：

"""使用自定义 AgentState、ToolRuntime 和 Command 在工具之间传递用户名。"""

from typing_extensions import NotRequired, TypedDict

import httpx
from langchain.agents import AgentState, create_agent
from langchain.tools import ToolRuntime, tool
from langchain_core.messages import AIMessage, ToolMessage
from langchain_core.utils.function_calling import convert_to_openai_tool
from langchain_openai import ChatOpenAI
from langgraph.types import Command


class UserContext(TypedDict):
    """定义调用开始时传入的静态用户上下文。"""

    user_name: str


class CustomState(AgentState):
    """在默认消息状态之外增加运行期间可变的用户名。"""

    user_name: NotRequired[str]


@tool
def load_user_name(runtime: ToolRuntime[UserContext, CustomState]) -> Command:
    """先从运行时上下文加载用户名，并把用户名写入 AgentState。"""

    if runtime.context is None or "user_name" not in runtime.context:
        raise ValueError("必须通过 context 传入 user_name")

    user_name = runtime.context["user_name"]

    # Tool 返回 Command 时，同时更新自定义字段和对应的工具结果消息。
    return Command(
        update={
            "user_name": user_name,
            "messages": [
                ToolMessage(
                    content=f"已经把用户名 {user_name} 写入 AgentState。",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
        }
    )


@tool
def greet_user(runtime: ToolRuntime[UserContext, CustomState]) -> str:
    """在用户名已经写入 AgentState 后，为当前用户生成祝福语。"""

    # 第二个工具读取的是可变 State，不再读取静态 Context。
    user_name = runtime.state.get("user_name")
    if not user_name:
        return "AgentState 中还没有用户名，请先调用 load_user_name 工具。"

    return f"祝你学习顺利，{user_name}！"


model = ChatOpenAI(
    model="default_model",
    base_url="http://127.0.0.1:18080/v1",
    api_key="not-needed",
    temperature=0,
    max_tokens=256,
    http_client=httpx.Client(trust_env=False),
)


agent = create_agent(
    model=model,
    tools=[load_user_name, greet_user],
    system_prompt=(
        "你是一个祝福助手。"
        "必须先调用 load_user_name 把用户名写入状态，"
        "再调用 greet_user 生成祝福，最后整理工具结果。"
    ),
    context_schema=UserContext,
    state_schema=CustomState,
)


def run_for_user(user_name: str) -> None:
    """使用指定用户名独立运行一次 Agent。"""

    result = agent.invoke(
        {"messages": [{"role": "user", "content": "请给当前用户一句祝福语。"}]},
        context={"user_name": user_name},
    )

    tool_names = []
    for message in result["messages"]:
        if isinstance(message, AIMessage) and message.tool_calls:
            tool_names.extend(call["name"] for call in message.tool_calls)
        elif isinstance(message, ToolMessage):
            print(f"工具结果：{message.content}")

    print(f"当前调用：{user_name}")
    print(f"工具调用顺序：{' -> '.join(tool_names)}")
    print(f"最终 State 用户名：{result.get('user_name')}")
    print(f"最终回答：{result['messages'][-1].content}\n")


# runtime 由框架注入，因此两个工具都没有模型可见参数。
for current_tool in [load_user_name, greet_user]:
    tool_schema = convert_to_openai_tool(current_tool)
    properties = tool_schema["function"]["parameters"]["properties"]
    print(f"{current_tool.name} 的模型可见参数：{properties}")

print()
run_for_user("小明")
run_for_user("小红")

运行：

python langgraph/p08_agent_state/02_custom_state_and_command.py

真实输出如下：

load_user_name 的模型可见参数：{}
greet_user 的模型可见参数：{}

工具结果：已经把用户名 小明 写入 AgentState。
工具结果：祝你学习顺利，小明！
当前调用：小明
工具调用顺序：load_user_name -> greet_user
最终 State 用户名：小明
最终回答：祝你学习顺利，小明！

工具结果：已经把用户名 小红 写入 AgentState。
工具结果：祝你学习顺利，小红！
当前调用：小红
工具调用顺序：load_user_name -> greet_user
最终 State 用户名：小红
最终回答：祝你学习顺利，小红！

这次运行验证了三件事：

  1. Qwen3 按顺序调用了两个 Tool。
  2. 第二个 Tool 能读取第一个 Tool 写入的 State。
  3. 小明和小红的两次 invoke() 没有互相覆盖。

工具调用 ID 是动态值，示例没有打印它，但 runtime.tool_call_id 已经用于创建匹配的 ToolMessage。

9. messages 为什么不会被直接覆盖

普通状态字段如果没有 Reducer，新值通常会覆盖旧值。但 AgentState.messages 已经使用 add_messages 作为内置 Reducer。

它至少解决三个问题：

  1. 新消息拥有新 ID 时追加到列表。
  2. 新消息与已有消息 ID 相同时替换原消息。
  3. 收到 RemoveMessage 时删除对应 ID 的消息。

这里只介绍默认 messages 字段必须理解的行为。自定义 Reducer、并行节点冲突和工作流状态合并留到后续 Workflow 章节。

9.1 完整代码

03_message_update_and_remove.py：

"""演示 AgentState.messages 使用的 add_messages 如何追加、替换和删除消息。"""

from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
from langgraph.graph.message import add_messages


def print_messages(title: str, messages: list) -> None:
    """打印当前消息 ID、类型和内容。"""

    print(title)
    for message in messages:
        print(f"  {message.id} | {type(message).__name__} | {message.content}")


# 新 ID 会把消息追加到列表末尾。
messages = add_messages(
    [],
    [HumanMessage(content="推荐一种适合周末的活动。", id="user-1")],
)
messages = add_messages(
    messages,
    [AIMessage(content="可以去公园散步。", id="ai-1")],
)
print_messages("1. 追加两条消息", messages)


# 相同 ID 不会重复追加，而是替换原来的消息。
messages = add_messages(
    messages,
    [AIMessage(content="可以去公园散步，也可以参观博物馆。", id="ai-1")],
)
print_messages("\n2. 使用相同 ID 更新 AI 消息", messages)


# 删除消息需要提交 RemoveMessage，而不是使用不存在的 Command(delete=...)。
messages = add_messages(messages, [RemoveMessage(id="user-1")])
print_messages("\n3. 删除指定的用户消息", messages)

运行：

python langgraph/p08_agent_state/03_message_update_and_remove.py

真实输出如下：

1. 追加两条消息
  user-1 | HumanMessage | 推荐一种适合周末的活动。
  ai-1 | AIMessage | 可以去公园散步。

2. 使用相同 ID 更新 AI 消息
  user-1 | HumanMessage | 推荐一种适合周末的活动。
  ai-1 | AIMessage | 可以去公园散步，也可以参观博物馆。

3. 删除指定的用户消息
  ai-1 | AIMessage | 可以去公园散步，也可以参观博物馆。

需要注意，当前 Command 构造函数没有 delete 参数。删除消息不是把 update 改成 delete，而是向 messages 提交一个带目标 ID 的 RemoveMessage，再由 add_messages 完成删除。

10. AgentState 与记忆存储的边界

State 描述 Agent 当前拥有哪些数据，以及节点更新数据时应该怎样合并。它解决的是运行过程中的数据传递问题。

但是下面这段代码连续执行两次时：

agent.invoke(input, context={"user_name": "小明"})
agent.invoke(input, context={"user_name": "小红"})

如果没有配置 Checkpointer，两次调用会创建相互独立的 State。第二次调用不会自动得到第一次调用的消息或 user_name。

后续记忆章节会继续介绍：

  • Checkpointer 如何按 Thread 保存短期状态。
  • 开发环境和 PostgreSQL 如何持久化 Checkpoint。
  • Store 如何保存跨 Thread 使用的长期数据。

这些能力建立在 State 之上，但不属于 AgentState Schema 本身。

11. 常见问题

11.1 使用了旧的 AgentState 导入路径

当前代码应使用：

from langchain.agents import AgentState

不要继续把 langgraph.prebuilt.chat_agent_executor.AgentState 作为新代码入口。

11.2 定义了字段却没有注册 state_schema

自定义 State 必须传给 create_agent()：

agent = create_agent(
    model=model,
    tools=tools,
    state_schema=CustomState,
)

11.3 普通返回值没有更新自定义 State

下面的返回值只是工具结果：

return {"user_name": "小明"}

需要更新 AgentState 时，应返回：

return Command(update={"user_name": "小明"})

如果这是模型发起的 Tool Call，还需要同时提供匹配调用 ID 的 ToolMessage。

11.4 Tool 调用顺序不符合预期

Agent 会自己决定调用哪个 Tool。Tool 名称、Docstring 和系统提示词必须清楚描述前置条件。

本例的 greet_user 还检查了 user_name 是否存在。即使模型提前调用它，也会收到“先调用 load_user_name”的明确结果，而不是出现未处理异常。

11.5 把 Context 和 State 混在一起

  • 调用开始前已经确定、运行期间不应该修改的数据放在 Context。
  • Agent 执行过程中生成或变化的数据放在 State。
  • thread_id、标签、回调和递归限制等执行参数放在 Config。

11.6 认为 State 会自动跨调用保存

没有 Checkpointer 时，State 只服务于当前运行。需要重启恢复或多轮记忆时，必须显式配置持久化组件。

11.7 本地请求受到代理影响

先检查模型服务：

curl --noproxy '*' http://127.0.0.1:18080/v1/models

示例中的 httpx.Client(trust_env=False) 用于避免系统代理或 VPN 转发本机请求。

12. 小结

默认 AgentState 已经负责管理 Agent 的消息历史。用户消息、工具调用、工具结果和最终回答都会依次进入 messages，add_messages 负责合并这些更新。

业务需要额外的运行状态时，可以继承 AgentState 增加字段，再通过 state_schema 注册。Tool 使用 ToolRuntime.state 读取状态，使用 Command(update=…) 修改状态。

本次实测中，Qwen3 依次调用两个 Tool，将 Context 中的用户名写入 CustomState，再从 State 读取用户名生成祝福。小明和小红两次调用的状态完全隔离。

对象或机制 保存内容 读取方式 更新方式 是否自动持久化
Runtime Context 用户身份、租户和权限等静态业务数据 runtime.context 由调用方在运行开始时传入 否
AgentState.messages 用户、模型和工具消息 runtime.state[“messages”] 节点更新或 Command(update=…)，由 add_messages 合并 否
自定义 State 字段 Agent 执行过程中产生的动态数据 runtime.state[“字段名”] Command(update={…}) 否
ToolMessage 某次工具调用的执行结果 state[“messages”] 使用匹配的 tool_call_id 写入 messages 否
RemoveMessage 指定要删除的消息 ID 由 add_messages 处理 作为 messages 更新提交 否
Checkpointer 同一 Thread 的 State Checkpoint 后续记忆章节介绍 Graph 执行时保存 是，需要显式配置
文章作者: hnbian
文章链接: https://www.hnbian.cn/posts/da8621bc.html
版权声明: 本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 hnbian !
langchain langgraph agentstate tool
评论
  目录