Add agent final answer first char metric

agent/agent_config.py

@@ -32,6 +32,7 @@ class AgentConfig:
     session_id: Optional[str] = None
     dataset_ids: Optional[List[str]] = field(default_factory=list)
     trace_id: Optional[str] = None  # Request trace ID, obtained from the X-Request-ID header
+    request_started_at: Optional[float] = None
 
     # Response control parameters
     stream: bool = False

routes/chat.py

@@ -25,6 +25,7 @@ from agent.agent_config import AgentConfig
 from agent.deep_assistant import init_agent
 from utils.daytona_sync import sync_sandbox_to_local
 from utils.settings import DAYTONA_ENABLED
+from utils.structured_log import emit_question_metric
 
 router = APIRouter()
 
@@ -43,6 +44,7 @@ async def enhanced_generate_stream_response(
 
     # Cancellation management
     cancel_event = None
+    request_started_at = config.request_started_at or time.monotonic()
 
     try:
         # Create output queue and control events
@@ -89,6 +91,8 @@ async def enhanced_generate_stream_response(
                 logger.info(f"Starting agent stream response")
                 chunk_id = 0
                 message_tag = ""
+                last_answer_first_char_duration_ms = None
+                waiting_for_answer_first_char = False
                 agent, checkpointer, sandbox = await init_agent(config)
                 async for msg, metadata in agent.astream({"messages": config.messages}, stream_mode="messages", config=config.invoke_config(), max_tokens=MAX_OUTPUT_TOKENS):
                     # Check whether a cancellation signal was received
@@ -102,6 +106,7 @@ async def enhanced_generate_stream_response(
                         # Handle tool calls
                         if msg.tool_call_chunks:
                             message_tag = "TOOL_CALL"
+                            waiting_for_answer_first_char = False
                             if config.tool_response:
                                 for tool_call_chunk in msg.tool_call_chunks:
                                     chunk_name = tool_call_chunk.get("name") if isinstance(tool_call_chunk, dict) else getattr(tool_call_chunk, "name", None)
@@ -120,12 +125,20 @@ async def enhanced_generate_stream_response(
                                 continue
                             if meta_message_tag != message_tag:
                                 message_tag = meta_message_tag
+                                waiting_for_answer_first_char = meta_message_tag == "ANSWER"
                                 new_content = f"[{meta_message_tag}]\n"
                             if msg.text:
+                                if meta_message_tag == "ANSWER" and waiting_for_answer_first_char and msg.text.strip():
+                                    last_answer_first_char_duration_ms = max(
+                                        int((time.monotonic() - request_started_at) * 1000),
+                                        0,
+                                    )
+                                    waiting_for_answer_first_char = False
                                 new_content += msg.text
                     # Handle tool responses
                     elif isinstance(msg, ToolMessage) and msg.content:
                         message_tag = "TOOL_RESPONSE"
+                        waiting_for_answer_first_char = False
                         if config.tool_response:
                             new_content = f"[{message_tag}] {msg.name}\n{msg.text}\n"
 
@@ -142,6 +155,25 @@ async def enhanced_generate_stream_response(
 
                 # Send final chunk
                 finish = "cancelled" if (cancel_event and cancel_event.is_set()) else "stop"
+                if last_answer_first_char_duration_ms is not None:
+                    emit_question_metric(
+                        stage="catalog_agent.final_answer_first_char",
+                        status="cancel" if finish == "cancelled" else "success",
+                        duration_ms=last_answer_first_char_duration_ms,
+                        first_response_time_ms=last_answer_first_char_duration_ms,
+                        trace_id=config.trace_id,
+                        ai_id=config.bot_id,
+                        session_id=config.session_id,
+                        robot_type="agent",
+                        model=config.model_name,
+                        stream=config.stream,
+                        extra={
+                            "bot_id": config.bot_id,
+                            "tool_response": config.tool_response,
+                            "enable_thinking": config.enable_thinking,
+                            "response_mode": "final_answer_first_char",
+                        },
+                    )
                 final_chunk = create_stream_chunk(f"chatcmpl-{chunk_id + 1}", config.model_name, finish_reason=finish)
                 await output_queue.put(("agent", f"data: {json.dumps(final_chunk, ensure_ascii=False)}\n\n"))
                 # ============ Execute PostAgent hooks ============
@@ -511,6 +543,7 @@ async def chat_completions(request: ChatRequest, authorization: Optional[str] =
         {"dataset_ids": ["project-123", "project-456"], "bot_id": "my-bot-002", "messages": [{"role": "user", "content": "Hello"}]}
         {"dataset_ids": ["project-123"], "bot_id": "my-catalog-bot",  "messages": [{"role": "user", "content": "Hello"}]}
     """
+    request_started_at = time.monotonic()
     try:
         # v1 endpoint: extract the API key from the Authorization header as the model API key
         api_key = extract_api_key_from_auth(authorization)
@@ -531,6 +564,7 @@ async def chat_completions(request: ChatRequest, authorization: Optional[str] =
         messages = process_messages(request.messages, request.language)
         # Create AgentConfig object
         config = await AgentConfig.from_v1_request(request, api_key, project_dir, generate_cfg, messages)
+        config.request_started_at = request_started_at
         # Call the shared agent creation and response generation logic
         return await create_agent_and_generate_response(config)
 
@@ -753,6 +787,7 @@ async def chat_completions_v2(request: ChatRequestV2, authorization: Optional[st
         - Uses MD5 hash of MASTERKEY:bot_id for backend API authentication
         - Optionally uses API key from bot config for model access
     """
+    request_started_at = time.monotonic()
     try:
         # Get bot_id (required parameter)
         bot_id = request.bot_id
@@ -799,6 +834,7 @@ async def chat_completions_v2(request: ChatRequestV2, authorization: Optional[st
         api_key = req_api_key if req_api_key and req_api_key != "whatever" else None
         # Create AgentConfig object
         config = await AgentConfig.from_v2_request(request, bot_config, project_dir, messages, generate_cfg, model_name=model_name, model_server=model_server, api_key=api_key)
+        config.request_started_at = request_started_at
         # Call the shared agent creation and response generation logic
         return await create_agent_and_generate_response(config)
 

utils/structured_log.py

@@ -0,0 +1,69 @@
+import json
+import logging
+import time
+from typing import Any, Optional
+
+logger = logging.getLogger("app")
+
+SCHEMA_VERSION = 1
+
+
+def _normalize_value(value: Any) -> Any:
+    if value is None:
+        return None
+    if isinstance(value, (str, int, float, bool)):
+        return value
+    return str(value)
+
+
+def emit_question_metric(
+    *,
+    stage: str,
+    status: str,
+    duration_ms: Optional[int] = None,
+    first_response_time_ms: Optional[int] = None,
+    trace_id: Optional[str] = None,
+    ai_id: Optional[str] = None,
+    session_id: Optional[str] = None,
+    robot_type: Optional[str] = None,
+    model: Optional[str] = None,
+    stream: Optional[bool] = None,
+    error_type: Optional[str] = None,
+    extra: Optional[dict[str, Any]] = None,
+) -> None:
+    payload: dict[str, Any] = {
+        "schema_version": SCHEMA_VERSION,
+        "event": {
+            "kind": "metric",
+            "category": ["question"],
+            "action": "question_perf",
+        },
+        "stage": stage,
+        "status": status,
+        "observed_at": int(time.time() * 1000),
+        "service": "catalog-agent",
+    }
+
+    optional_fields = {
+        "trace_id": trace_id,
+        "duration_ms": duration_ms,
+        "first_response_time_ms": first_response_time_ms,
+        "ai_id": ai_id,
+        "session_id": session_id,
+        "robot_type": robot_type,
+        "model": model,
+        "stream": stream,
+        "error_type": error_type,
+    }
+    for key, value in optional_fields.items():
+        normalized = _normalize_value(value)
+        if normalized is not None:
+            payload[key] = normalized
+
+    if extra:
+        for key, value in extra.items():
+            normalized = _normalize_value(value)
+            if normalized is not None:
+                payload[key] = normalized
+
+    logger.info(json.dumps(payload, ensure_ascii=False, separators=(",", ":")))