()
| 20 | |
| 21 | |
| 22 | async def main() -> None: |
| 23 | user_context = {"user_id": "101"} |
| 24 | |
| 25 | customer_directory: dict[str, str] = { |
| 26 | "101": ( |
| 27 | "Customer Kaz S. (tier gold) can be reached at +1-415-555-AAAA. " |
| 28 | "Notes: Prefers SMS follow ups and values concise summaries." |
| 29 | ), |
| 30 | "104": ( |
| 31 | "Customer Yu S. (tier platinum) can be reached at +1-415-555-BBBB. " |
| 32 | "Notes: Recently reported sync issues. Flagged for a proactive onboarding call." |
| 33 | ), |
| 34 | "205": ( |
| 35 | "Customer Ken S. (tier standard) can be reached at +1-415-555-CCCC. " |
| 36 | "Notes: Interested in automation tutorials sent last week." |
| 37 | ), |
| 38 | } |
| 39 | |
| 40 | lookup_customer_profile = create_lookup_customer_profile_tool(directory=customer_directory) |
| 41 | |
| 42 | instructions = ( |
| 43 | "You assist support agents. For every user turn you must call lookup_customer_profile. " |
| 44 | "If a tool reports a transient failure, request approval and retry the same call once before " |
| 45 | "responding. Keep responses under three sentences." |
| 46 | ) |
| 47 | |
| 48 | agent = Agent( |
| 49 | name="File HITL assistant", |
| 50 | instructions=instructions, |
| 51 | tools=[lookup_customer_profile], |
| 52 | ) |
| 53 | |
| 54 | session = FileSession(dir="examples/memory/tmp") |
| 55 | session_id = await session.get_session_id() |
| 56 | print(f"Session id: {session_id}") |
| 57 | print("Enter a message to chat with the agent. Submit an empty line to exit.") |
| 58 | auto_mode = is_auto_mode() |
| 59 | |
| 60 | saved_state = await session.load_state_json() |
| 61 | if saved_state: |
| 62 | print("Found saved run state. Resuming pending interruptions before new input.") |
| 63 | try: |
| 64 | state = await RunState.from_json(agent, saved_state, context_override=user_context) |
| 65 | result = await Runner.run(agent, state, session=session) |
| 66 | while result.interruptions: |
| 67 | state = result.to_state() |
| 68 | for interruption in result.interruptions: |
| 69 | args = format_tool_arguments(interruption) |
| 70 | approved = await prompt_yes_no( |
| 71 | f"Agent {interruption.agent.name} wants to call {interruption.name} with {args or 'no arguments'}" |
| 72 | ) |
| 73 | if approved: |
| 74 | state.approve(interruption) |
| 75 | print("Approved tool call.") |
| 76 | else: |
| 77 | state.reject(interruption) |
| 78 | print("Rejected tool call.") |
| 79 | result = await Runner.run(agent, state, session=session) |
no test coverage detected