openai-agents-python REPL 实用工具:用 run_demo_loop 在终端中交互式调试你的 Agent
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
run_demo_loop是 openai-agents-python SDK 内置的 REPL(Read-Eval-Print Loop)调试工具,让你无需编写额外的 Web 界面或测试脚手架,直接在终端里以多轮对话的方式快速验证 Agent 的行为。读完本文,你将掌握run_demo_loop的完整用法、全部参数含义、底层流式事件处理机制,以及如何用它调试工具调用、Handoff 等复杂场景。
一、什么是 run_demo_loop
SDK 在 src/agents/repl.py 中提供了run_demo_loop,官方文档 docs/repl.md 将其定位为「直接在终端中快速、交互式地测试 Agent 行为」的实用工具。它本质上是一个基于asyncio的死循环:不断提示用户输入 → 交给 Runner 执行 → 打印模型输出 → 继续等待下一条输入,直到用户主动退出。
该函数通过 src/agents/init.py 导出为公开 API(from agents import run_demo_loop),并收录在 docs/ref/repl.md 的 API 参考中,属于稳定的公开接口。
二、最小可运行示例
官方文档给出的入门示例(见 docs/repl.md)如下:
import asyncio from agents import Agent, run_demo_loop async def main() -> None: agent = Agent(name="Assistant", instructions="You are a helpful assistant.") await run_demo_loop(agent) if __name__ == "__main__": asyncio.run(main())将上述代码保存为demo.py,在项目目录下运行:
python demo.py注意:运行前需要设置OPENAI_API_KEY环境变量(或配置其他模型提供商),并确保已安装openai-agents及其依赖。之后run_demo_loop会启动一个交互式聊天会话,终端出现>提示符等待你输入。
三、交互行为详解
从 docs/repl.md 的说明和 src/agents/repl.py 的实现来看,这个循环具备以下核心行为:
1. 多轮对话历史自动保留
每一轮用户输入都会被追加到input_items列表中,并在下一轮执行时作为完整的输入历史传给 Runner。具体来说,每轮循环结束时:
current_agent = result.last_agent input_items = result.to_input_list()result.to_input_list()会把当前轮的所有输入项(用户消息、模型响应、工具调用等)序列化为可继续传递的输入列表,从而实现跨轮记忆;result.last_agent用于处理 Handoff 场景:如果 Agent 在对话中把控制权交给了另一个 Agent,下一轮对话会自动从新的 Agent 继续(此时终端会打印[Agent updated: 新Agent名]提示)。
因此,Agent 能记住整个会话期间讨论过的内容,实现真正的多轮上下文。
2. 退出方式
结束聊天会话有三种方式(实现见 src/agents/repl.py):
| 方式 | 说明 |
|---|---|
输入quit并按 Enter | 触发退出(不区分大小写,QUIT同样有效) |
输入exit并按 Enter | 同上 |
按Ctrl-D(EOF)或Ctrl-C(KeyboardInterrupt) | 捕获EOFError/KeyboardInterrupt后正常退出,并打印一个空行收尾 |
3. 空输入自动跳过
如果直接按 Enter 提交空字符串,循环会continue跳过该轮,不会调用模型、不会产生多余的对话轮次(对应测试见 tests/test_repl.py 中的test_run_demo_loop_skips_empty_input)。
四、完整函数签名与参数说明
run_demo_loop的完整签名(源码见 src/agents/repl.py):
async def run_demo_loop( agent: Agent[Any], *, stream: bool = True, context: TContext | None = None, max_turns: int | None = DEFAULT_MAX_TURNS, ) -> None:| 参数 | 类型 | 默认值 | 作用 |
|---|---|---|---|
agent | Agent[Any] | 必填 | 起始 Agent,即对话开始时执行的那个 Agent |
stream | bool | True | 是否流式输出模型结果,True时边生成边打印 |
context | TContext | None | 传递给 Runner 的额外上下文对象,可携带会话级状态 |
max_turns | int \| None | DEFAULT_MAX_TURNS | 单次运行的最大轮数上限,传None表示不限制 |
其中DEFAULT_MAX_TURNS定义在 src/agents/run_config.py,默认值为10。这里的「turn」指一次模型调用(可能包含若干工具调用),注意它是针对单次 Runner 执行的限制,而 REPL 的每一条用户输入都会触发一次新的 Runner 执行,所以多轮对话总体不受该值限制。
关闭流式输出
若把stream设为False,循环会走非流式路径,等待完整结果后一次性打印final_output(见 src/agents/repl.py):
await run_demo_loop(agent, stream=False)这在输出稳定、不需要实时反馈的场景下更省资源。
五、流式模式底层机制:事件驱动的实时输出
当stream=True(默认)时,run_demo_loop调用Runner.run_streamed(...),然后遍历result.stream_events()产生的事件流,按事件类型分别处理(见 src/agents/repl.py)。
事件类型定义于 src/agents/stream_events.py,主要包括三类:
1. RawResponsesStreamEvent(原始模型流事件)
if isinstance(event, RawResponsesStreamEvent): if isinstance(event.data, ResponseTextDeltaEvent): print(event.data.delta, end="", flush=True)这是 LLM 直接透传的原始事件。其中ResponseTextDeltaEvent携带文本增量(delta),代码用end=""和flush=True实现逐字流式打印——这正是「模型输出边生成边实时显示」的原理。
2. RunItemStreamEvent(语义级运行项事件)
当 Agent 调用工具时,会触发工具调用与工具输出事件,REPL 用[tool called]/[tool output: ...]标记打印:
elif isinstance(event, RunItemStreamEvent): if event.item.type == "tool_call_item": print("\n[tool called]", flush=True) elif event.item.type == "tool_call_output_item": print(f"\n[tool output: {event.item.output}]", flush=True)这让你在终端中就能实时观察到 Agent 正在调用什么工具、工具返回了什么结果,非常适合调试工具链。
3. AgentUpdatedStreamEvent(Agent 切换事件)
elif isinstance(event, AgentUpdatedStreamEvent): print(f"\n[Agent updated: {event.new_agent.name}]", flush=True)当发生 Handoff(Agent 把控制权移交给另一个 Agent)时,打印新 Agent 的名称,让你清楚当前对话由谁在应答。
上述三条分支在 tests/test_repl.py 的test_run_demo_loop_streaming中被完整覆盖:该测试构造了一个单轮内依次触发「工具调用 → 工具输出 → Handoff → 文本回答」的模型脚本,断言终端输出同时包含[tool called]、[tool output: tool_result]、[Agent updated: target]和all done。
六、用 REPL 调试工具调用与多 Agent 场景
流式事件机制让 REPL 成为调试复杂 Agent 的利器。例如,一个带工具和 Handoff 的 Agent:
import asyncio from agents import Agent, function_tool, run_demo_loop @function_tool def get_weather(city: str) -> str: """查询指定城市的天气""" return f"{city} 今天晴,25℃" async def main() -> None: weather_agent = Agent( name="WeatherAgent", instructions="使用天气工具回答用户问题。", tools=[get_weather], ) research_agent = Agent( name="ResearchAgent", instructions="你是研究助手,复杂问题转交给 WeatherAgent。", handoffs=[weather_agent], ) await run_demo_loop(research_agent, max_turns=5) if __name__ == "__main__": asyncio.run(main())运行后你会看到典型的交互过程:
> 北京天气如何? [tool called] [tool output: 北京 今天晴,25℃] 北京今天天气晴朗,气温 25℃。当问题触发 Handoff 时,终端会先打印[Agent updated: WeatherAgent],随后由新 Agent 继续应答——相当于在终端里可视化整个多 Agent 协作链路。
七、调试与测试保障
仓库为run_demo_loop提供了系统性的单元测试(见 tests/test_repl.py),通过ScriptedModel(src/agents/testing)预置模型响应,用monkeypatch模拟builtins.input,从而无需真实 API 即可验证 REPL 的全部行为:
test_run_demo_loop_conversation:验证多轮对话历史确实传递给了模型(断言第二轮的输入包含第一轮的用户消息与模型回答);test_run_demo_loop_streaming:验证流式模式下工具调用、工具输出、Agent 切换三条打印分支;test_run_demo_loop_exits_on_eof:验证Ctrl-D(EOF)能干净退出且不触发任何模型调用;test_run_demo_loop_skips_empty_input:验证空输入被跳过、不污染对话历史。
这些测试直接印证了上文描述的全部行为,如果你想扩展 REPL(例如自定义退出指令或增加彩色输出),可以参考这些测试来保证新行为的正确性。
八、使用建议与注意事项
- 适合快速原型验证:在编写正式 CLI、Web 应用或测试用例之前,用
run_demo_loop几分钟内验证 Agent 的指令、工具和 Handoff 是否符合预期,是最快的反馈回路。 max_turns限制的是单轮执行:若 Agent 单个任务需要多轮工具调用,注意DEFAULT_MAX_TURNS = 10(src/agents/run_config.py)可能不够,可显式调大或传None。context参数可注入会话状态:如果你的工具依赖外部上下文(如用户身份、数据库连接),可通过context传入,Runner 会将其包装进RunContextWrapper供工具访问(相关机制见 src/agents/run_context.py)。- 生产环境请自行封装:
run_demo_loop是面向终端调试的轻量工具,生产环境的多轮对话建议使用 docs/sessions.md 中介绍的 Session 持久化方案,或基于 Runner.run 构建自己的对话循环。
总结
run_demo_loop以约 60 行代码(src/agents/repl.py)实现了终端交互调试所需的全部要素:多轮历史记忆、默认流式输出、工具调用可视化、Handoff 感知和三种退出方式。它既是快速上手 openai-agents-python 的入口,也是日常开发中验证 Agent 行为最直接的工具。结合官方文档 docs/repl.md、API 参考 docs/ref/repl.md 与测试用例 tests/test_repl.py,你可以放心地将它纳入自己的调试工作流。
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考