Agent 出了 bug 你怎么查 — 交互日志与实时调试

class InteractionLogger:
    """Persist complete LLM request/response payloads for a session."""

    def __init__(
        self,
        log_dir: str | Path = ".logs",
        session_id: str = "default",
        *,
        pretty: bool = True,
    ) -> None:
        self._log_dir = Path(log_dir)
        self._session_id = self._validate_session_id(session_id)
        self._pretty = pretty
        self._seq = 0
        self._session_dir: Path | None = None
        self._event_lock = threading.Lock()
        self._event_subscribers: set[queue.Queue[dict[str, Any]]] = set()

.logs/
├── session-abc123/
│   ├── 000_request.json
│   ├── 000_response.json
│   ├── 001_request.json
│   ├── 001_response.json
│   └── 002_request.json      ← response 可能因异常缺失
└── session-def456/
    ├── 000_request.json
    └── 000_response.json

def _safe_log_request(
    *,
    interaction_logger,
    messages, tools, provider, console,
) -> tuple[int | None, float]:
    if interaction_logger is None:
        return None, 0.0
    try:
        log_seq = interaction_logger.log_request(
            messages, tools,
            provider_info=_provider_info(provider),
        )
    except Exception as exc:
        _report_log_failure(console, "request", exc)
        return None, 0.0
    return log_seq, time.monotonic()


def _safe_log_response(
    *,
    interaction_logger, log_seq, console,
    **kwargs,
) -> None:
    if interaction_logger is None or log_seq is None:
        return
    try:
        interaction_logger.log_response(log_seq, **kwargs)
    except Exception as exc:
        _report_log_failure(console, "response", exc)

log_seq, log_started_at = _safe_log_request(
    interaction_logger=interaction_logger,
    messages=messages, tools=tools,
    provider=provider, console=console,
)

try:
    response = _invoke_provider(provider, messages, tools, ...)
except BaseException as exc:
    _safe_log_response(
        interaction_logger=interaction_logger,
        log_seq=log_seq, console=console,
        duration_ms=(time.monotonic() - log_started_at) * 1000,
        error=str(exc) or type(exc).__name__,
    )
    raise

_safe_log_response(
    interaction_logger=interaction_logger,
    log_seq=log_seq, console=console,
    text=response.text, thinking=response.thinking,
    tool_calls=_serialize_tool_calls(response.tool_calls),
    input_tokens=response.input_tokens,
    output_tokens=response.output_tokens,
    duration_ms=(time.monotonic() - log_started_at) * 1000,
)

def _push_event(self, event_type, seq, payload):
    event = {
        "event": event_type,
        "session_id": self._session_id,
        "seq": seq,
        "timestamp": payload.get("timestamp"),
    }
    with self._event_lock:
        subscribers = tuple(self._event_subscribers)

    for event_queue in subscribers:
        self._publish_event(event_queue, event)

def _publish_event(self, event_queue, event):
    try:
        event_queue.put_nowait(event)
        return
    except queue.Full:
        pass

    # queue 满了：丢掉最旧的事件，腾出空间
    try:
        event_queue.get_nowait()
    except queue.Empty:
        return

    try:
        event_queue.put_nowait(event)
    except queue.Full:
        return

class DebugViewerServer(ThreadingMixIn, HTTPServer):
    daemon_threads = True
    allow_reuse_address = True

    def __init__(self, logger, port=8321, host="127.0.0.1"):
        self.logger = logger
        super().__init__((host, port), DebugViewerHandler)

def _serve_sse(self):
    self.send_response(200)
    self.send_header("Content-Type", "text/event-stream")
    self.send_header("Cache-Control", "no-cache")
    self.send_header("Connection", "keep-alive")
    self.end_headers()

    event_queue = self.server.logger.subscribe_events()
    try:
        while True:
            try:
                event = event_queue.get(timeout=30)
                payload = json.dumps(event, ensure_ascii=False, default=str)
                self.wfile.write(f"data: {payload}\n\n".encode("utf-8"))
            except Empty:
                self.wfile.write(b": heartbeat\n\n")
            self.wfile.flush()
    except (BrokenPipeError, ConnectionResetError, OSError):
        return
    finally:
        self.server.logger.unsubscribe_events(event_queue)

def _validate_session_id(self, session_id):
    value = str(session_id)
    if value in {"", ".", ".."}:
        raise ValueError("Session ID must be a single relative path segment.")
    if "/" in value or "\\" in value:
        raise ValueError("Session ID must not contain path separators.")

    posix_path = PurePosixPath(value)
    windows_path = PureWindowsPath(value)
    if posix_path.is_absolute() or windows_path.is_absolute():
        raise ValueError("Session ID must not be an absolute path.")
    if len(posix_path.parts) != 1 or len(windows_path.parts) != 1:
        raise ValueError("Session ID must be a single path segment.")
    return value

json.dumps(payload, ensure_ascii=False, indent=2, default=str)