From c071fd12e1a71acae3cb8dd679e4865099399872 Mon Sep 17 00:00:00 2001 From: Ryan Chen Date: Sat, 12 Sep 2026 15:48:19 -0400 Subject: [PATCH] Add structured timing log to stream_query for slow-request diagnosis Every SSE request emits a JSON summary line on completion, cancellation, or error: {"event":"stream_request","conv":"...","user":"ryan", "total_ms":45210,"thinking_ms":38100,"ttft_ms":38200, "tokens":112,"tools":[{"tool":"web_search","ms":820}], "model":"llama","cancelled":false} Fields: - total_ms end-to-end wall time - thinking_ms time from first LLM call to first content token (reasoning) - ttft_ms time to first content token from request start - tokens approximate content token count (stream deltas) - tools each tool invoked with wall-clock duration - model LLM name from LangChain metadata (catches fallback to OpenAI) - cancelled true when client disconnected mid-stream Tool timers use the LangChain run_id to correctly pair start/end even when the same tool is called multiple times in one turn. Co-Authored-By: Claude Sonnet 4.6 --- blueprints/conversation/__init__.py | 65 +++++++++++++++++++++++++++-- 1 file changed, 61 insertions(+), 4 deletions(-) diff --git a/blueprints/conversation/__init__.py b/blueprints/conversation/__init__.py index 0694a0f..3265142 100644 --- a/blueprints/conversation/__init__.py +++ b/blueprints/conversation/__init__.py @@ -225,6 +225,16 @@ async def stream_query(): last_thinking = 0.0 turn_has_content = False + # Timing / observability + t0 = time.monotonic() + llm_start_at: float | None = None + first_content_at: float | None = None + tool_timers: dict[str, float] = {} # run_id -> start time + tool_spans: list[dict] = [] + n_tokens = 0 + model_name: str | None = None + cancelled = False + async def persist(text: str) -> None: await add_message_to_conversation( conversation=conversation, @@ -233,6 +243,34 @@ async def stream_query(): user=user, ) + def _log_summary(*, error: str | None = None) -> None: + now = time.monotonic() + thinking_ms = ( + round((first_content_at - llm_start_at) * 1000) + if llm_start_at and first_content_at + else None + ) + ttft_ms = ( + round((first_content_at - t0) * 1000) if first_content_at else None + ) + logging.info( + json.dumps( + { + "event": "stream_request", + "conv": str(conversation.id), + "user": user.username, + "total_ms": round((now - t0) * 1000), + "thinking_ms": thinking_ms, + "ttft_ms": ttft_ms, + "tokens": n_tokens, + "tools": tool_spans, + "model": model_name, + "cancelled": cancelled, + **({"error": error} if error else {}), + } + ) + ) + try: async for event in main_agent.astream_events( payload, version="v2", config=agent_config @@ -240,9 +278,19 @@ async def stream_query(): event_type = event.get("event") if event_type == "on_tool_start": + tool_timers[event["run_id"]] = time.monotonic() yield f"data: {json.dumps({'type': 'tool_start', 'tool': event['name']})}\n\n" elif event_type == "on_tool_end": + start = tool_timers.pop(event["run_id"], None) + tool_spans.append( + { + "tool": event["name"], + "ms": round((time.monotonic() - start) * 1000) + if start + else None, + } + ) yield f"data: {json.dumps({'type': 'tool_end', 'tool': event['name']})}\n\n" # Persist tool output so later turns keep the fetched context output = event.get("data", {}).get("output") @@ -256,12 +304,21 @@ async def stream_query(): elif event_type == "on_chat_model_start": turn_has_content = False + if llm_start_at is None: + llm_start_at = time.monotonic() + if model_name is None: + model_name = event.get("metadata", {}).get( + "ls_model_name" + ) or event.get("name") elif event_type == "on_chat_model_stream": chunk = event.get("data", {}).get("chunk") content = getattr(chunk, "content", None) if isinstance(content, str) and content: + if first_content_at is None: + first_content_at = time.monotonic() turn_has_content = True + n_tokens += 1 streamed += content yield f"data: {json.dumps({'type': 'content', 'delta': content})}\n\n" elif not turn_has_content: @@ -281,15 +338,14 @@ async def stream_query(): final_message = content except asyncio.CancelledError: - # The client disconnected mid-stream (e.g. a route change aborts the - # fetch). CancelledError is not an Exception, so without this branch - # persistence below would be skipped and the answer lost. Save what - # was generated in a detached task, then honor the cancellation. + cancelled = True + _log_summary() text = final_message or streamed if text: _persist_in_background(persist(text)) raise except Exception as e: + _log_summary(error=str(e)) yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n" final_message = final_message or streamed @@ -302,6 +358,7 @@ async def stream_query(): else: yield f"data: {json.dumps({'type': 'error', 'message': 'No response generated'})}\n\n" + _log_summary() yield "data: [DONE]\n\n" return await make_response(