zhuchaowe/qwen_agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge feature/agent-final-answer-first-char into developing

Browse Source
This commit is contained in:
csh28 2026-05-11 14:00:34 +08:00
commit 1e608f0092
7 changed files with 228 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
agent/agent_config.py
View File

@ -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

2
agent/deep_assistant.py
View File

@ -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]:

97
agent/mcp_trace_meta.py Normal file
View File

@ -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

36
routes/chat.py
View File

@ -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)

16
skills/autoload/onprem/rag-retrieve/rag_retrieve_server.py
View File

@ -73,7 +73,7 @@ Format: `<CITATION file="file_id" filename="name.xlsx" sheet=1 rows=[2, 4] />`
"""
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",

18
skills/autoload/support/rag-retrieve/rag_retrieve_server.py
View File

@ -73,7 +73,7 @@ Format: `<CITATION file="file_id" filename="name.xlsx" sheet=1 rows=[2, 4] />`
"""
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",

69
utils/structured_log.py Normal file
View File

@ -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=(",", ":")))