197 lines
7.6 KiB
Python
197 lines
7.6 KiB
Python
"""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
|
|
from services.webhooks.events import enqueue_analysis_completed
|
|
|
|
|
|
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 enqueue_analysis_completed(session, conversation, result)
|
|
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()
|