运行代理

代理支持使用.invoke() / await .invoke()进行同步或异步执行以获取完整响应，或者使用.stream() / .astream()进行**增量**流式处理输出。本节解释了如何提供输入、解释输出、启用流式处理以及控制执行限制。

基本用法

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

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

同步调用异步调用

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"}]})

输入和输出

代理使用一种语言模型，该模型期望输入为一个包含消息的列表。因此，代理的输入和输出被存储在代理state的messages键下的messages列表中。

输入格式

代理输入必须是一个包含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文档。

使用自定义代理状态

您可以直接在输入字典中提供代理状态模式中定义的额外字段。这允许基于运行时数据或先前工具输出的动态行为。 详见上下文指南。

注意

当messages参数为字符串时，会被转换为一个HumanMessage。这种行为与create_react_agent中的prompt参数不同，后者作为字符串时被解释为一个SystemMessage。

输出格式

代理输出是一个包含以下内容的字典：

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

更多关于使用自定义状态模式以及访问上下文的信息，请参阅上下文指南。

流式输出

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

• 每一步后的**进度更新**
• 生成中的**LLM标记**
• 执行期间的**自定义工具消息**

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

同步流式异步流式

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "旧金山的天气如何"}]},
    stream_mode="updates"
):
    print(chunk)

async for chunk in agent.astream(
    {"messages": [{"role": "user", "content": "旧金山的天气如何"}]},
    stream_mode="updates"
):
    print(chunk)

提示

有关完整细节，请参阅流式指南。

最大迭代次数

为了控制代理执行并避免无限循环，可以设置递归限制。这定义了代理在引发 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": "旧金山的天气如何"}]},
        # 高亮下一行
        {"recursion_limit": recursion_limit},
    )
except GraphRecursionError:
    print("由于最大迭代次数，代理已停止。")

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": "旧金山的天气如何"}]},
    )
except GraphRecursionError:
    print("由于最大迭代次数，代理已停止。")

额外资源

• LangChain中的异步编程

Comments