feat: 新增 Agent 评估运行流式入口并结合 Langfuse 的观测性

Author: xerrors

backend/package/yuxi/services/agent_eval_run_service.py

@@ -0,0 +1,112 @@
+"""Agent evaluation run service.
+
+This service intentionally does not implement dataset storage or judging. It
+creates a normal conversation-backed AgentRun, blocks until it finishes, and
+returns the run's final result by reusing the shared agent_run base capability.
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+from fastapi import HTTPException
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from yuxi.repositories.agent_repository import AgentRepository
+from yuxi.repositories.conversation_repository import ConversationRepository
+from yuxi.services.agent_run_service import await_agent_run_result, create_agent_run_view
+from yuxi.storage.postgres.models_business import User
+
+EVALUATION_SOURCE = "agent_evaluation"
+EVALUATION_FIELDS = ("dataset_name", "dataset_item_id", "experiment_name")
+MAX_REQUEST_ID_LENGTH = 64
+
+
+def _normalize_evaluation(evaluation: dict[str, Any] | None) -> dict[str, str]:
+    """仅保留已知评估字段，并统一转成去空白的非空字符串。"""
+    if not isinstance(evaluation, dict):
+        return {}
+
+    normalized: dict[str, str] = {}
+    for key in EVALUATION_FIELDS:
+        value = evaluation.get(key)
+        if value is None:
+            continue
+        text = str(value).strip()
+        if text:
+            normalized[key] = text
+    return normalized
+
+
+def _normalize_request_id(meta: dict[str, Any] | None) -> str:
+    """返回去空白并校验长度的 request_id；缺省时生成新的 UUID。"""
+    raw_request_id = (meta or {}).get("request_id")
+    if raw_request_id is None or not str(raw_request_id).strip():
+        return str(uuid.uuid4())
+
+    request_id = str(raw_request_id).strip()
+    if len(request_id) > MAX_REQUEST_ID_LENGTH:
+        raise HTTPException(status_code=422, detail=f"request_id 不能超过 {MAX_REQUEST_ID_LENGTH} 个字符")
+    return request_id
+
+
+async def run_agent_eval(
+    *,
+    query: str,
+    agent_slug: str,
+    evaluation: dict[str, Any] | None,
+    meta: dict[str, Any] | None,
+    image_content: str | None,
+    model_spec: str | None,
+    current_user: User,
+    db: AsyncSession,
+) -> dict[str, Any]:
+    """创建评估 AgentRun，阻塞至运行结束并返回最终结果。
+
+    评估调用方只关心最终输出，因此不做 SSE 流式封装：建 run 后直接复用
+    ``await_agent_run_result`` 等待运行终结并返回结果。注意这会让 HTTP 请求阻塞至
+    运行结束（无中间字节），网关链路上长运行需相应放宽空闲超时。
+    """
+    agent_slug = agent_slug.strip()
+    if not agent_slug:
+        raise HTTPException(status_code=422, detail="agent_slug 不能为空")
+    if not query:
+        raise HTTPException(status_code=422, detail="query 不能为空")
+
+    agent_item = await AgentRepository(db).get_visible_by_slug(slug=agent_slug, user=current_user)
+    if not agent_item:
+        raise HTTPException(status_code=404, detail="智能体不存在")
+
+    evaluation_metadata = _normalize_evaluation(evaluation)
+    request_id = _normalize_request_id(meta)
+    thread_id = str(uuid.uuid4())
+
+    await ConversationRepository(db).create_conversation(
+        uid=str(current_user.uid),
+        agent_id=agent_item.slug,
+        title="Agent Evaluation Run",
+        thread_id=thread_id,
+        metadata={
+            "source": EVALUATION_SOURCE,
+            "evaluation": evaluation_metadata,
+        },
+    )
+
+    run_meta = {
+        "request_id": request_id,
+        "source": EVALUATION_SOURCE,
+        "evaluation": evaluation_metadata,
+        "attachment_file_ids": (meta or {}).get("attachment_file_ids") or [],
+    }
+    run_response = await create_agent_run_view(
+        query=query,
+        agent_id=agent_item.slug,
+        thread_id=thread_id,
+        meta=run_meta,
+        image_content=image_content,
+        current_uid=str(current_user.uid),
+        db=db,
+        model_spec=model_spec,
+    )
+    return await await_agent_run_result(run_id=run_response["run_id"], current_uid=str(current_user.uid))

backend/package/yuxi/services/agent_run_service.py

@@ -7,6 +7,7 @@
 import os
 import uuid
 from collections.abc import AsyncIterator
+from typing import Any
 
 from fastapi import HTTPException
 from sqlalchemy import select
@@ -318,6 +319,8 @@ async def create_agent_run_view(
         "uid": str(current_uid),
         "request_id": request_id,
         "attachment_file_ids": (meta or {}).get("attachment_file_ids") or [],
+        "source": (meta or {}).get("source"),
+        "evaluation": (meta or {}).get("evaluation") or None,
         "created_at": utc_now_naive().isoformat(),
     }
     try:
@@ -344,6 +347,10 @@ async def create_agent_run_view(
             "attachments": [],
             "model_spec": resolved_model_spec,
         }
+        if (meta or {}).get("source"):
+            input_metadata["source"] = (meta or {}).get("source")
+        if (meta or {}).get("evaluation"):
+            input_metadata["evaluation"] = (meta or {}).get("evaluation")
         if run_type == "resume":
             input_metadata["source"] = "ask_user_question_resume"
 
@@ -383,6 +390,77 @@ async def get_agent_run_view(*, run_id: str, current_uid: str, db: AsyncSession)
     return {"run": run.to_dict()}
 
 
+def _select_output_message(messages: list[Message], *, output_message_id: int | None) -> Message | None:
+    """优先选用运行记录的输出消息，否则回退到最后一条 assistant 消息。"""
+    if output_message_id:
+        for message in messages:
+            if message.id == output_message_id and message.role == "assistant":
+                return message
+
+    for message in reversed(messages):
+        if message.role == "assistant":
+            return message
+    return None
+
+
+async def get_agent_run_result(*, run_id: str, current_uid: str, db: AsyncSession) -> dict:
+    """加载某个 run 的最终结果（状态/输出/Langfuse trace/错误），供 chat/eval/cron 等统一复用。"""
+    run = await AgentRunRepository(db).get_run_for_user(run_id, str(current_uid))
+    if not run:
+        return {
+            "status": "failed",
+            "agent_run_id": run_id,
+            "output": "",
+            "error": {"type": "run_not_found", "message": "运行任务不存在"},
+        }
+
+    messages: list[Message] = []
+    if run.conversation_id:
+        result = await db.execute(
+            select(Message)
+            .where(Message.conversation_id == run.conversation_id)
+            .order_by(Message.created_at.asc(), Message.id.asc())
+        )
+        messages = list(result.scalars().unique().all())
+
+    output_message = _select_output_message(messages, output_message_id=run.output_message_id)
+    output_metadata = (
+        output_message.extra_metadata if output_message and isinstance(output_message.extra_metadata, dict) else {}
+    )
+
+    payload: dict[str, Any] = {
+        "status": run.status,
+        "output": output_message.content if output_message else "",
+        "agent_slug": run.agent_id,
+        "thread_id": run.thread_id,
+        "conversation_id": run.conversation_id,
+        "agent_run_id": run.id,
+        "request_id": run.request_id,
+        "final_message_id": output_message.id if output_message else None,
+        "langfuse_trace_id": output_metadata.get("langfuse_trace_id"),
+    }
+    if run.error_type or run.error_message:
+        payload["error"] = {"type": run.error_type, "message": run.error_message}
+    return payload
+
+
+async def load_agent_run_result(*, run_id: str, current_uid: str) -> dict:
+    """自开独立会话读取 run 结果，用于流结束/后台调用等请求会话已不可用的场景。"""
+    async with pg_manager.get_async_session_context() as db:
+        return await get_agent_run_result(run_id=run_id, current_uid=current_uid, db=db)
+
+
+async def await_agent_run_result(*, run_id: str, current_uid: str) -> dict:
+    """阻塞至 run 终结并返回最终结果，供 cron 等 in-process 调用。
+
+    复用有限事件流 ``stream_agent_run_events``：它在 run 终结或超时后自然结束，
+    因此排空即等待，无需额外轮询。等待上限继承事件流内部的 ``SSE_MAX_CONNECTION_MINUTES``。
+    """
+    async for _ in stream_agent_run_events(run_id=run_id, after_seq="0-0", current_uid=current_uid, verbose=False):
+        pass
+    return await load_agent_run_result(run_id=run_id, current_uid=current_uid)
+
+
 async def cancel_agent_run_view(*, run_id: str, current_uid: str, db: AsyncSession) -> dict:
     repo = AgentRunRepository(db)
     run = await repo.get_run_for_user(run_id, str(current_uid))

backend/package/yuxi/services/chat_service.py

@@ -85,7 +85,30 @@ def _build_langfuse_run_context(
     operation: str,
     backend_id: str | None = None,
     message_type: str | None = None,
+    meta: dict | None = None,
 ) -> LangfuseRunContext:
+    extra_metadata = None
+    extra_tags = None
+    evaluation = (meta or {}).get("evaluation") if isinstance(meta, dict) else None
+    # 如果请求来自智能体评测，添加评测相关的 metadata 和 tags，方便在 Langfuse 中进行过滤和分析
+    if (meta or {}).get("source") == "agent_evaluation" or (isinstance(evaluation, dict) and evaluation):
+        extra_metadata = {
+            "source": "agent_evaluation",
+            "feature": "agent_evaluation",
+        }
+        extra_tags = ["agent_evaluation"]
+        if isinstance(evaluation, dict):
+            dataset_name = evaluation.get("dataset_name")
+            experiment_name = evaluation.get("experiment_name")
+            for key in ("dataset_name", "dataset_item_id", "experiment_name"):
+                value = evaluation.get(key)
+                if value:
+                    extra_metadata[f"evaluation_{key}"] = str(value)
+            if dataset_name:
+                extra_tags.append(f"dataset:{dataset_name}")
+            if experiment_name:
+                extra_tags.append(f"experiment:{experiment_name}")
+
     return build_run_context(
         user_id=str(getattr(current_user, "uid", current_user.id)),
         thread_id=thread_id,
@@ -97,6 +120,8 @@ def _build_langfuse_run_context(
         username=getattr(current_user, "username", None),
         login_user_id=getattr(current_user, "uid", None),
         department_id=getattr(current_user, "department_id", None),
+        extra_metadata=extra_metadata,
+        extra_tags=extra_tags,
     )
 
 
@@ -799,6 +824,7 @@ async def agent_chat(
         request_id=meta["request_id"],
         operation="agent_chat_sync",
         message_type=message_type,
+        meta=meta,
     )
     trace_info: dict[str, Any] = {}
 
@@ -1010,6 +1036,7 @@ def make_chunk(content=None, **kwargs):
         request_id=meta["request_id"],
         operation="agent_chat_stream",
         message_type=message_type,
+        meta=meta,
     )
     full_msg = None
     accumulated_content: list[str] = []
@@ -1309,6 +1336,7 @@ def make_resume_chunk(content=None, **kwargs):
         request_id=meta.get("request_id") or str(uuid.uuid4()),
         operation="agent_chat_resume",
         message_type="resume",
+        meta=meta,
     )
     trace_info: dict[str, Any] = {}
     last_agent_state_signature = ""

backend/package/yuxi/services/langfuse_service.py

@@ -70,6 +70,7 @@ def build_trace_metadata(
     username: str | None = None,
     login_user_id: str | None = None,
     department_id: int | str | None = None,
+    extra_metadata: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     metadata: dict[str, Any] = {
         "langfuse_user_id": user_id,
@@ -92,14 +93,25 @@ def build_trace_metadata(
         metadata["login_user_id"] = login_user_id
     if department_id is not None:
         metadata["department_id"] = str(department_id)
+    if extra_metadata:
+        metadata.update(extra_metadata)
 
     return metadata
 
 
-def build_trace_tags(*, agent_id: str, operation: str, message_type: str | None = None) -> list[str]:
+def build_trace_tags(
+    *,
+    agent_id: str,
+    operation: str,
+    message_type: str | None = None,
+    extra_tags: list[str] | None = None,
+) -> list[str]:
     tags = ["yuxi", "chat", operation, f"agent:{agent_id}"]
     if message_type:
         tags.append(f"message_type:{message_type}")
+    for tag in extra_tags or []:
+        if tag and tag not in tags:
+            tags.append(tag)
     return tags
 
 
@@ -115,6 +127,8 @@ def build_run_context(
     username: str | None = None,
     login_user_id: str | None = None,
     department_id: int | str | None = None,
+    extra_metadata: dict[str, Any] | None = None,
+    extra_tags: list[str] | None = None,
 ) -> LangfuseRunContext:
     metadata = build_trace_metadata(
         user_id=user_id,
@@ -127,8 +141,14 @@ def build_run_context(
         username=username,
         login_user_id=login_user_id,
         department_id=department_id,
+        extra_metadata=extra_metadata,
+    )
+    tags = build_trace_tags(
+        agent_id=agent_id,
+        operation=operation,
+        message_type=message_type,
+        extra_tags=extra_tags,
     )
-    tags = build_trace_tags(agent_id=agent_id, operation=operation, message_type=message_type)
 
     client = get_langfuse_client()
     if client is None or CallbackHandler is None:

backend/package/yuxi/services/run_worker.py

@@ -314,6 +314,10 @@ async def process_agent_run(ctx, run_id: str):
         "attachment_file_ids": payload.get("attachment_file_ids") or [],
         "model_spec": payload.get("model_spec"),
     }
+    if payload.get("source"):
+        meta["source"] = payload.get("source")
+    if isinstance(payload.get("evaluation"), dict):
+        meta["evaluation"] = payload.get("evaluation") or {}
 
     await mark_run_running(run_id)
     run_ctx = RunContext(run_id=run_id)
@@ -332,6 +336,8 @@ async def process_agent_run(ctx, run_id: str):
             "agent_id": agent_id,
             "backend_id": payload.get("backend_id"),
             "uid": uid,
+            "source": payload.get("source"),
+            "evaluation": payload.get("evaluation") or {},
         },
         thread_id=thread_id,
     )