diff --git a/agent/agent_config.py b/agent/agent_config.py index c6d8eeb..89b8e87 100644 --- a/agent/agent_config.py +++ b/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 diff --git a/routes/chat.py b/routes/chat.py index 22b20d5..275f6d0 100644 --- a/routes/chat.py +++ b/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) diff --git a/utils/structured_log.py b/utils/structured_log.py new file mode 100644 index 0000000..596048f --- /dev/null +++ b/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=(",", ":")))