diff --git a/backend/app.py b/backend/app.py index 15bb26b..d040ca2 100644 --- a/backend/app.py +++ b/backend/app.py @@ -12,6 +12,7 @@ /api/webrtc/ice-servers WebRTC STUN/TURN 配置 """ +import asyncio import os # NLTK 3.9+ blocks dependency imports when .venv lives under cwd (local dev layout). @@ -24,6 +25,7 @@ import settings import uvicorn from db.session import sync_default_tools, sync_interface_definitions from services.knowledge import recover_interrupted_documents +from services.post_call.worker import recover_interrupted_analyses, run_analysis_worker from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware @@ -48,9 +50,15 @@ async def lifespan(_app: FastAPI): await sync_interface_definitions() await sync_default_tools() await recover_interrupted_documents() + await recover_interrupted_analyses() + analysis_worker = asyncio.create_task( + run_analysis_worker(), name="post-call-analysis-worker" + ) try: yield finally: + analysis_worker.cancel() + await asyncio.gather(analysis_worker, return_exceptions=True) await voice_webrtc.shutdown_active_sessions() diff --git a/backend/db/models.py b/backend/db/models.py index f12fb63..02c385f 100644 --- a/backend/db/models.py +++ b/backend/db/models.py @@ -173,6 +173,7 @@ class Assistant(Base): # ---- 瘦类型专属字段(真列,稀疏:按 type 用其中几列) ---- prompt: Mapped[str] = mapped_column(String(8192), default="") # prompt / opencode dynamic_variable_definitions: Mapped[dict] = mapped_column(JSON, default=dict) + analysis_config: Mapped[dict] = mapped_column(JSONB, default=dict) api_url: Mapped[str] = mapped_column(String(512), default="") # dify / fastgpt / opencode api_key: Mapped[str] = mapped_column(String(512), default="") # dify / fastgpt / opencode(打码/哨兵) app_id: Mapped[str] = mapped_column(String(128), default="") # fastgpt @@ -313,6 +314,11 @@ class ConversationSession(Base): ended_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True ) + analysis_status: Mapped[str] = mapped_column( + String(16), index=True, default="none" + ) + analysis_data: Mapped[dict] = mapped_column(JSONB, default=dict) + analysis_error: Mapped[str] = mapped_column(String(2048), default="") extra: Mapped[dict] = mapped_column(JSONB, default=dict) diff --git a/backend/migrations/versions/20260807_0013_add_post_call_analysis.py b/backend/migrations/versions/20260807_0013_add_post_call_analysis.py new file mode 100644 index 0000000..142add7 --- /dev/null +++ b/backend/migrations/versions/20260807_0013_add_post_call_analysis.py @@ -0,0 +1,75 @@ +"""add post-call analysis configuration and results + +Revision ID: 20260807_0013 +Revises: 20260804_0012 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + + +revision: str = "20260807_0013" +down_revision: str | Sequence[str] | None = "20260804_0012" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.add_column( + "assistants", + sa.Column( + "analysis_config", + postgresql.JSONB(astext_type=sa.Text()), + nullable=False, + server_default=sa.text("'{}'::jsonb"), + ), + ) + op.add_column( + "conversation_sessions", + sa.Column( + "analysis_status", + sa.String(length=16), + nullable=False, + server_default="none", + ), + ) + op.add_column( + "conversation_sessions", + sa.Column( + "analysis_data", + postgresql.JSONB(astext_type=sa.Text()), + nullable=False, + server_default=sa.text("'{}'::jsonb"), + ), + ) + op.add_column( + "conversation_sessions", + sa.Column( + "analysis_error", + sa.String(length=2048), + nullable=False, + server_default="", + ), + ) + op.create_index( + "ix_conversation_sessions_analysis_status", + "conversation_sessions", + ["analysis_status"], + unique=False, + ) + + +def downgrade() -> None: + op.drop_index( + "ix_conversation_sessions_analysis_status", + table_name="conversation_sessions", + ) + op.drop_column("conversation_sessions", "analysis_error") + op.drop_column("conversation_sessions", "analysis_data") + op.drop_column("conversation_sessions", "analysis_status") + op.drop_column("assistants", "analysis_config") diff --git a/backend/models.py b/backend/models.py index c71f6da..07df225 100644 --- a/backend/models.py +++ b/backend/models.py @@ -107,6 +107,7 @@ class AssistantConfig(BaseModel): enableInterrupt: bool = True turnConfig: dict = Field(default_factory=dict) startup: dict = Field(default_factory=dict) + analysis_config: dict = Field(default_factory=dict) # ``tools`` is the complete runtime pool (conversation + lifecycle actions). # ``llm_tool_ids`` limits which tools are advertised to a Prompt model. None diff --git a/backend/routes/assistants.py b/backend/routes/assistants.py index 3f05b1f..a7ad45f 100644 --- a/backend/routes/assistants.py +++ b/backend/routes/assistants.py @@ -277,6 +277,17 @@ async def _validate_vision_model( raise HTTPException(400, "视觉模型必须支持图片输入") +async def _validate_analysis_model( + session: AsyncSession, body: AssistantUpsert +) -> None: + config = body.analysis_config + if not config.enabled: + return + resource = await session.get(ModelResource, config.model_resource_id) + if not resource or not resource.enabled or resource.capability != "LLM": + raise HTTPException(400, "分析模型必须引用已启用的 LLM 模型资源") + + async def _validate_knowledge_base(session: AsyncSession, body: AssistantUpsert) -> None: if body.runtime_mode != "pipeline" or body.type not in {"prompt", "workflow"}: body.knowledge_base_id = None @@ -384,6 +395,7 @@ async def _to_out(session: AsyncSession, assistant: Assistant) -> AssistantOut: tool_ids=await _tool_ids(session, assistant.id), prompt=assistant.prompt, dynamic_variable_definitions=assistant.dynamic_variable_definitions or {}, + analysis_config=assistant.analysis_config or {}, api_url=assistant.api_url, api_key=mask(assistant.api_key), app_id=assistant.app_id, @@ -413,6 +425,7 @@ async def create_assistant( await _validate_system_tool_selection(session, body) await _validate_startup_actions(session, body) await _validate_vision_model(session, body) + await _validate_analysis_model(session, body) await _validate_knowledge_base(session, body) data = body.model_dump() resource_ids = data.pop("model_resource_ids") @@ -459,6 +472,7 @@ async def duplicate_assistant( knowledge_retrieval_config=dict(source.knowledge_retrieval_config or {}), prompt=source.prompt, dynamic_variable_definitions=dict(source.dynamic_variable_definitions or {}), + analysis_config=dict(source.analysis_config or {}), api_url=source.api_url, api_key=source.api_key, app_id=source.app_id, @@ -489,6 +503,7 @@ async def update_assistant( await _validate_system_tool_selection(session, body) await _validate_startup_actions(session, body) await _validate_vision_model(session, body) + await _validate_analysis_model(session, body) await _validate_knowledge_base(session, body) data = body.model_dump() resource_ids = data.pop("model_resource_ids") diff --git a/backend/routes/conversations.py b/backend/routes/conversations.py index 490ac21..028e001 100644 --- a/backend/routes/conversations.py +++ b/backend/routes/conversations.py @@ -8,6 +8,8 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Response from loguru import logger from schemas import ( ConversationArtifactOut, + ConversationAnalysisFieldOut, + ConversationAnalysisOut, ConversationDetailOut, ConversationListOut, ConversationMessageOut, @@ -40,6 +42,33 @@ def _session_out(row: ConversationSession) -> ConversationOut: ) +def _analysis_out(row: ConversationSession) -> ConversationAnalysisOut: + data = row.analysis_data or {} + plan = data.get("plan") if isinstance(data.get("plan"), dict) else {} + definitions = plan.get("fields") if isinstance(plan.get("fields"), list) else [] + result = data.get("result") if isinstance(data.get("result"), dict) else {} + fields = [] + for definition in definitions: + if not isinstance(definition, dict): + continue + name = str(definition.get("name") or "") + if not name: + continue + fields.append( + ConversationAnalysisFieldOut( + name=name, + type=str(definition.get("type") or "string"), + value=result.get(name), + ) + ) + return ConversationAnalysisOut( + status=row.analysis_status or "none", + fields=fields, + error=row.analysis_error or "", + completed_at=data.get("completedAt"), + ) + + @router.get("", response_model=ConversationListOut) async def list_conversations( page: int = Query(1, ge=1), @@ -119,6 +148,7 @@ async def get_conversation( return ConversationDetailOut( **_session_out(conversation).model_dump(), extra=conversation.extra or {}, + analysis=_analysis_out(conversation), messages=[ ConversationMessageOut( id=message.id, diff --git a/backend/schemas.py b/backend/schemas.py index 3634c8b..c82e2f2 100644 --- a/backend/schemas.py +++ b/backend/schemas.py @@ -90,6 +90,45 @@ class KnowledgeRetrievalConfig(CamelModel): return value +class AnalysisField(CamelModel): + id: str = Field(min_length=1, max_length=64) + name: str = Field(pattern=r"^[A-Za-z][A-Za-z0-9_]{0,63}$") + type: Literal["string", "boolean", "integer", "number", "enum"] = "string" + description: str = Field(default="", max_length=500) + enum_values: list[str] = Field(default_factory=list, max_length=30) + + @model_validator(mode="after") + def validate_enum_values(self): + normalized = list( + dict.fromkeys( + value.strip() for value in self.enum_values if value.strip() + ) + ) + if self.type == "enum" and not normalized: + raise ValueError("enum 字段必须至少配置一个枚举值") + self.enum_values = normalized if self.type == "enum" else [] + return self + + +class AnalysisConfig(CamelModel): + enabled: bool = False + model_resource_id: str = "" + fields: list[AnalysisField] = Field(default_factory=list, max_length=20) + + @model_validator(mode="after") + def validate_enabled_config(self): + if not self.enabled: + return self + if not self.model_resource_id: + raise ValueError("开启通话后分析时必须选择分析模型") + if not self.fields: + raise ValueError("开启通话后分析时必须配置至少一个关键信息字段") + names = [field.name for field in self.fields] + if len(names) != len(set(names)): + raise ValueError("关键信息字段名不能重复") + return self + + # 各 type 允许的瘦字段(其余字段写入时清零,防止跨类型脏数据) ALLOWED_FIELDS: dict[str, set[str]] = { "prompt": {"prompt"}, @@ -171,6 +210,7 @@ class AssistantUpsert(CamelModel): dynamic_variable_definitions: dict[str, "DynamicVariableDefinition"] = Field( default_factory=dict ) + analysis_config: AnalysisConfig = Field(default_factory=AnalysisConfig) api_url: str = "" api_key: str = "" # 写时:占位符/空 → 保留旧(哨兵) app_id: str = "" @@ -500,9 +540,25 @@ class ConversationOut(CamelModel): ended_at: datetime | None +class ConversationAnalysisFieldOut(CamelModel): + name: str + type: str + value: Any = None + + +class ConversationAnalysisOut(CamelModel): + status: Literal[ + "none", "pending", "processing", "completed", "failed" + ] = "none" + fields: list[ConversationAnalysisFieldOut] = Field(default_factory=list) + error: str = "" + completed_at: datetime | None = None + + class ConversationDetailOut(ConversationOut): extra: dict[str, Any] = Field(default_factory=dict) messages: list[ConversationMessageOut] = Field(default_factory=list) + analysis: ConversationAnalysisOut = Field(default_factory=ConversationAnalysisOut) class ConversationListOut(CamelModel): diff --git a/backend/services/config_resolver.py b/backend/services/config_resolver.py index c551ca2..9b64fc9 100644 --- a/backend/services/config_resolver.py +++ b/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, diff --git a/backend/services/conversation_history.py b/backend/services/conversation_history.py index 79e62c1..3b2ae83 100644 --- a/backend/services/conversation_history.py +++ b/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}") diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index e4e67c6..1cd9e41 100644 --- a/backend/services/pipecat/pipeline.py +++ b/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" diff --git a/backend/services/post_call/__init__.py b/backend/services/post_call/__init__.py new file mode 100644 index 0000000..c6542be --- /dev/null +++ b/backend/services/post_call/__init__.py @@ -0,0 +1 @@ +"""Post-call structured analysis, independent from the realtime pipeline.""" diff --git a/backend/services/post_call/analyzer.py b/backend/services/post_call/analyzer.py new file mode 100644 index 0000000..ba11215 --- /dev/null +++ b/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() + diff --git a/backend/services/post_call/worker.py b/backend/services/post_call/worker.py new file mode 100644 index 0000000..855c29b --- /dev/null +++ b/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) diff --git a/backend/tests/test_conversation_history.py b/backend/tests/test_conversation_history.py index 52ae028..e0b90f6 100644 --- a/backend/tests/test_conversation_history.py +++ b/backend/tests/test_conversation_history.py @@ -9,6 +9,45 @@ from services.conversation_history import ConversationRecorder class ConversationRecorderTest(unittest.IsolatedAsyncioTestCase): + async def test_completed_conversation_queues_enabled_analysis(self): + conversation = SimpleNamespace( + status="active", + ended_at=None, + message_count=0, + analysis_status="none", + analysis_data={}, + analysis_error="", + ) + + class FakeSession: + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return None + + async def get(self, _model, _session_id): + return conversation + + async def commit(self): + return None + + plan = { + "enabled": True, + "model_resource_id": "model_001", + "fields": [{"name": "customer_name", "type": "string"}], + } + recorder = ConversationRecorder("conv_test", plan) + recorder._sequence = 1 + with patch( + "services.conversation_history.SessionLocal", + return_value=FakeSession(), + ): + await recorder._finish(status="completed") + + self.assertEqual(conversation.analysis_status, "pending") + self.assertEqual(conversation.analysis_data, {"plan": plan}) + async def test_finish_waits_for_database_cleanup_when_cancelled(self): recorder = ConversationRecorder("conv_test") started = asyncio.Event() diff --git a/backend/tests/test_post_call_analysis.py b/backend/tests/test_post_call_analysis.py new file mode 100644 index 0000000..8d68baa --- /dev/null +++ b/backend/tests/test_post_call_analysis.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +import unittest + +from pydantic import ValidationError +from schemas import AnalysisConfig +from services.post_call.analyzer import ( + _json_schema, + _parse_json_content, + _validated_result, +) + + +FIELDS = [ + { + "id": "intent", + "name": "customer_intent", + "type": "enum", + "description": "客户意向", + "enum_values": ["high", "low"], + }, + { + "id": "follow_up", + "name": "need_follow_up", + "type": "boolean", + "description": "是否需要跟进", + "enum_values": [], + }, +] + + +class AnalysisConfigTest(unittest.TestCase): + def test_enabled_analysis_requires_model_and_fields(self): + with self.assertRaises(ValidationError): + AnalysisConfig(enabled=True) + + def test_field_names_must_be_unique(self): + field = { + "id": "one", + "name": "customer_name", + "type": "string", + "description": "客户姓名", + } + with self.assertRaises(ValidationError): + AnalysisConfig( + enabled=True, + model_resource_id="model_001", + fields=[field, {**field, "id": "two"}], + ) + + +class StructuredExtractionTest(unittest.TestCase): + def test_schema_marks_every_field_nullable_and_required(self): + schema = _json_schema(FIELDS) + self.assertEqual( + schema["required"], ["customer_intent", "need_follow_up"] + ) + self.assertEqual( + schema["properties"]["customer_intent"]["enum"], + ["high", "low", None], + ) + + def test_invalid_values_are_normalized_to_null(self): + result = _validated_result( + {"customer_intent": "medium", "need_follow_up": "yes"}, + FIELDS, + ) + self.assertEqual( + result, + {"customer_intent": None, "need_follow_up": None}, + ) + + def test_json_parser_accepts_fenced_model_output(self): + self.assertEqual( + _parse_json_content('```json\n{"need_follow_up": true}\n```'), + {"need_follow_up": True}, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/frontend/src/components/assistant-editor/analysis-config.tsx b/frontend/src/components/assistant-editor/analysis-config.tsx index bcde654..b088631 100644 --- a/frontend/src/components/assistant-editor/analysis-config.tsx +++ b/frontend/src/components/assistant-editor/analysis-config.tsx @@ -1,7 +1,6 @@ "use client"; -import { useState } from "react"; -import { Loader2, Plus, Send, Trash2 } from "lucide-react"; +import { Plus, Trash2 } from "lucide-react"; import { ResourceSelectField, ToggleRow } from "@/components/assistant-editor/editor-controls"; import { Button } from "@/components/ui/button"; @@ -13,47 +12,13 @@ import { SelectTrigger, SelectValue, } from "@/components/ui/select"; +import type { + AnalysisConfig, + AnalysisField, + AnalysisFieldType, +} from "@/lib/api"; -export type AnalysisFieldType = - | "string" - | "boolean" - | "integer" - | "number" - | "enum"; - -export type AnalysisField = { - id: string; - name: string; - type: AnalysisFieldType; - description: string; - enumValues: string[]; -}; - -export type AnalysisConfig = { - enabled: boolean; - modelResourceId: string; - fields: AnalysisField[]; -}; - -export type WebhookConfig = { - url: string; - secret: string; -}; - -export function defaultAnalysisConfig(): AnalysisConfig { - return { - enabled: false, - modelResourceId: "", - fields: [], - }; -} - -export function defaultWebhookConfig(): WebhookConfig { - return { - url: "", - secret: "", - }; -} +export type { AnalysisConfig, AnalysisField, AnalysisFieldType } from "@/lib/api"; const FIELD_TYPE_OPTIONS: Array<{ value: AnalysisFieldType; @@ -76,37 +41,6 @@ function createField(): AnalysisField { }; } -function buildMockPayload(fields: AnalysisField[]) { - const extracted: Record = {}; - for (const field of fields) { - if (!field.name.trim()) continue; - switch (field.type) { - case "boolean": - extracted[field.name] = true; - break; - case "integer": - extracted[field.name] = 1; - break; - case "number": - extracted[field.name] = 1.5; - break; - case "enum": - extracted[field.name] = field.enumValues[0] ?? "option_a"; - break; - default: - extracted[field.name] = "示例值"; - } - } - - return { - event: "call.analysis.completed", - conversation_id: "mock-conv-001", - assistant_id: "mock-assistant-001", - timestamp: new Date().toISOString(), - analysis: extracted, - }; -} - type AnalysisConfigEditorProps = { config: AnalysisConfig; onChange: (config: AnalysisConfig) => void; @@ -186,7 +120,7 @@ export function AnalysisConfigEditor({ 还没有关键信息字段

- 点击右上角加号添加字段，例如「客户意向」「是否预约」等。 + 点击右上角加号添加字段，例如 customer_intent、booked。

) : ( @@ -201,7 +135,7 @@ export function AnalysisConfigEditor({ onChange={(event) => updateField(field.id, { name: event.target.value }) } - placeholder="字段名" + placeholder="字段名，如 customer_intent" aria-label={`字段 ${index + 1} 名称`} className="h-9 border-hairline-strong bg-background" /> @@ -276,111 +210,3 @@ export function AnalysisConfigEditor({ ); } - -type WebhookConfigEditorProps = { - config: WebhookConfig; - onChange: (config: WebhookConfig) => void; - analysisFields: AnalysisField[]; -}; - -export function WebhookConfigEditor({ - config, - onChange, - analysisFields, -}: WebhookConfigEditorProps) { - const [testStatus, setTestStatus] = useState< - "idle" | "loading" | "success" | "error" - >("idle"); - const [testMessage, setTestMessage] = useState(null); - - function patch(partial: Partial) { - onChange({ ...config, ...partial }); - } - - async function sendTestEvent() { - if (!config.url.trim()) { - setTestStatus("error"); - setTestMessage("请先填写 Webhook URL。"); - return; - } - - setTestStatus("loading"); - setTestMessage(null); - - const payload = buildMockPayload(analysisFields); - - await new Promise((resolve) => window.setTimeout(resolve, 900)); - - setTestStatus("success"); - setTestMessage( - `测试事件已模拟发送至 ${config.url.trim()}（Mock，未实际请求网络）。示例 payload：${JSON.stringify(payload)}`, - ); - } - - return ( -
- - - - -
- - {testStatus === "success" && ( - - 模拟发送成功 - - )} - {testStatus === "error" && ( - 发送失败 - )} -
- - {testMessage && ( -

- {testMessage} -

- )} -
- ); -} diff --git a/frontend/src/components/assistant-editor/prompt-editor.tsx b/frontend/src/components/assistant-editor/prompt-editor.tsx index 261dcfd..f04f5f5 100644 --- a/frontend/src/components/assistant-editor/prompt-editor.tsx +++ b/frontend/src/components/assistant-editor/prompt-editor.tsx @@ -3,11 +3,6 @@ import { useEffect, useRef, useState } from "react"; import { AnalysisConfigEditor, - WebhookConfigEditor, - defaultAnalysisConfig, - defaultWebhookConfig, - type AnalysisConfig, - type WebhookConfig, } from "@/components/assistant-editor/analysis-config"; import { Braces, @@ -22,7 +17,6 @@ import { Save, Sparkles, Trash2, - Webhook, Wrench, } from "lucide-react"; @@ -105,7 +99,6 @@ const promptSections = [ { id: "interaction", label: "交互策略" }, { id: "variables", label: "动态变量" }, { id: "analysis", label: "分析" }, - { id: "webhook", label: "Webhook" }, ] as const; type PromptSectionId = (typeof promptSections)[number]["id"]; @@ -169,14 +162,7 @@ export function PromptEditor({ interaction: null, variables: null, analysis: null, - webhook: null, }); - const [analysisConfig, setAnalysisConfig] = useState( - defaultAnalysisConfig, - ); - const [webhookConfig, setWebhookConfig] = useState( - defaultWebhookConfig, - ); const selectedAnchorRef = useRef(null); const [activeSection, setActiveSection] = useState("conversation"); @@ -781,31 +767,14 @@ export function PromptEditor({ description="通话结束后自动提取关键信息" > + updateForm("analysisConfig", analysisConfig) + } modelOptions={llmOptions} /> - -
{ - sectionRefs.current.webhook = element; - }} - className="scroll-mt-3 space-y-3" - > - } - title="Webhook" - description="通话分析完成后，将结果 POST 到指定地址" - > - - -
diff --git a/frontend/src/components/assistant-editor/types.ts b/frontend/src/components/assistant-editor/types.ts index 74db0e8..d4f8b7a 100644 --- a/frontend/src/components/assistant-editor/types.ts +++ b/frontend/src/components/assistant-editor/types.ts @@ -1,4 +1,5 @@ import type { + AnalysisConfig, DynamicVariableDefinition, KnowledgeRetrievalConfig, StartupConfig, @@ -12,6 +13,7 @@ export type AssistantForm = { greeting: string; prompt: string; dynamicVariableDefinitions: Record; + analysisConfig: AnalysisConfig; runtimeMode: RuntimeMode; realtimeModel: string; model: string; diff --git a/frontend/src/components/pages/AssistantPage.tsx b/frontend/src/components/pages/AssistantPage.tsx index e69afc8..aae2a1e 100644 --- a/frontend/src/components/pages/AssistantPage.tsx +++ b/frontend/src/components/pages/AssistantPage.tsx @@ -157,6 +157,11 @@ function blankPromptForm(name: string): AssistantForm { greeting: "", prompt: "", dynamicVariableDefinitions: {}, + analysisConfig: { + enabled: false, + modelResourceId: "", + fields: [], + }, runtimeMode: "pipeline", realtimeModel: "", model: "", @@ -406,6 +411,11 @@ export function AssistantPage(props: AssistantPageProps) { greeting: a.greeting, prompt: a.prompt, dynamicVariableDefinitions: a.dynamicVariableDefinitions ?? {}, + analysisConfig: a.analysisConfig ?? { + enabled: false, + modelResourceId: "", + fields: [], + }, runtimeMode: a.runtimeMode, realtimeModel: a.modelResourceIds.Realtime ?? "", model: a.modelResourceIds.LLM ?? "", @@ -481,6 +491,11 @@ export function AssistantPage(props: AssistantPageProps) { toolIds: [], prompt: "", dynamicVariableDefinitions: {}, + analysisConfig: { + enabled: false, + modelResourceId: "", + fields: [], + }, apiUrl: "", apiKey: "", appId: "", @@ -545,6 +560,7 @@ export function AssistantPage(props: AssistantPageProps) { toolIds: form.toolIds, prompt: form.prompt, dynamicVariableDefinitions: effectiveDynamicVariableDefinitions, + analysisConfig: form.analysisConfig, }), ); } diff --git a/frontend/src/components/pages/CreateAssistantPage.tsx b/frontend/src/components/pages/CreateAssistantPage.tsx index 9d4732c..f1f69d9 100644 --- a/frontend/src/components/pages/CreateAssistantPage.tsx +++ b/frontend/src/components/pages/CreateAssistantPage.tsx @@ -114,6 +114,11 @@ function baseUpsert(over: Partial): AssistantUpsert { toolIds: [], prompt: "", dynamicVariableDefinitions: {}, + analysisConfig: { + enabled: false, + modelResourceId: "", + fields: [], + }, apiUrl: "", apiKey: "", appId: "", diff --git a/frontend/src/components/pages/HistoryPage.tsx b/frontend/src/components/pages/HistoryPage.tsx index ea51305..0e6b7a6 100644 --- a/frontend/src/components/pages/HistoryPage.tsx +++ b/frontend/src/components/pages/HistoryPage.tsx @@ -425,6 +425,35 @@ export function HistoryPage() { } }, []); + useEffect(() => { + if ( + !dialogOpen || + !detail || + !["pending", "processing"].includes(detail.analysis.status) + ) { + return; + } + let stopped = false; + let loadingLatest = false; + const timer = window.setInterval(() => { + if (loadingLatest) return; + loadingLatest = true; + void conversationsApi + .get(detail.id) + .then((latest) => { + if (!stopped) setDetail(latest); + }) + .catch(() => undefined) + .finally(() => { + loadingLatest = false; + }); + }, 2000); + return () => { + stopped = true; + window.clearInterval(timer); + }; + }, [detail, dialogOpen]); + const remove = useCallback( async (conversation: Conversation) => { const label = conversation.assistantName || "调试会话"; @@ -729,7 +758,9 @@ export function HistoryPage() { {detail && detailTab === "session" && ( )} - {detail && detailTab === "analysis" && } + {detail && detailTab === "analysis" && ( + + )} @@ -879,7 +910,79 @@ function SessionInfoPanel({ detail }: { detail: ConversationDetail }) { ); } -function AnalysisPanel() { +function analysisValue(value: unknown): string { + if (value === null || value === undefined || value === "") return "未提取到"; + if (typeof value === "boolean") return value ? "是" : "否"; + if (typeof value === "object") return JSON.stringify(value); + return String(value); +} + +function AnalysisPanel({ + analysis, +}: { + analysis: ConversationDetail["analysis"]; +}) { + if (analysis.status === "pending" || analysis.status === "processing") { + return ( +
+ + 正在分析通话内容 +
+ ); + } + + if (analysis.status === "failed") { + return ( +
+
分析失败
+

+ {analysis.error || "分析服务没有返回有效结果。"} +

+
+ ); + } + + if (analysis.status === "completed") { + return ( +
+
+
关键信息
+ {analysis.completedAt && ( +
+ 完成于 {formatDate(analysis.completedAt)} +
+ )} +
+
+ {analysis.fields.map((field) => ( +
+
+
+ {field.name} +
+
+ {field.type} +
+
+
+ {analysisValue(field.value)} +
+
+ ))} +
+
+ ); + } + return (
@@ -889,7 +992,7 @@ function AnalysisPanel() { 暂无分析结果

- 通话后分析接入后，将在这里展示从对话中提取的关键信息字段。 + 该会话未开启通话后分析，或没有可分析的对话内容。

); diff --git a/frontend/src/components/pages/TemplateLibraryPage.tsx b/frontend/src/components/pages/TemplateLibraryPage.tsx index 7d3ed3e..d2f1d2b 100644 --- a/frontend/src/components/pages/TemplateLibraryPage.tsx +++ b/frontend/src/components/pages/TemplateLibraryPage.tsx @@ -102,6 +102,11 @@ function baseUpsertFromTemplate( toolIds: [], prompt: template.prompt, dynamicVariableDefinitions: {}, + analysisConfig: { + enabled: false, + modelResourceId: "", + fields: [], + }, apiUrl: "", apiKey: "", appId: "", diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index c329f7a..6b79567 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -246,6 +246,27 @@ export type SystemToolKind = | "skip_turn" | "request_human_handoff"; +export type AnalysisFieldType = + | "string" + | "boolean" + | "integer" + | "number" + | "enum"; + +export type AnalysisField = { + id: string; + name: string; + type: AnalysisFieldType; + description: string; + enumValues: string[]; +}; + +export type AnalysisConfig = { + enabled: boolean; + modelResourceId: string; + fields: AnalysisField[]; +}; + /** 后端 AssistantOut(宽表 STI:瘦字段平铺,workflow 用 graph)。apiKey 读时打码 */ export type Assistant = { id: string; @@ -264,6 +285,7 @@ export type Assistant = { toolIds: string[]; prompt: string; dynamicVariableDefinitions: Record; + analysisConfig: AnalysisConfig; apiUrl: string; apiKey: string; appId: string; @@ -354,6 +376,16 @@ export type ConversationDetail = Conversation & { }; workflowTrace?: Array>; }; + analysis: { + status: "none" | "pending" | "processing" | "completed" | "failed"; + fields: Array<{ + name: string; + type: AnalysisFieldType; + value: unknown; + }>; + error: string; + completedAt: string | null; + }; messages: ConversationMessage[]; };