Run a simple REPL loop with the given agent. This utility allows quick manual testing and debugging of an agent from the command line. Conversation state is preserved across turns. Enter ``exit`` or ``quit`` to stop the loop. Args: agent: The starting agent to run.
(
agent: Agent[Any],
*,
stream: bool = True,
context: TContext | None = None,
max_turns: int | None = DEFAULT_MAX_TURNS,
)
| 13 | |
| 14 | |
| 15 | async def run_demo_loop( |
| 16 | agent: Agent[Any], |
| 17 | *, |
| 18 | stream: bool = True, |
| 19 | context: TContext | None = None, |
| 20 | max_turns: int | None = DEFAULT_MAX_TURNS, |
| 21 | ) -> None: |
| 22 | """Run a simple REPL loop with the given agent. |
| 23 | |
| 24 | This utility allows quick manual testing and debugging of an agent from the |
| 25 | command line. Conversation state is preserved across turns. Enter ``exit`` |
| 26 | or ``quit`` to stop the loop. |
| 27 | |
| 28 | Args: |
| 29 | agent: The starting agent to run. |
| 30 | stream: Whether to stream the agent output. |
| 31 | context: Additional context information to pass to the runner. |
| 32 | max_turns: Maximum number of turns for the runner to iterate. Pass ``None`` to disable |
| 33 | the turn limit. |
| 34 | """ |
| 35 | |
| 36 | current_agent = agent |
| 37 | input_items: list[TResponseInputItem] = [] |
| 38 | while True: |
| 39 | try: |
| 40 | user_input = input(" > ") |
| 41 | except (EOFError, KeyboardInterrupt): |
| 42 | print() |
| 43 | break |
| 44 | if user_input.strip().lower() in {"exit", "quit"}: |
| 45 | break |
| 46 | if not user_input: |
| 47 | continue |
| 48 | |
| 49 | input_items.append({"role": "user", "content": user_input}) |
| 50 | |
| 51 | result: RunResultBase |
| 52 | if stream: |
| 53 | result = Runner.run_streamed( |
| 54 | current_agent, input=input_items, context=context, max_turns=max_turns |
| 55 | ) |
| 56 | async for event in result.stream_events(): |
| 57 | if isinstance(event, RawResponsesStreamEvent): |
| 58 | if isinstance(event.data, ResponseTextDeltaEvent): |
| 59 | print(event.data.delta, end="", flush=True) |
| 60 | elif isinstance(event, RunItemStreamEvent): |
| 61 | if event.item.type == "tool_call_item": |
| 62 | print("\n[tool called]", flush=True) |
| 63 | elif event.item.type == "tool_call_output_item": |
| 64 | print(f"\n[tool output: {event.item.output}]", flush=True) |
| 65 | elif isinstance(event, AgentUpdatedStreamEvent): |
| 66 | print(f"\n[Agent updated: {event.new_agent.name}]", flush=True) |
| 67 | print() |
| 68 | else: |
| 69 | result = await Runner.run( |
| 70 | current_agent, input_items, context=context, max_turns=max_turns |
| 71 | ) |
| 72 | if result.final_output is not None: |