diff --git a/Dockerfile.modelscope b/Dockerfile.modelscope index 5a07676..841e04f 100644 --- a/Dockerfile.modelscope +++ b/Dockerfile.modelscope @@ -56,9 +56,9 @@ FROM base AS bytecode-builder # 复制应用代码，仅在构建阶段编译为字节码 COPY . . -RUN python -m compileall -b /app && \ - find /app -type f -name '*.py' -delete && \ - find /app -type d -name '__pycache__' -prune -exec rm -rf {} + +RUN python -m compileall -b -x '(/projects/|/skills/)' /app && \ + find /app -not -path '/app/projects/*' -not -path '/app/skills/*' -type f -name '*.py' -delete && \ + find /app -not -path '/app/projects/*' -not -path '/app/skills/*' -type d -name '__pycache__' -prune -exec rm -rf {} + FROM base AS runtime 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/agent/deep_assistant.py b/agent/deep_assistant.py index fa92c3f..7415413 100644 --- a/agent/deep_assistant.py +++ b/agent/deep_assistant.py @@ -24,6 +24,7 @@ from .guideline_middleware import GuidelineMiddleware from .tool_output_length_middleware import ToolOutputLengthMiddleware from .tool_use_cleanup_middleware import ToolUseCleanupMiddleware from .filepath_fix_middleware import FilePathFixMiddleware +from .mcp_trace_meta import patch_mcp_client_session_trace_meta from utils.settings import ( SUMMARIZATION_MAX_TOKENS, SUMMARIZATION_TOKENS_TO_KEEP, @@ -123,6 +124,7 @@ def read_system_prompt(): async def get_tools_from_mcp(mcp): """Extract tools from MCP configuration with caching.""" + patch_mcp_client_session_trace_meta() start_time = time.time() # Defensive handling: ensure mcp is a non-empty list containing mcpServers if not isinstance(mcp, list) or len(mcp) == 0 or "mcpServers" not in mcp[0]: diff --git a/agent/mcp_trace_meta.py b/agent/mcp_trace_meta.py new file mode 100644 index 0000000..4e4b5c7 --- /dev/null +++ b/agent/mcp_trace_meta.py @@ -0,0 +1,97 @@ +import logging +from functools import wraps +from typing import Any + +try: + from mcp import ClientSession, types +except ImportError: + from mcp.client.session import ClientSession + from mcp import types + +from utils.log_util.context import g + +logger = logging.getLogger("app") + +_PATCHED_ATTR = "_catalog_trace_meta_patched" +_TRACE_META_TOOL_NAMES = {"rag_retrieve", "table_rag_retrieve"} + + +def _get_trace_id() -> str: + try: + trace_id = getattr(g, "trace_id", "") + except (LookupError, KeyError): + return "" + return str(trace_id) if trace_id else "" + + +def _get_tool_name(args: tuple[Any, ...], kwargs: dict[str, Any]) -> str: + name = args[0] if args else kwargs.get("name") + return str(name) if name else "" + + +def patch_mcp_client_session_trace_meta() -> None: + """Attach catalog trace id to MCP tools/call params._meta.""" + if getattr(ClientSession.call_tool, _PATCHED_ATTR, False): + return + + original_call_tool = ClientSession.call_tool + + @wraps(original_call_tool) + async def call_tool_with_trace_meta(self: ClientSession, *args: Any, **kwargs: Any) -> Any: + tool_name = _get_tool_name(args, kwargs) + trace_id = _get_trace_id() if tool_name in _TRACE_META_TOOL_NAMES else "" + if trace_id: + meta = kwargs.get("meta") + if isinstance(meta, dict): + meta = {**meta, "trace_id": meta.get("trace_id") or trace_id} + else: + meta = {"trace_id": trace_id} + kwargs["meta"] = meta + + try: + return await original_call_tool(self, *args, **kwargs) + except TypeError as exc: + if trace_id and "meta" in kwargs and "unexpected keyword argument" in str(exc): + return await _call_tool_with_meta_compat(self, *args, **kwargs) + raise + + setattr(call_tool_with_trace_meta, _PATCHED_ATTR, True) + ClientSession.call_tool = call_tool_with_trace_meta + + +async def _call_tool_with_meta_compat(self: ClientSession, *args: Any, **kwargs: Any) -> Any: + """Call tools/call with _meta for MCP SDK versions before call_tool(meta=...).""" + name = _get_tool_name(args, kwargs) + if not name: + raise TypeError("call_tool() missing required argument: 'name'") + + arguments = args[1] if len(args) > 1 else kwargs.get("arguments", kwargs.get("args")) + read_timeout_seconds = ( + args[2] if len(args) > 2 else kwargs.get("read_timeout_seconds") + ) + progress_callback = ( + args[3] if len(args) > 3 else kwargs.get("progress_callback") + ) + meta = kwargs.get("meta") + + request_meta = meta if isinstance(meta, dict) else None + result = await self.send_request( + types.ClientRequest( + types.CallToolRequest( + params=types.CallToolRequestParams( + name=name, + arguments=arguments, + _meta=request_meta, + ), + ) + ), + types.CallToolResult, + request_read_timeout_seconds=read_timeout_seconds, + progress_callback=progress_callback, + ) + + validate_tool_result = getattr(self, "_validate_tool_result", None) + if validate_tool_result and not result.isError: + await validate_tool_result(name, result) + + return result 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/skills/autoload/onprem/rag-retrieve/rag_retrieve_server.py b/skills/autoload/onprem/rag-retrieve/rag_retrieve_server.py index 2575644..6f308e2 100644 --- a/skills/autoload/onprem/rag-retrieve/rag_retrieve_server.py +++ b/skills/autoload/onprem/rag-retrieve/rag_retrieve_server.py @@ -73,7 +73,7 @@ Format: `` """ -def rag_retrieve(query: str, top_k: int = 100) -> Dict[str, Any]: +def rag_retrieve(query: str, top_k: int = 100, trace_id: str = "") -> Dict[str, Any]: """Call the RAG retrieval API.""" try: bot_id = "" @@ -100,6 +100,8 @@ def rag_retrieve(query: str, top_k: int = 100) -> Dict[str, Any]: "content-type": "application/json", "authorization": f"Bearer {auth_token}" } + if trace_id: + headers["X-Request-ID"] = trace_id data = { "query": query, "top_k": top_k @@ -172,7 +174,7 @@ def rag_retrieve(query: str, top_k: int = 100) -> Dict[str, Any]: } -def table_rag_retrieve(query: str) -> Dict[str, Any]: +def table_rag_retrieve(query: str, trace_id: str = "") -> Dict[str, Any]: """Call the Table RAG retrieval API.""" try: bot_id = "" @@ -189,6 +191,8 @@ def table_rag_retrieve(query: str) -> Dict[str, Any]: "content-type": "application/json", "authorization": f"Bearer {auth_token}" } + if trace_id: + headers["X-Request-ID"] = trace_id data = { "query": query, } @@ -220,7 +224,7 @@ def table_rag_retrieve(query: str) -> Dict[str, Any]: if "markdown" in response_data: markdown_content = response_data["markdown"] if re.search(r"^no excel files found", markdown_content, re.IGNORECASE): - rag_result = rag_retrieve(query) + rag_result = rag_retrieve(query, trace_id=trace_id) content = rag_result.get("content", []) if content and content[0].get("type") == "text": content[0]["text"] = "No table_rag_retrieve results were found. The content below is the fallback result from rag_retrieve:\n\n" + content[0]["text"] @@ -302,6 +306,8 @@ async def handle_request(request: Dict[str, Any]) -> Dict[str, Any]: elif method == "tools/call": tool_name = params.get("name") arguments = params.get("arguments", {}) + meta = params.get("_meta") or params.get("meta") or {} + trace_id = meta.get("trace_id", "") if isinstance(meta, dict) else "" if tool_name == "rag_retrieve": query = arguments.get("query", "") @@ -310,7 +316,7 @@ async def handle_request(request: Dict[str, Any]) -> Dict[str, Any]: if not query: return create_error_response(request_id, -32602, "Missing required parameter: query") - result = rag_retrieve(query, top_k) + result = rag_retrieve(query, top_k, trace_id) return { "jsonrpc": "2.0", @@ -324,7 +330,7 @@ async def handle_request(request: Dict[str, Any]) -> Dict[str, Any]: if not query: return create_error_response(request_id, -32602, "Missing required parameter: query") - result = table_rag_retrieve(query) + result = table_rag_retrieve(query, trace_id) return { "jsonrpc": "2.0", diff --git a/skills/autoload/support/rag-retrieve/rag_retrieve_server.py b/skills/autoload/support/rag-retrieve/rag_retrieve_server.py index 671a456..09e0924 100644 --- a/skills/autoload/support/rag-retrieve/rag_retrieve_server.py +++ b/skills/autoload/support/rag-retrieve/rag_retrieve_server.py @@ -73,7 +73,7 @@ Format: `` """ -def rag_retrieve(query: str, top_k: int = 100) -> Dict[str, Any]: +def rag_retrieve(query: str, top_k: int = 100, trace_id: str = "") -> Dict[str, Any]: """Call the RAG retrieval API.""" try: bot_id = "" @@ -100,6 +100,8 @@ def rag_retrieve(query: str, top_k: int = 100) -> Dict[str, Any]: "content-type": "application/json", "authorization": f"Bearer {auth_token}" } + if trace_id: + headers["X-Request-ID"] = trace_id data = { "query": query, "top_k": top_k @@ -172,7 +174,7 @@ def rag_retrieve(query: str, top_k: int = 100) -> Dict[str, Any]: } -def table_rag_retrieve(query: str) -> Dict[str, Any]: +def table_rag_retrieve(query: str, trace_id: str = "") -> Dict[str, Any]: """Call the Table RAG retrieval API.""" try: bot_id = "" @@ -189,6 +191,8 @@ def table_rag_retrieve(query: str) -> Dict[str, Any]: "content-type": "application/json", "authorization": f"Bearer {auth_token}" } + if trace_id: + headers["X-Request-ID"] = trace_id data = { "query": query, } @@ -220,7 +224,7 @@ def table_rag_retrieve(query: str) -> Dict[str, Any]: if "markdown" in response_data: markdown_content = response_data["markdown"] if re.search(r"^no excel files found", markdown_content, re.IGNORECASE): - rag_result = rag_retrieve(query) + rag_result = rag_retrieve(query, trace_id=trace_id) content = rag_result.get("content", []) if content and content[0].get("type") == "text": content[0]["text"] = "No table_rag_retrieve results were found. The content below is the fallback result from rag_retrieve：\n\n" + content[0]["text"] @@ -302,7 +306,9 @@ async def handle_request(request: Dict[str, Any]) -> Dict[str, Any]: elif method == "tools/call": tool_name = params.get("name") arguments = params.get("arguments", {}) - + meta = params.get("_meta") or params.get("meta") or {} + trace_id = meta.get("trace_id", "") if isinstance(meta, dict) else "" + if tool_name == "rag_retrieve": query = arguments.get("query", "") top_k = arguments.get("top_k", 100) @@ -310,7 +316,7 @@ async def handle_request(request: Dict[str, Any]) -> Dict[str, Any]: if not query: return create_error_response(request_id, -32602, "Missing required parameter: query") - result = rag_retrieve(query, top_k) + result = rag_retrieve(query, top_k, trace_id) return { "jsonrpc": "2.0", @@ -324,7 +330,7 @@ async def handle_request(request: Dict[str, Any]) -> Dict[str, Any]: if not query: return create_error_response(request_id, -32602, "Missing required parameter: query") - result = table_rag_retrieve(query) + result = table_rag_retrieve(query, trace_id) return { "jsonrpc": "2.0", 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=(",", ":")))