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 <noreply@anthropic.com>
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
import uuid
|
||||
|
||||
from quart import Blueprint, jsonify, make_response, request
|
||||
@@ -214,6 +215,16 @@ async def stream_query():
|
||||
final_message = None
|
||||
streamed = ""
|
||||
|
||||
# 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,
|
||||
@@ -222,6 +233,35 @@ async def stream_query():
|
||||
user=user,
|
||||
)
|
||||
|
||||
def _log_summary(*, error: str | None = None) -> None:
|
||||
now = time.monotonic()
|
||||
total_ms = round((now - t0) * 1000)
|
||||
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": total_ms,
|
||||
"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
|
||||
@@ -229,15 +269,36 @@ 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"
|
||||
|
||||
elif event_type == "on_chat_model_start":
|
||||
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()
|
||||
n_tokens += 1
|
||||
streamed += content
|
||||
yield f"data: {json.dumps({'type': 'content', 'delta': content})}\n\n"
|
||||
|
||||
@@ -252,27 +313,25 @@ 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
|
||||
if final_message:
|
||||
# Persist via a detached, shielded task so a disconnect at this
|
||||
# point (after generation completes) still commits the answer.
|
||||
task = _persist_in_background(persist(final_message))
|
||||
await asyncio.shield(task)
|
||||
yield f"data: {json.dumps({'type': 'response', 'message': final_message})}\n\n"
|
||||
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(
|
||||
|
||||
Reference in New Issue
Block a user