Simplify the run state data

Author: seratch

.gitignore

@@ -148,3 +148,5 @@ cython_debug/
 
 # Redis database files
 dump.rdb
+
+tmp/

examples/memory/file_hitl_example.py

@@ -0,0 +1,149 @@
+"""
+File-backed session example with human-in-the-loop tool approval.
+
+This mirrors the JS `file-hitl.ts` sample: a session persisted on disk and tools that
+require approval before execution.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from typing import Any
+
+from agents import Agent, Runner, function_tool
+from agents.run_context import RunContextWrapper
+from agents.run_state import RunState
+
+from .file_session import FileSession
+
+
+async def main() -> None:
+    user_context = {"user_id": "101"}
+
+    customer_directory: dict[str, str] = {
+        "101": (
+            "Customer Kaz S. (tier gold) can be reached at +1-415-555-AAAA. "
+            "Notes: Prefers SMS follow ups and values concise summaries."
+        ),
+        "104": (
+            "Customer Yu S. (tier platinum) can be reached at +1-415-555-BBBB. "
+            "Notes: Recently reported sync issues. Flagged for a proactive onboarding call."
+        ),
+        "205": (
+            "Customer Ken S. (tier standard) can be reached at +1-415-555-CCCC. "
+            "Notes: Interested in automation tutorials sent last week."
+        ),
+    }
+
+    lookup_customer_profile = create_lookup_customer_profile_tool(directory=customer_directory)
+
+    instructions = (
+        "You assist support agents. For every user turn you must call lookup_customer_profile. "
+        "If a tool reports a transient failure, request approval and retry the same call once before "
+        "responding. Keep responses under three sentences."
+    )
+
+    agent = Agent(
+        name="File HITL assistant",
+        instructions=instructions,
+        tools=[lookup_customer_profile],
+    )
+
+    session = FileSession(dir="examples/memory/tmp")
+    session_id = await session.get_session_id()
+    print(f"Session id: {session_id}")
+    print("Enter a message to chat with the agent. Submit an empty line to exit.")
+
+    saved_state = await session.load_state_json()
+    if saved_state:
+        print("Found saved run state. Resuming pending interruptions before new input.")
+        try:
+            state = await RunState.from_json(agent, saved_state, context_override=user_context)
+            result = await Runner.run(agent, state, session=session)
+            while result.interruptions:
+                state = result.to_state()
+                for interruption in result.interruptions:
+                    args = format_tool_arguments(interruption)
+                    approved = await prompt_yes_no(
+                        f"Agent {interruption.agent.name} wants to call {interruption.name} with {args or 'no arguments'}"
+                    )
+                    if approved:
+                        state.approve(interruption)
+                        print("Approved tool call.")
+                    else:
+                        state.reject(interruption)
+                        print("Rejected tool call.")
+                result = await Runner.run(agent, state, session=session)
+            await session.save_state_json(result.to_state().to_json())
+            reply = result.final_output or "[No final output produced]"
+            print(f"Assistant (resumed): {reply}\n")
+        except Exception as exc:  # noqa: BLE001
+            print(f"Failed to resume saved state: {exc}. Starting a new session.")
+
+    while True:
+        print("You: ", end="", flush=True)
+        loop = asyncio.get_event_loop()
+        user_message = await loop.run_in_executor(None, input)
+        if not user_message.strip():
+            break
+
+        result = await Runner.run(agent, user_message, session=session, context=user_context)
+        while result.interruptions:
+            state = result.to_state()
+            for interruption in result.interruptions:
+                args = format_tool_arguments(interruption)
+                approved = await prompt_yes_no(
+                    f"Agent {interruption.agent.name} wants to call {interruption.name} with {args or 'no arguments'}"
+                )
+                if approved:
+                    state.approve(interruption)
+                    print("Approved tool call.")
+                else:
+                    state.reject(interruption)
+                    print("Rejected tool call.")
+            result = await Runner.run(agent, state, session=session)
+        await session.save_state_json(result.to_state().to_json())
+
+        reply = result.final_output or "[No final output produced]"
+        print(f"Assistant: {reply}\n")
+
+
+def create_lookup_customer_profile_tool(
+    *,
+    directory: dict[str, str],
+    missing_customer_message: str = "No customer found for that id.",
+):
+    @function_tool(
+        name_override="lookup_customer_profile",
+        description_override="Look up stored profile details for a customer by their internal id.",
+        needs_approval=True,
+    )
+    def lookup_customer_profile(ctx: RunContextWrapper[Any]) -> str:
+        return directory.get(ctx.context.get("user_id"), missing_customer_message)
+
+    return lookup_customer_profile
+
+
+def format_tool_arguments(interruption: Any) -> str:
+    args = getattr(interruption, "arguments", None)
+    if args is None:
+        return ""
+    if isinstance(args, str):
+        return args
+    try:
+        return json.dumps(args)
+    except Exception:
+        return str(args)
+
+
+async def prompt_yes_no(question: str) -> bool:
+    print(f"{question} (y/n): ", end="", flush=True)
+    loop = asyncio.get_event_loop()
+    answer = await loop.run_in_executor(None, input)
+    normalized = answer.strip().lower()
+    return normalized in {"y", "yes"}
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/memory/file_session.py

@@ -0,0 +1,121 @@
+"""
+Simple file-backed session implementation for examples.
+
+Persists conversation history as JSON on disk so runs can resume across processes.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+from agents.memory.session import Session
+
+
+class FileSession(Session):
+    """Persist session items to a JSON file on disk."""
+
+    def __init__(self, *, dir: str | Path | None = None, session_id: str | None = None) -> None:
+        self._dir = Path(dir) if dir is not None else Path.cwd() / ".agents-sessions"
+        self.session_id = session_id or ""
+        # Ensure the directory exists up front so subsequent file operations do not race.
+        self._dir.mkdir(parents=True, exist_ok=True)
+
+    async def _ensure_session_id(self) -> str:
+        if not self.session_id:
+            timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
+            # Prefix with wall-clock time so recent sessions are easy to spot on disk.
+            self.session_id = f"{timestamp}-{uuid4().hex[:12]}"
+        await asyncio.to_thread(self._dir.mkdir, parents=True, exist_ok=True)
+        file_path = self._items_path(self.session_id)
+        if not file_path.exists():
+            await asyncio.to_thread(file_path.write_text, "[]", encoding="utf-8")
+        return self.session_id
+
+    async def get_session_id(self) -> str:
+        """Return the session id, creating one if needed."""
+        return await self._ensure_session_id()
+
+    async def get_items(self, limit: int | None = None) -> list[Any]:
+        session_id = await self._ensure_session_id()
+        items = await self._read_items(session_id)
+        if limit is not None and limit >= 0:
+            return items[-limit:]
+        return items
+
+    async def add_items(self, items: list[Any]) -> None:
+        if not items:
+            return
+        session_id = await self._ensure_session_id()
+        current = await self._read_items(session_id)
+        # Deep-copy via JSON to avoid persisting live references that might mutate later.
+        cloned = json.loads(json.dumps(items))
+        await self._write_items(session_id, current + cloned)
+
+    async def pop_item(self) -> Any | None:
+        session_id = await self._ensure_session_id()
+        items = await self._read_items(session_id)
+        if not items:
+            return None
+        popped = items.pop()
+        await self._write_items(session_id, items)
+        return popped
+
+    async def clear_session(self) -> None:
+        if not self.session_id:
+            return
+        file_path = self._items_path(self.session_id)
+        state_path = self._state_path(self.session_id)
+        try:
+            await asyncio.to_thread(file_path.unlink)
+        except FileNotFoundError:
+            pass
+        try:
+            await asyncio.to_thread(state_path.unlink)
+        except FileNotFoundError:
+            pass
+        self.session_id = ""
+
+    def _items_path(self, session_id: str) -> Path:
+        return self._dir / f"{session_id}.json"
+
+    def _state_path(self, session_id: str) -> Path:
+        return self._dir / f"{session_id}-state.json"
+
+    async def _read_items(self, session_id: str) -> list[Any]:
+        file_path = self._items_path(session_id)
+        try:
+            data = await asyncio.to_thread(file_path.read_text, "utf-8")
+            parsed = json.loads(data)
+            return parsed if isinstance(parsed, list) else []
+        except FileNotFoundError:
+            return []
+
+    async def _write_items(self, session_id: str, items: list[Any]) -> None:
+        file_path = self._items_path(session_id)
+        payload = json.dumps(items, indent=2, ensure_ascii=False)
+        await asyncio.to_thread(self._dir.mkdir, parents=True, exist_ok=True)
+        await asyncio.to_thread(file_path.write_text, payload, encoding="utf-8")
+
+    async def load_state_json(self) -> dict[str, Any] | None:
+        """Load a previously saved RunState JSON payload, if present."""
+        session_id = await self._ensure_session_id()
+        state_path = self._state_path(session_id)
+        try:
+            data = await asyncio.to_thread(state_path.read_text, "utf-8")
+            parsed = json.loads(data)
+            return parsed if isinstance(parsed, dict) else None
+        except FileNotFoundError:
+            return None
+
+    async def save_state_json(self, state: dict[str, Any]) -> None:
+        """Persist the serialized RunState JSON payload alongside session items."""
+        session_id = await self._ensure_session_id()
+        state_path = self._state_path(session_id)
+        payload = json.dumps(state, indent=2, ensure_ascii=False)
+        await asyncio.to_thread(self._dir.mkdir, parents=True, exist_ok=True)
+        await asyncio.to_thread(state_path.write_text, payload, encoding="utf-8")