运行智能体
你可以通过 Runner 类来运行智能体。共有 3 种方式:
Runner.run():异步运行并返回一个RunResult。Runner.run_sync():同步方法,内部调用.run()。Runner.run_streamed():异步运行并返回一个RunResultStreaming。它以流式传输模式调用 LLM,并在接收时将这些事件流式传输给你。
from agents import Agent, Runner
async def main():
agent = Agent(name="Assistant", instructions="You are a helpful assistant")
result = await Runner.run(agent, "Write a haiku about recursion in programming.")
print(result.final_output)
# Code within the code,
# Functions calling themselves,
# Infinite loop's dance
在结果指南中了解更多。
智能体循环
当你在 Runner 中使用 run 方法时,需要传入一个起始智能体和输入。输入可以是一个字符串(视为用户消息),也可以是一个输入项列表,这些输入项即 OpenAI Responses API 中的条目。
runner 会运行如下循环:
- 我们使用当前输入为当前智能体调用 LLM。
- LLM 生成输出。
- 如果 LLM 返回
final_output,循环结束并返回结果。 - 如果 LLM 进行任务转移,我们会更新当前智能体和输入,并重新运行循环。
- 如果 LLM 产生工具调用,我们会运行这些工具调用,追加结果并重新运行循环。
- 如果 LLM 返回
- 如果超过传入的
max_turns,我们会抛出MaxTurnsExceeded异常。
Note
判断 LLM 输出是否为“最终输出”的规则是:它生成了具有期望类型的文本输出,且没有工具调用。
流式传输
流式传输允许你在 LLM 运行时同时接收流式事件。流结束后,RunResultStreaming 将包含关于本次运行的完整信息,包括所有新生成的输出。你可以调用 .stream_events() 获取流式事件。更多内容参见流式传输指南。
运行配置
run_config 参数可用于配置本次智能体运行的一些全局设置:
model:允许设置一个全局的 LLM 模型使用,而不受各个 Agent 的model配置影响。model_provider:用于查找模型名称的模型提供方,默认是 OpenAI。model_settings:覆盖智能体级别的设置。例如,你可以设置全局的temperature或top_p。input_guardrails,output_guardrails:要在所有运行中包含的输入或输出安全防护措施列表。handoff_input_filter:一个全局输入过滤器,若某次任务转移未定义,则应用于所有任务转移。输入过滤器允许你编辑发送给新智能体的输入。详见Handoff.input_filter。nest_handoff_history:当为True(默认)时,runner 会在调用下一个智能体前,将先前的对话记录折叠为单个助手消息。帮助器将内容放入一个<CONVERSATION HISTORY>块中,并在后续任务转移发生时不断追加新轮次。如果你希望传递原始对话记录,请将其设为False,或提供自定义的 handoff 过滤器。所有Runner方法 在你未显式传入时会自动创建一个RunConfig,因此快速上手与 code examples 会自动采用该默认值,且任何显式的Handoff.input_filter回调仍会覆盖它。单次任务转移可通过Handoff.nest_handoff_history覆盖该设置。handoff_history_mapper:可选的可调用对象。当nest_handoff_history为True时,它会接收规范化后的对话记录(history + handoff items)。它必须返回要转发给下一个智能体的精确输入项列表,使你无需编写完整的 handoff 过滤器即可替换内置摘要。tracing_disabled:允许为整个运行禁用追踪。trace_include_sensitive_data:配置追踪中是否包含潜在的敏感数据,例如 LLM 与工具调用的输入/输出。workflow_name,trace_id,group_id:为此次运行设置追踪的工作流名称、追踪 ID 和追踪分组 ID。我们建议至少设置workflow_name。分组 ID 是可选字段,允许你关联多次运行之间的追踪。trace_metadata:要包含在所有追踪上的元数据。
默认情况下,当一个智能体将任务转移给另一个智能体时,SDK 会将先前的轮次嵌套到单个助手摘要消息中。这减少了重复的助手消息,并将完整的对话记录保存在一个单独的块中,便于新智能体快速扫描。如果你希望恢复旧行为,可以传入 RunConfig(nest_handoff_history=False),或者提供一个 handoff_input_filter(或 handoff_history_mapper)以按需原样转发对话。你也可以为特定的任务转移选择退出(或启用),通过设置 handoff(..., nest_handoff_history=False) 或 True。若想在不编写自定义映射器的情况下更改生成摘要所用的包装文本,请调用 set_conversation_history_wrappers(以及用于恢复默认值的 reset_conversation_history_wrappers)。
会话/聊天线程
调用任一运行方法,可能会导致一个或多个智能体运行(因此可能会有一次或多次 LLM 调用),但其代表一次聊天会话中的单个逻辑轮次。例如:
- 用户轮次:用户输入文本
- Runner 运行:第一个智能体调用 LLM、运行工具、将任务转移给第二个智能体,第二个智能体运行更多工具,然后生成输出。
在智能体运行结束时,你可以选择向用户展示的内容。例如,你可以展示智能体生成的每一个新条目,或仅展示最终输出。无论哪种方式,用户都可能继续追问,此时你可以再次调用 run 方法。
手动会话管理
你可以使用 RunResultBase.to_input_list() 方法手动管理会话历史,以获取下一轮的输入:
async def main():
agent = Agent(name="Assistant", instructions="Reply very concisely.")
thread_id = "thread_123" # Example thread ID
with trace(workflow_name="Conversation", group_id=thread_id):
# First turn
result = await Runner.run(agent, "What city is the Golden Gate Bridge in?")
print(result.final_output)
# San Francisco
# Second turn
new_input = result.to_input_list() + [{"role": "user", "content": "What state is it in?"}]
result = await Runner.run(agent, new_input)
print(result.final_output)
# California
使用 Sessions 的自动会话管理
若想更简化,你可以使用 Sessions 自动处理会话历史,而无需手动调用 .to_input_list():
from agents import Agent, Runner, SQLiteSession
async def main():
agent = Agent(name="Assistant", instructions="Reply very concisely.")
# Create session instance
session = SQLiteSession("conversation_123")
thread_id = "thread_123" # Example thread ID
with trace(workflow_name="Conversation", group_id=thread_id):
# First turn
result = await Runner.run(agent, "What city is the Golden Gate Bridge in?", session=session)
print(result.final_output)
# San Francisco
# Second turn - agent automatically remembers previous context
result = await Runner.run(agent, "What state is it in?", session=session)
print(result.final_output)
# California
Sessions 会自动:
- 在每次运行前检索会话历史
- 在每次运行后存储新消息
- 为不同的 session ID 维护独立会话
更多详情参见 Sessions 文档。
由服务管理的会话
你也可以让 OpenAI 的对话状态功能在服务端管理会话状态,而不是通过 to_input_list() 或 Sessions 在本地处理。这样可以在无需手动重发全部历史消息的情况下保留会话历史。更多内容参见 OpenAI Conversation state 指南。
OpenAI 提供两种跨轮次跟踪状态的方法:
1. 使用 conversation_id
你首先通过 OpenAI Conversations API 创建一个会话,然后在后续每次调用中复用其 ID:
from agents import Agent, Runner
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def main():
agent = Agent(name="Assistant", instructions="Reply very concisely.")
# Create a server-managed conversation
conversation = await client.conversations.create()
conv_id = conversation.id
while True:
user_input = input("You: ")
result = await Runner.run(agent, user_input, conversation_id=conv_id)
print(f"Assistant: {result.final_output}")
2. 使用 previous_response_id
另一种方式是响应链(response chaining),每一轮显式链接到上一轮的响应 ID。
from agents import Agent, Runner
async def main():
agent = Agent(name="Assistant", instructions="Reply very concisely.")
previous_response_id = None
while True:
user_input = input("You: ")
# Setting auto_previous_response_id=True enables response chaining automatically
# for the first turn, even when there's no actual previous response ID yet.
result = await Runner.run(
agent,
user_input,
previous_response_id=previous_response_id,
auto_previous_response_id=True,
)
previous_response_id = result.last_response_id
print(f"Assistant: {result.final_output}")
长时间运行的智能体与人类参与
你可以使用 Agents SDK 的 Temporal 集成来运行持久、长时间运行的工作流,包括 human-in-the-loop 任务。可在此视频中观看 Temporal 与 Agents SDK 协同完成长任务的演示,并在此查看文档。
异常
SDK 在某些情况下会抛出异常。完整列表见 agents.exceptions。概览如下:
AgentsException:SDK 内抛出的所有异常的基类。它是所有其他特定异常的通用父类型。MaxTurnsExceeded:当智能体的运行超过传给Runner.run、Runner.run_sync或Runner.run_streamed方法的max_turns限制时抛出。表示智能体未能在指定的交互轮次数内完成任务。ModelBehaviorError:当底层模型(LLM)产生意外或无效输出时发生。这可能包括:- 格式错误的 JSON:当模型为工具调用或其直接输出提供格式错误的 JSON 结构,尤其是在指定了特定
output_type时。 - 与工具相关的意外失败:当模型未能以预期方式使用工具
- 格式错误的 JSON:当模型为工具调用或其直接输出提供格式错误的 JSON 结构,尤其是在指定了特定
UserError:当你(使用 SDK 编写代码的人)在使用 SDK 时发生错误时抛出。通常由不正确的代码实现、无效的配置或对 SDK API 的误用导致。InputGuardrailTripwireTriggered,OutputGuardrailTripwireTriggered:当输入或输出安全防护措施的触发条件分别被满足时抛出。输入安全防护措施在处理之前检查传入消息,而输出安全防护措施在交付之前检查智能体的最终响应。