openai-agents-python REPL 实用工具:用 run_demo_loop 在终端快速调试智能体
【免费下载链接】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 内置的终端交互调试工具,让你不必编写完整的前端或 Web 界面,就能在命令行里与智能体进行多轮对话、验证提示词效果、测试工具调用与智能体交接(handoff)流程。读完本文,你将掌握run_demo_loop的完整用法、流式与非流式两种运行模式的行为差异、它的底层实现原理,以及如何结合仓库源码与测试用例深入理解它的内部机制。
快速上手:三行代码启动一个交互式聊天会话
run_demo_loop位于src/agents/repl.py,并从 src/agents/init.py 导出,因此可以直接从agents包导入。最基础的用法如下(完整示例见 docs/zh/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())运行这段代码后,终端会进入一个持续循环的交互式会话:程序不断用>提示符请求你的输入,并把每一轮对话(包括智能体的回复)追加到对话历史中,因此智能体能够"记住"此前讨论过的内容。默认情况下,模型输出是实时流式传输的——智能体在生成回复的同时,文字会逐字出现在终端上,而不是等全部生成完才一次性打印。
退出会话的方式有三种:
- 输入
quit并回车; - 输入
exit并回车; - 按下
Ctrl-D快捷键(EOF)。
从实现上看,run_demo_loop内部捕获了EOFError与KeyboardInterrupt异常(即Ctrl-D与Ctrl-C),遇到时先打印一个空行再干净地退出循环;同时它会忽略空白输入行,避免把空内容送进模型(见 src/agents/repl.py)。
完整函数签名与参数说明
run_demo_loop是一个async函数,签名如下(见 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] | 必填 | 起始智能体(starting agent),即 REPL 会话开始时运行的智能体 |
stream | bool | True | 是否流式传输智能体输出。True走Runner.run_streamed,逐块打印文本;False走Runner.run,完成后一次性打印final_output |
context | TContext \| None | None | 透传给 Runner 的上下文信息,可用于向智能体运行注入外部状态 |
max_turns | int \| None | DEFAULT_MAX_TURNS | Runner 单次执行允许的最大轮数;传None可关闭轮数上限 |
其中DEFAULT_MAX_TURNS定义在 src/agents/run_config.py,值为10,即默认情况下智能体在一轮用户输入内最多迭代 10 次(包括模型调用与工具调用循环)。
值得注意的是,max_turns的语义是"单次用户输入触发的执行轮数上限",而不是整个 REPL 会话的总轮数。REPL 会话本身没有总轮数限制——只要不输入退出指令,就可以无限聊下去。如果超过轮数上限,底层 Runner 会抛出MaxTurnsExceeded异常(定义于 src/agents/exceptions.py),此时 REPL 循环会因未捕获异常而终止。
流式模式:实时观察模型生成与工具调用过程
当stream=True(默认)时,run_demo_loop调用Runner.run_streamed并对返回的流式事件做逐类处理(见 src/agents/repl.py):
- 文本增量:当收到
RawResponsesStreamEvent且其数据为ResponseTextDeltaEvent时,用print(event.data.delta, end="", flush=True)逐字打印,flush=True保证字符立即出现在终端; - 工具调用:当收到
RunItemStreamEvent且条目类型为tool_call_item时,打印[tool called]; - 工具输出:当条目类型为
tool_call_output_item时,打印[tool output: <输出内容>],让你直接看到工具返回了什么; - 智能体更新(交接):当收到
AgentUpdatedStreamEvent时,打印[Agent updated: <新智能体名>],提示当前执行已经交接到了另一个智能体。
这意味着,即使你的智能体配置了工具、多智能体交接(handoff),REPL 也能把整个过程可视化地呈现在终端里——你会看到工具被调用、工具返回结果、控制权交接给新智能体,最后才是最终文本回复。这套行为被 tests/test_repl.py 的test_run_demo_loop_streaming用例完整验证:该测试构造了一个"工具调用 → 工具输出 → handoff → 文本回复"的完整流程,并断言输出中同时包含[tool called]、[tool output: tool_result]与[Agent updated: target]。
非流式模式
当stream=False时,run_demo_loop改用await Runner.run(...)一次性执行整个回合,并在结果就绪后打印result.final_output(见 src/agents/repl.py)。这种模式更适合在输出量小、或你需要精确控制终端输出格式的场景下使用。注意非流式模式不会打印工具调用过程,只输出最终文本。
会话状态管理:多轮记忆与智能体切换
REPL 之所以能在多轮之间保留对话历史,关键在于循环体末尾的两行代码(见 src/agents/repl.py):
current_agent = result.last_agent input_items = result.to_input_list()result.last_agent是本次运行结束时的智能体。如果发生了 handoff,它就是交接后的目标智能体,因此下一轮输入会继续由"当前最新智能体"处理,而非最初的起始智能体;result.to_input_list()把本次运行产生的新条目(用户消息、模型回复、工具调用等)转换为下一轮的输入条目列表,从而把整段对话历史无缝传递给下一轮。该方法的实现位于 src/agents/result.py,默认使用mode="preserve_all",即将new_items转换为完整的纯条目历史。
tests/test_repl.py中的test_run_demo_loop_conversation(tests/test_repl.py)验证了多轮记忆:它向 REPL 依次输入Hi与How are you?,随后断言模型在第二轮收到的输入是"第一条用户消息 + 第一轮模型回复 + 第二条用户消息"的完整历史,证明会话状态确实跨轮保留。
输入处理细节:退出、EOF 与空白行
run_demo_loop对用户输入的处理逻辑(见 src/agents/repl.py)包含几个容易被忽略的细节:
- 退出指令不区分大小写:
user_input.strip().lower() in {"exit", "quit"},因此EXIT、Quit等写法都能退出; Ctrl-D与Ctrl-C:输入循环被包在try/except (EOFError, KeyboardInterrupt)中,遇到 EOF 或键盘中断都会先打印空行再break,保证终端状态干净。tests/test_repl.py的test_run_demo_loop_exits_on_eof(tests/test_repl.py)专门验证了 EOF 时循环能干净退出且不会触发任何模型调用;- 空白输入被跳过:空行或纯空白行直接
continue,不会进入对话历史,也不会消耗模型调用。对应测试为test_run_demo_loop_skips_empty_input(tests/test_repl.py)。
源码结构速览
run_demo_loop的实现集中在单个文件 src/agents/repl.py,整个文件只暴露这一个公开函数。它依赖的底层能力包括:
- src/agents/run.py 中的
Runner.run/Runner.run_streamed,负责实际的模型调用与工具执行循环; - src/agents/run_config.py 中的
DEFAULT_MAX_TURNS = 10,作为默认轮数上限; - src/agents/result.py 中的
RunResultBase(last_agent、final_output、to_input_list等); - src/agents/stream_events.py 中的
RawResponsesStreamEvent、RunItemStreamEvent、AgentUpdatedStreamEvent等事件类型,定义了流式模式下可观察的各类事件。
完整的 API 文档页面见 docs/ref/repl.md,其中通过::: agents.repl指令自动生成run_demo_loop的签名与 docstring 文档。docstring 中对各参数的含义也有精确定义(见 src/agents/repl.py):max_turns传None可以禁用轮数上限。
典型使用场景小结
- 提示词调试:快速验证系统提示词(
instructions)的效果,无需搭建 UI; - 工具与交接验证:观察
[tool called]、[tool output: ...]、[Agent updated: ...]输出,确认工具调用链与多智能体交接是否按预期执行; - 上下文注入测试:通过
context参数向智能体运行传入自定义上下文,验证上下文对行为的影响; - 教学与演示:作为最小可运行的交互示例,向团队成员或读者直观展示智能体的多轮对话能力。
一句话总结:run_demo_loop把"创建智能体 → 交互测试 → 观察内部执行过程"压缩成了一段极简代码,是 openai-agents-python 项目中进行快速原型验证和调试的最佳入口之一。
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考