Add analysis feature

backend/services/config_resolver.py

@@ -266,6 +266,7 @@ async def resolve_runtime_config(
         enableInterrupt=assistant.enable_interrupt,
         turnConfig=assistant.turn_config or {},
         startup=assistant.startup or {},
+        analysis_config=assistant.analysis_config or {},
         tools=runtime_tools,
         llm_tool_ids=llm_tool_ids,
         knowledge_base_id=assistant.knowledge_base_id,

backend/services/conversation_history.py

@@ -31,8 +31,9 @@ def _parse_timestamp(value: object) -> datetime:
 class ConversationRecorder:
     """按事件顺序写入一通会话；写库失败不应中断实时通话。"""
 
-    def __init__(self, session_id: str):
+    def __init__(self, session_id: str, analysis_plan: dict | None = None):
         self.session_id = session_id
+        self._analysis_plan = deepcopy(analysis_plan or {})
         self._sequence = 0
         self._trace_sequence = 0
         self._lock = asyncio.Lock()
@@ -49,6 +50,7 @@ class ConversationRecorder:
         runtime_mode: str,
         session_id: str | None = None,
         extra: dict | None = None,
+        analysis_plan: dict | None = None,
     ) -> "ConversationRecorder | None":
         session_id = session_id or f"conv_{uuid4().hex[:20]}"
         try:
@@ -62,11 +64,17 @@ class ConversationRecorder:
                         runtime_mode=runtime_mode,
                         status="active",
                         message_count=0,
+                        analysis_status="none",
+                        analysis_data=(
+                            {"plan": deepcopy(analysis_plan)}
+                            if analysis_plan and analysis_plan.get("enabled")
+                            else {}
+                        ),
                         extra=deepcopy(extra or {}),
                     )
                 )
                 await db.commit()
-            return cls(session_id)
+            return cls(session_id, analysis_plan)
         except Exception as exc:
             logger.error(f"创建对话历史会话失败，不影响本次通话: {exc}")
             return None
@@ -289,6 +297,16 @@ class ConversationRecorder:
                         conversation.status = status
                         conversation.ended_at = datetime.now(UTC)
                         conversation.message_count = self._sequence
+                        if (
+                            status == "completed"
+                            and self._sequence > 0
+                            and self._analysis_plan.get("enabled")
+                        ):
+                            conversation.analysis_status = "pending"
+                            conversation.analysis_error = ""
+                            conversation.analysis_data = {
+                                "plan": deepcopy(self._analysis_plan)
+                            }
                         await db.commit()
             except Exception as exc:
                 logger.error(f"结束对话历史会话失败: {exc}")

backend/services/pipecat/pipeline.py

@@ -623,6 +623,7 @@ async def run_pipeline(
         channel=channel,
         runtime_mode=cfg.runtimeMode,
         session_id=cfg.conversation_id or None,
+        analysis_plan=cfg.analysis_config,
         extra=(workflow_engine.session_metadata() if workflow_engine else None),
     )
     pipeline = Pipeline(
@@ -952,6 +953,7 @@ async def run_realtime_pipeline(
         channel=channel,
         runtime_mode=cfg.runtimeMode,
         session_id=cfg.conversation_id or None,
+        analysis_plan=cfg.analysis_config,
         extra=(
             WorkflowEngine(cfg.graph).session_metadata()
             if cfg.type == "workflow"

backend/services/post_call/__init__.py

@@ -0,0 +1 @@
+"""Post-call structured analysis, independent from the realtime pipeline."""

backend/services/post_call/analyzer.py

@@ -0,0 +1,195 @@
+"""Extract configured structured fields from one completed conversation."""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime
+from typing import Any
+
+import httpx
+from db.models import ConversationMessage, ConversationSession, ModelResource
+from db.session import SessionLocal
+from sqlalchemy import select
+
+
+ANALYSIS_TIMEOUT_SECONDS = 60.0
+MAX_TRANSCRIPT_CHARS = 120_000
+
+
+def _endpoint(base_url: str, path: str) -> str:
+    return f"{base_url.rstrip('/')}/{path.lstrip('/')}"
+
+
+def _json_schema(fields: list[dict[str, Any]]) -> dict[str, Any]:
+    properties: dict[str, Any] = {}
+    for field in fields:
+        field_type = str(field.get("type") or "string")
+        schema: dict[str, Any] = {
+            "description": str(field.get("description") or ""),
+        }
+        if field_type == "enum":
+            schema.update(
+                {
+                    "type": ["string", "null"],
+                    "enum": [*(field.get("enum_values") or []), None],
+                }
+            )
+        else:
+            schema["type"] = [field_type, "null"]
+        properties[str(field["name"])] = schema
+    return {
+        "type": "object",
+        "properties": properties,
+        "required": list(properties),
+        "additionalProperties": False,
+    }
+
+
+def _transcript(messages: list[ConversationMessage]) -> str:
+    lines = [
+        f"[{message.role}] {message.content.strip()}"
+        for message in messages
+        if message.content_type == "text" and message.content.strip()
+    ]
+    transcript = "\n".join(lines)
+    if len(transcript) <= MAX_TRANSCRIPT_CHARS:
+        return transcript
+    return "[较早内容已截断]\n" + transcript[-MAX_TRANSCRIPT_CHARS:]
+
+
+def _parse_json_content(content: object) -> dict[str, Any]:
+    text = str(content or "").strip()
+    if text.startswith("```"):
+        lines = text.splitlines()
+        if lines and lines[0].startswith("```"):
+            lines = lines[1:]
+        if lines and lines[-1].strip() == "```":
+            lines = lines[:-1]
+        text = "\n".join(lines).strip()
+    parsed = json.loads(text)
+    if not isinstance(parsed, dict):
+        raise ValueError("分析模型必须返回 JSON 对象")
+    return parsed
+
+
+def _validated_result(
+    raw: dict[str, Any], fields: list[dict[str, Any]]
+) -> dict[str, Any]:
+    result: dict[str, Any] = {}
+    for field in fields:
+        name = str(field["name"])
+        field_type = str(field.get("type") or "string")
+        value = raw.get(name)
+        valid = value is None
+        if field_type == "string":
+            valid = valid or isinstance(value, str)
+        elif field_type == "boolean":
+            valid = valid or isinstance(value, bool)
+        elif field_type == "integer":
+            valid = valid or (isinstance(value, int) and not isinstance(value, bool))
+        elif field_type == "number":
+            valid = valid or (
+                isinstance(value, (int, float)) and not isinstance(value, bool)
+            )
+        elif field_type == "enum":
+            valid = valid or (
+                isinstance(value, str)
+                and value in list(field.get("enum_values") or [])
+            )
+        result[name] = value if valid else None
+    return result
+
+
+async def _request_analysis(
+    resource: ModelResource,
+    fields: list[dict[str, Any]],
+    transcript: str,
+) -> dict[str, Any]:
+    values = resource.values or {}
+    secrets = resource.secrets or {}
+    api_url = str(values.get("apiUrl") or "")
+    api_key = str(secrets.get("apiKey") or "")
+    model_id = str(values.get("modelId") or "")
+    if resource.interface_type != "openai-llm":
+        raise ValueError(f"分析暂不支持模型接口:{resource.interface_type}")
+    if not api_url or not api_key or not model_id:
+        raise ValueError("分析模型资源缺少 apiUrl、apiKey 或 modelId")
+
+    schema = _json_schema(fields)
+    system_prompt = (
+        "你是通话关键信息提取器。只能使用对话中明确出现的信息，禁止猜测、"
+        "补全或编造。无法确定的字段必须返回 null。严格按照给定 JSON Schema "
+        "返回一个 JSON 对象，不要输出解释或 Markdown。\n\nJSON Schema:\n"
+        + json.dumps(schema, ensure_ascii=False)
+    )
+    async with httpx.AsyncClient(timeout=ANALYSIS_TIMEOUT_SECONDS) as client:
+        response = await client.post(
+            _endpoint(api_url, "chat/completions"),
+            headers={"Authorization": f"Bearer {api_key}"},
+            json={
+                "model": model_id,
+                "messages": [
+                    {"role": "system", "content": system_prompt},
+                    {"role": "user", "content": transcript},
+                ],
+                "temperature": 0,
+                "stream": False,
+                "response_format": {"type": "json_object"},
+            },
+        )
+        response.raise_for_status()
+        payload = response.json()
+    choices = payload.get("choices") if isinstance(payload, dict) else None
+    if not isinstance(choices, list) or not choices:
+        raise ValueError("分析模型没有返回 choices")
+    message = choices[0].get("message") if isinstance(choices[0], dict) else None
+    content = message.get("content") if isinstance(message, dict) else None
+    return _validated_result(_parse_json_content(content), fields)
+
+
+async def analyze_conversation(conversation_id: str) -> None:
+    """Analyze one claimed conversation and persist its terminal state."""
+    try:
+        async with SessionLocal() as session:
+            conversation = await session.get(ConversationSession, conversation_id)
+            if not conversation or conversation.analysis_status != "processing":
+                return
+            data = dict(conversation.analysis_data or {})
+            plan = data.get("plan") if isinstance(data.get("plan"), dict) else {}
+            fields = plan.get("fields") if isinstance(plan.get("fields"), list) else []
+            resource_id = str(plan.get("model_resource_id") or "")
+            resource = await session.get(ModelResource, resource_id)
+            if not resource or not resource.enabled or resource.capability != "LLM":
+                raise ValueError("分析模型不存在、未启用或不是 LLM 资源")
+            messages = (
+                await session.execute(
+                    select(ConversationMessage)
+                    .where(ConversationMessage.session_id == conversation_id)
+                    .order_by(ConversationMessage.sequence)
+                )
+            ).scalars().all()
+
+        transcript = _transcript(list(messages))
+        if not transcript:
+            raise ValueError("会话没有可分析的文本转写")
+        result = await _request_analysis(resource, list(fields), transcript)
+
+        async with SessionLocal() as session:
+            conversation = await session.get(ConversationSession, conversation_id)
+            if not conversation:
+                return
+            data = dict(conversation.analysis_data or {})
+            data["result"] = result
+            data["completedAt"] = datetime.now(UTC).isoformat()
+            conversation.analysis_data = data
+            conversation.analysis_status = "completed"
+            conversation.analysis_error = ""
+            await session.commit()
+    except Exception as exc:
+        async with SessionLocal() as session:
+            conversation = await session.get(ConversationSession, conversation_id)
+            if conversation:
+                conversation.analysis_status = "failed"
+                conversation.analysis_error = str(exc)[:2048]
+                await session.commit()
+

backend/services/post_call/worker.py

@@ -0,0 +1,59 @@
+"""Small PostgreSQL-backed worker for post-call analysis."""
+
+from __future__ import annotations
+
+import asyncio
+
+from db.models import ConversationSession
+from db.session import SessionLocal
+from loguru import logger
+from services.post_call.analyzer import analyze_conversation
+from sqlalchemy import select, update
+
+
+POLL_INTERVAL_SECONDS = 2.0
+
+
+async def recover_interrupted_analyses() -> None:
+    """Return work interrupted by a previous process shutdown to the queue."""
+    async with SessionLocal() as session:
+        await session.execute(
+            update(ConversationSession)
+            .where(ConversationSession.analysis_status == "processing")
+            .values(analysis_status="pending", analysis_error="")
+        )
+        await session.commit()
+
+
+async def claim_pending_analysis() -> str | None:
+    async with SessionLocal() as session:
+        row = (
+            await session.execute(
+                select(ConversationSession)
+                .where(ConversationSession.analysis_status == "pending")
+                .order_by(ConversationSession.ended_at)
+                .with_for_update(skip_locked=True)
+                .limit(1)
+            )
+        ).scalar_one_or_none()
+        if not row:
+            return None
+        row.analysis_status = "processing"
+        row.analysis_error = ""
+        await session.commit()
+        return row.id
+
+
+async def run_analysis_worker() -> None:
+    logger.info("通话后分析 worker 已启动")
+    while True:
+        try:
+            conversation_id = await claim_pending_analysis()
+            if conversation_id:
+                await analyze_conversation(conversation_id)
+                continue
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:
+            logger.error(f"通话后分析 worker 暂时不可用:{exc}")
+        await asyncio.sleep(POLL_INTERVAL_SECONDS)