运行代理¶

代理支持使用 .invoke() / await .ainvoke() 来获取完整响应，或使用 .stream() / .astream() 来进行增量流式输出，从而实现同步和异步执行。本节将解释如何提供输入、解读输出、启用流式传输以及控制执行限制。

基本用法¶

代理可以以两种主要模式执行：

同步模式使用 .invoke() 或 .stream()
异步模式使用 await .ainvoke() 或使用 .astream() 与 async for

同步调用异步调用

from langgraph.prebuilt import create_react_agent

agent = create_react_agent(...)

response = agent.invoke({"messages": [{"role": "user", "content": "what is the weather in sf"}]})

from langgraph.prebuilt import create_react_agent

agent = create_react_agent(...)
response = await agent.ainvoke({"messages": [{"role": "user", "content": "what is the weather in sf"}]})

输入和输出¶

代理使用一个语言模型，该模型期望以 messages 列表作为输入。因此，代理的输入和输出都存储在代理状态的 messages 键下的 messages 列表中。

输入格式¶

Agent 的输入必须是一个包含 messages 键的字典。支持的格式如下：

格式	示例
字符串	`{"messages": "Hello"}` — 被解释为 HumanMessage
消息字典	`{"messages": {"role": "user", "content": "Hello"}}`
消息列表	`{"messages": [{"role": "user", "content": "Hello"}]}`
带自定义状态	`{"messages": [{"role": "user", "content": "Hello"}], "user_name": "Alice"}` — 如果使用了自定义的 `state_schema`

消息会自动转换为 LangChain 内部的消息格式。你可以在 LangChain 文档中阅读更多关于 LangChain 消息的内容。

使用自定义 Agent 状态

你可以直接在输入字典中提供你在 Agent 状态模式中定义的额外字段。这允许基于运行时数据或先前工具输出实现动态行为。有关详细信息，请参阅上下文指南。

Note

messages 的字符串输入会被转换为 HumanMessage。这种行为与 create_react_agent 中的 prompt 参数不同，当作为字符串传递时，prompt 会被解释为 SystemMessage。

输出格式¶

Agent 的输出是一个字典，包含：

messages：执行过程中交换的所有消息列表（用户输入、助手回复、工具调用）。
可选的 structured_response，如果配置了结构化输出。
如果使用了自定义的 state_schema，输出中可能还包含与您定义的字段相对应的其他键。这些键可以保存从工具执行或提示逻辑中更新的状态值。

有关如何使用自定义状态模式和访问上下文的更多细节，请参阅上下文指南。

流式输出¶

代理支持流式响应，以实现更响应的应用程序。这包括：

每个步骤后的进度更新
**LLM 令牌**在生成时
**自定义工具消息**在执行期间

流式输出在同步和异步模式中都可用：

同步流式异步流式

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    stream_mode="updates"
):
    print(chunk)

async for chunk in agent.astream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    stream_mode="updates"
):
    print(chunk)

Tip

有关完整详细信息，请参阅流式指南。

最大迭代次数¶

为了控制代理的执行并避免无限循环，请设置一个递归限制。这定义了代理在引发 GraphRecursionError 之前可以采取的最大步骤数。您可以在运行时或通过 .with_config() 定义代理时配置 recursion_limit：

运行时.with_config()

from langgraph.errors import GraphRecursionError
from langgraph.prebuilt import create_react_agent

max_iterations = 3
recursion_limit = 2 * max_iterations + 1
agent = create_react_agent(
    model="anthropic:claude-3-5-haiku-latest",
    tools=[get_weather]
)

try:
    response = agent.invoke(
        {"messages": [{"role": "user", "content": "what's the weather in sf"}]},
        {"recursion_limit": recursion_limit},
    )
except GraphRecursionError:
    print("Agent stopped due to max iterations.")

from langgraph.errors import GraphRecursionError
from langgraph.prebuilt import create_react_agent

max_iterations = 3
recursion_limit = 2 * max_iterations + 1
agent = create_react_agent(
    model="anthropic:claude-3-5-haiku-latest",
    tools=[get_weather]
)
agent_with_recursion_limit = agent.with_config(recursion_limit=recursion_limit)

try:
    response = agent_with_recursion_limit.invoke(
        {"messages": [{"role": "user", "content": "what's the weather in sf"}]},
    )
except GraphRecursionError:
    print("Agent stopped due to max iterations.")

其他资源¶

LangChain 中的异步编程