214 lines
7.8 KiB
Python
214 lines
7.8 KiB
Python
"""
|
||
识别任务：入队 Redis Stream，Worker HTTP 槽认领后处理。
|
||
|
||
本文件职责：enqueue；process_recognition 按 kind 分支。
|
||
kind=vision 跑私聊图片波次，kind=attachment 跑附件材料，
|
||
kind=quote_template_parse 跑报价模板 AI 映射；其它 kind 仍占位。
|
||
禁止：回调线程同步等识别完成；禁止本模块写主账六态。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from dataclasses import dataclass
|
||
from typing import Any, Optional
|
||
|
||
from agent.redis_coord import get_redis_runtime, try_acquire_idempotency
|
||
from agent.redis_coord.runtime import RedisRuntime
|
||
from agent.redis_coord.stream_queue import StreamQueue
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class RecognitionEnqueueResult:
|
||
accepted: bool
|
||
entry_id: str = ""
|
||
reason: str = ""
|
||
|
||
|
||
def recognition_stream(redis: Optional[RedisRuntime] = None) -> StreamQueue:
|
||
"""返回识别 Stream 句柄。"""
|
||
rt = redis or get_redis_runtime()
|
||
return rt.recognition_stream
|
||
|
||
|
||
def enqueue_recognition(
|
||
*,
|
||
kind: str,
|
||
payload: dict[str, Any],
|
||
idempotency_key: str = "",
|
||
redis: Optional[RedisRuntime] = None,
|
||
) -> RecognitionEnqueueResult:
|
||
"""
|
||
识别任务入队（快回）。
|
||
|
||
kind 例：attachment / vision / quote_template_parse。
|
||
"""
|
||
rt = redis or get_redis_runtime()
|
||
if idempotency_key:
|
||
ok = try_acquire_idempotency(
|
||
rt.client,
|
||
scope="recognition",
|
||
idempotency_key=idempotency_key,
|
||
ttl_seconds=600,
|
||
)
|
||
if not ok:
|
||
return RecognitionEnqueueResult(accepted=False, reason="duplicate_idempotency_key")
|
||
body = {"kind": kind, "payload": payload}
|
||
entry_id = rt.recognition_stream.enqueue(
|
||
payload=body,
|
||
idempotency_key=idempotency_key,
|
||
)
|
||
logger.info("recognition.enqueue kind=%s id=%s", kind, entry_id)
|
||
return RecognitionEnqueueResult(accepted=True, entry_id=entry_id)
|
||
|
||
|
||
def process_recognition_stub(data: dict[str, Any]) -> dict[str, Any]:
|
||
"""
|
||
识别占位：回显 kind，不读附件、不调模型。
|
||
|
||
兼容旧调用名；正式路径请用 process_recognition。
|
||
禁止再调本函数以免与 process_recognition 互相递归。
|
||
"""
|
||
return process_recognition(data, allow_network=False)
|
||
|
||
|
||
def process_recognition(data: dict[str, Any], *, allow_network: bool = True) -> dict[str, Any]:
|
||
"""
|
||
Worker 槽：按 kind 处理识别/映射任务。
|
||
|
||
vision：私聊图片波次收口。
|
||
attachment：H5 附件材料收口。
|
||
quote_template_parse：调千问 sheet_map，再 PUT 主账 parseResult。
|
||
禁止在本函数写六态。
|
||
"""
|
||
kind = str(data.get("kind") or "unknown")
|
||
payload = data.get("payload") if isinstance(data.get("payload"), dict) else {}
|
||
if not isinstance(payload, dict):
|
||
payload = {}
|
||
if kind == "vision":
|
||
from agent.handlers.image_inquiry import finish_queued_image
|
||
|
||
return finish_queued_image(payload)
|
||
if kind == "attachment":
|
||
from agent.handlers.attachment_inquiry import finish_queued_attachment
|
||
|
||
return finish_queued_attachment(payload)
|
||
if kind == "quote_template_parse":
|
||
return _process_quote_template_parse(payload, allow_network=allow_network)
|
||
logger.info(
|
||
"recognition.stub kind=%s keys=%s",
|
||
kind,
|
||
sorted(payload.keys()),
|
||
)
|
||
return {
|
||
"ok": True,
|
||
"stub": True,
|
||
"kind": kind,
|
||
"note": "识别壳：未跑真实抽取",
|
||
}
|
||
|
||
|
||
def _process_quote_template_parse(payload: dict[str, Any], *, allow_network: bool) -> dict[str, Any]:
|
||
"""
|
||
跑 sheet_map 并回写主账；失败也必须回写 failed+原因。
|
||
|
||
mergeExisting=true（未换文件重新解析）时：与 existingMapping 补缺合并后再回写，
|
||
旧字段优先保留；换文件/首传不传该标记，整份覆盖。
|
||
"""
|
||
from agent.config import get_settings
|
||
from agent.llm.mode_sheet_map import (
|
||
assess_mapping_gaps,
|
||
invoke_sheet_map,
|
||
merge_mapping_prefer_existing,
|
||
)
|
||
|
||
version_id = str(payload.get("versionId") or "").strip()
|
||
settings = get_settings()
|
||
net = bool(allow_network and settings.llm_allow_network)
|
||
result = invoke_sheet_map(payload, allow_network=net, timeout_seconds=240.0, max_rounds=3)
|
||
merge_existing = bool(payload.get("mergeExisting") or payload.get("merge_existing"))
|
||
existing = payload.get("existingMapping") or payload.get("existing_mapping")
|
||
body: dict[str, Any]
|
||
if result.get("ok") and isinstance(result.get("mapping"), dict):
|
||
mapping = dict(result["mapping"])
|
||
if merge_existing and isinstance(existing, dict):
|
||
mapping = merge_mapping_prefer_existing(existing, mapping)
|
||
# 合并后再对照清单复评缺口，避免角标沿用合并前数字
|
||
remaining = assess_mapping_gaps(payload, mapping)
|
||
mapping["mappingGaps"] = [str(g) for g in remaining if g]
|
||
if not remaining:
|
||
mapping["unmappedRequired"] = []
|
||
mapping["unmappedLabels"] = []
|
||
else:
|
||
gaps = result.get("gaps_after")
|
||
if not isinstance(gaps, list):
|
||
gaps = mapping.get("mappingGaps") if isinstance(mapping.get("mappingGaps"), list) else []
|
||
mapping["mappingGaps"] = [str(g) for g in gaps if g]
|
||
# 运营可读摘要：点数之外固定带上未解析项
|
||
n_fields = len(mapping.get("fields") or {})
|
||
n_fee = len(mapping.get("feeRegions") or []) if isinstance(mapping.get("feeRegions"), list) else 0
|
||
n_route = len(mapping.get("routeRegions") or []) if isinstance(mapping.get("routeRegions"), list) else 0
|
||
points = n_fields + n_fee + n_route
|
||
parse_message = f"已解析 {points} 个映射点"
|
||
if merge_existing:
|
||
parse_message += "（保留原映射并补缺）"
|
||
if mapping.get("mappingGaps"):
|
||
preview = "；".join(mapping["mappingGaps"][:8])
|
||
more = len(mapping["mappingGaps"]) - 8
|
||
parse_message += f"；未解析到 {len(mapping['mappingGaps'])} 项：{preview}"
|
||
if more > 0:
|
||
parse_message += f" 等共{len(mapping['mappingGaps'])}项"
|
||
body = {
|
||
"versionId": version_id,
|
||
"ok": True,
|
||
"mappingJson": mapping,
|
||
"parseStatus": "parsed",
|
||
"parseMessage": parse_message,
|
||
"mergeExisting": merge_existing,
|
||
}
|
||
else:
|
||
err = str(result.get("error") or "AI 解析失败")
|
||
body = {
|
||
"versionId": version_id,
|
||
"ok": False,
|
||
"parseStatus": "failed",
|
||
"parseMessage": err,
|
||
"mergeExisting": merge_existing,
|
||
}
|
||
posted = _post_parse_result(body)
|
||
logger.info(
|
||
"quote_template_parse done versionId=%s ok=%s merge=%s posted=%s",
|
||
version_id,
|
||
body.get("ok"),
|
||
merge_existing,
|
||
posted.get("ok"),
|
||
)
|
||
return {"ok": bool(body.get("ok")), "posted": posted, "kind": "quote_template_parse"}
|
||
|
||
|
||
def _post_parse_result(body: dict[str, Any]) -> dict[str, Any]:
|
||
"""PUT /inquiry/agent/quoteTemplate/parseResult。"""
|
||
try:
|
||
from agent.ledger import LedgerClient
|
||
|
||
client = LedgerClient.from_settings()
|
||
resp = client.request_json(
|
||
"PUT",
|
||
"/inquiry/agent/quoteTemplate/parseResult",
|
||
json_body=body,
|
||
idempotency_key=f"parseResult:{body.get('versionId')}",
|
||
)
|
||
if not resp.ok:
|
||
logger.warning(
|
||
"parseResult 回写失败 status=%s err=%s",
|
||
resp.status_code,
|
||
resp.error,
|
||
)
|
||
return {"ok": False, "error": resp.error or f"http_{resp.status_code}"}
|
||
return {"ok": True, "data": resp.data}
|
||
except Exception as exc:
|
||
logger.exception("parseResult 回写异常")
|
||
return {"ok": False, "error": str(exc)}
|