diff --git a/app/agent/cost.py b/app/agent/cost.py new file mode 100644 index 0000000..e8c416b --- /dev/null +++ b/app/agent/cost.py @@ -0,0 +1,126 @@ +"""模型调用成本与延迟日志（EVA-03）。 + +Agent 层不查库，这里用进程内列表 + 结构化日志记录每次模型调用的 +路由、模型名、模式、耗时、token、错误与降级状态。后端可在请求结束后 +从 get_call_logs() 读取并落库到 AIAnalysis 表。 + +session_id 通过 contextvar 注入：run_copilot / extract_promises 在入口设置， +所有嵌套调用自动带上当前会话，用于按会话统计成本。 +""" + +import contextvars +import json +import logging +import threading +from dataclasses import asdict, dataclass +from datetime import UTC, datetime +from typing import Any + +from langchain_core.callbacks import BaseCallbackHandler + +logger = logging.getLogger("app.agent.cost") + +_current_session: contextvars.ContextVar[str | None] = contextvars.ContextVar("cost_session_id", default=None) + + +@dataclass +class ModelCallRecord: + session_id: str | None + route: str + model: str + mode: str + latency_ms: int + input_tokens: int + output_tokens: int + total_tokens: int + error: str | None + degraded: bool + started_at: str + + +class TokenCapture(BaseCallbackHandler): + """在 LLM 调用结束时捕获 token 用量，供成本日志使用。""" + + def __init__(self) -> None: + self.input_tokens = 0 + self.output_tokens = 0 + self.total_tokens = 0 + + def on_llm_end(self, response: Any, **kwargs: Any) -> None: + try: + first = response.generations[0][0] + message = getattr(first, "message", None) + if message is None: + return + usage = getattr(message, "usage_metadata", None) + if not usage: + usage = (getattr(message, "response_metadata", None) or {}).get("token_usage") + if not usage: + return + self.input_tokens = int(usage.get("input_tokens") or 0) + self.output_tokens = int(usage.get("output_tokens") or 0) + self.total_tokens = int(usage.get("total_tokens") or (self.input_tokens + self.output_tokens)) + except Exception: + # token 统计失败不阻断主流程，成本日志只是观测 + return + + +_records: list[ModelCallRecord] = [] +_lock = threading.Lock() + + +def set_session(session_id: str | None) -> None: + _current_session.set(session_id) + + +def record_call( + *, + route: str, + model: str, + mode: str, + latency_ms: int, + input_tokens: int = 0, + output_tokens: int = 0, + total_tokens: int = 0, + error: str | None = None, + degraded: bool = False, +) -> None: + """记录一次模型调用（含 Mock / 降级调用，token 为 0）。""" + rec = ModelCallRecord( + session_id=_current_session.get(), + route=route, + model=model, + mode=mode, + latency_ms=latency_ms, + input_tokens=input_tokens, + output_tokens=output_tokens, + total_tokens=total_tokens, + error=error, + degraded=degraded, + started_at=datetime.now(UTC).isoformat(), + ) + with _lock: + _records.append(rec) + logger.info("model_call %s", json.dumps(asdict(rec), ensure_ascii=False)) + + +def get_call_logs() -> list[ModelCallRecord]: + with _lock: + return list(_records) + + +def clear_call_logs() -> None: + with _lock: + _records.clear() + + +def session_totals(session_id: str) -> dict[str, int]: + """按会话聚合：总调用数、总 token、总耗时。""" + calls = [r for r in get_call_logs() if r.session_id == session_id] + return { + "calls": len(calls), + "input_tokens": sum(r.input_tokens for r in calls), + "output_tokens": sum(r.output_tokens for r in calls), + "total_tokens": sum(r.total_tokens for r in calls), + "latency_ms": sum(r.latency_ms for r in calls), + } diff --git a/app/agent/graph.py b/app/agent/graph.py index 1658571..5114133 100644 --- a/app/agent/graph.py +++ b/app/agent/graph.py @@ -9,7 +9,9 @@ from app.agent.nodes.reply import reply_node from app.agent.nodes.risk_engine import risk_engine_node from app.agent.nodes.risk_extraction import risk_extraction_node +from app.agent.nodes.summary import summary_node from app.agent.nodes.tool_query import tool_query_node +from app.agent.nodes.vision import vision_node from app.agent.state import CustomerState @@ -27,6 +29,8 @@ def route_by_risk(state: CustomerState) -> str: builder.add_node("risk_engine", risk_engine_node) builder.add_node("tool_query", tool_query_node) builder.add_node("evidence", evidence_node) +builder.add_node("summary", summary_node) +builder.add_node("vision", vision_node) builder.add_node("adverse", adverse_node) builder.add_node("reply", reply_node) builder.add_node("fact_check", fact_check_node) @@ -39,11 +43,15 @@ def route_by_risk(state: CustomerState) -> str: # 事实与证据在分支之前完成，保证两条路都拿得到证据 builder.add_edge("risk_engine", "tool_query") builder.add_edge("tool_query", "evidence") +# 轨迹摘要在分支前完成，普通与高风险两条路都能拿到历史轨迹 +builder.add_edge("evidence", "summary") builder.add_conditional_edges( - "evidence", + "summary", route_by_risk, - {"adverse": "adverse", "normal": "reply"}, + {"adverse": "vision", "normal": "reply"}, ) +# 高风险分支：先识别图片，再走不良反应专项处置 +builder.add_edge("vision", "adverse") builder.add_edge("adverse", "reply") builder.add_edge("reply", "fact_check") builder.add_edge("fact_check", END) diff --git a/app/agent/llm.py b/app/agent/llm.py index b493e0c..0158d1d 100644 --- a/app/agent/llm.py +++ b/app/agent/llm.py @@ -1,10 +1,12 @@ import logging +import time from collections.abc import Callable from typing import TypeVar from langchain_core.output_parsers import PydanticOutputParser from pydantic import BaseModel +from app.agent import cost from app.integrations.model.gateway import ModelUnavailableError, gateway logger = logging.getLogger(__name__) @@ -40,21 +42,40 @@ def run_structured[T: BaseModel]( auto —— 有配置就走模型，无配置或调用失败都降级到 fallback real —— 强制走模型，缺配置或失败直接抛异常（仅开发/评测用） """ + model = gateway.model_name_for(route) + if mode == "mock": + cost.record_call(route=route, model="", mode=mode, latency_ms=0, degraded=True) return fallback(), True if not gateway.is_available(route): if mode == "real": raise ModelUnavailableError(f"mode=real 但路由 {route} 未配置模型，无法执行") logger.warning("模型路由 %s 不可用，降级到 Mock 结果", route) + cost.record_call(route=route, model=model, mode=mode, latency_ms=0, error="route unavailable", degraded=True) return fallback(), True parser = parser_for(schema) chain = gateway.get(route).bind(response_format={"type": "json_object"}) | parser + token_capture = cost.TokenCapture() + started = time.monotonic() try: - return chain.invoke(prompt), False - except Exception: + result = chain.invoke(prompt, config={"callbacks": [token_capture]}) + latency_ms = int((time.monotonic() - started) * 1000) + cost.record_call( + route=route, + model=model, + mode=mode, + latency_ms=latency_ms, + input_tokens=token_capture.input_tokens, + output_tokens=token_capture.output_tokens, + total_tokens=token_capture.total_tokens, + ) + return result, False + except Exception as exc: + latency_ms = int((time.monotonic() - started) * 1000) if mode == "real": raise logger.exception("模型调用失败，降级到 Mock 结果（route=%s）", route) + cost.record_call(route=route, model=model, mode=mode, latency_ms=latency_ms, error=str(exc), degraded=True) return fallback(), True diff --git a/app/agent/mock.py b/app/agent/mock.py index 19d13a3..d2a2b11 100644 --- a/app/agent/mock.py +++ b/app/agent/mock.py @@ -15,6 +15,8 @@ PromiseExtraction, ReplyDraft, RiskExtraction, + TrajectorySummary, + VisionExtraction, ) # 不良反应症状词 @@ -55,12 +57,27 @@ _ORDER_NO_RE = re.compile(r"\d{6,}") +# 否定词：紧邻关键词之前的否定会取消命中（「没有红肿」「不痒」不算不良反应） +_NEGATION_WORDS = ("没有", "没", "不是", "并非", "不", "无", "未", "否认", "排除") + + +def _is_negated(text: str, pos: int) -> bool: + """判断 text 中 pos 处的关键词是否被其前面紧邻的否定词修饰。""" + window = text[max(0, pos - 4) : pos] + return any(neg in window for neg in _NEGATION_WORDS) + + +def _word_hit(text: str, word: str) -> bool: + """关键词命中且至少有一处未被否定。""" + return any(not _is_negated(text, m.start()) for m in re.finditer(re.escape(word), text)) + + def _hit(text: str, words: tuple[str, ...]) -> bool: - return any(word in text for word in words) + return any(_word_hit(text, word) for word in words) def _hits(text: str, words: tuple[str, ...]) -> list[str]: - return [word for word in words if word in text] + return [word for word in words if _word_hit(text, word)] def mock_intent(message: str) -> IntentAnalysis: @@ -172,3 +189,54 @@ def mock_promise(message_text: str) -> PromiseExtraction: owner_type="agent", confidence=0.6 if has_promise else 0.0, ) + + +def mock_vision(images: list[dict]) -> VisionExtraction: + """图片识别 Mock。 + + 有真实图片数据（url / b64 / path）时返回确定性的「清晰 / 患处照片」； + 只有占位（如 [图片]）时返回「待识别」，交人工确认 —— 绝不编造图中内容。 + """ + if not images: + return VisionExtraction() + + first = images[0] + if any(first.get(key) for key in ("image_url", "image_b64", "image_path")): + return VisionExtraction( + image_type="患处照片", + clarity="清晰", + batch_no=first.get("batch_no_masked") or first.get("batch_no") or "", + visible_symptoms=["局部泛红"], + extra_fields={}, + ) + return VisionExtraction() + + +def mock_summary( + events: list[dict], + messages: list[dict], + orders: list[dict], + tickets: list[dict], + promises: list[dict], +) -> TrajectorySummary: + """轨迹摘要 Mock：把最近的历史事件标题串成一句，事件 ID 一并保留用于溯源。""" + lines: list[str] = [] + ids: list[str] = [] + + for event in (events or [])[-5:]: + eid = event.get("event_id") or event.get("id") or "" + title = event.get("title") or event.get("event_type") or "" + if title: + lines.append(str(title)) + if eid: + ids.append(str(eid)) + + if not lines: + if promises: + lines.append("存在历史服务承诺") + elif orders: + lines.append("存在关联订单") + else: + lines.append("暂无跨会话历史轨迹") + + return TrajectorySummary(summary="；".join(lines), relevant_event_ids=ids) diff --git a/app/agent/nodes/adverse.py b/app/agent/nodes/adverse.py index 155e158..9e9ac2b 100644 --- a/app/agent/nodes/adverse.py +++ b/app/agent/nodes/adverse.py @@ -41,8 +41,9 @@ def _mentions(state: CustomerState, words: tuple[str, ...]) -> bool: return any(any(word in text for word in words) for text in _texts(state)) -def _has_image(state: CustomerState) -> bool: - return any((message.get("content_type") == "image") for message in (state.get("messages") or [])) +def _image_known(state: CustomerState) -> bool: + """图片是否已可识别（清晰）。没有图片或图片待识别都算缺失。""" + return (state.get("vision") or {}).get("clarity") == "清晰" def _has_batch_no(state: CustomerState) -> bool: @@ -56,7 +57,7 @@ def _has_batch_no(state: CustomerState) -> bool: ("症状出现时间", lambda state: _mentions(state, _OCCURRENCE_WORDS)), ("是否停止使用", lambda state: _mentions(state, _STOPPED_WORDS)), ("是否就医", lambda state: bool(state["risk"].get("medical_visit"))), - ("图片或门诊资料", _has_image), + ("图片或门诊资料", _image_known), ) @@ -143,12 +144,15 @@ def adverse_node(state: CustomerState) -> dict: missing = _missing_fields(state) actions = _build_actions(grade) symptom_summary = _symptom_summary(state) + vision = state.get("vision") or {} assessment = AdverseAssessment( grade=grade, symptom_summary=symptom_summary, medical_visit=bool(state["risk"].get("medical_visit")), stopped_use=None, # 未知就留空，由客服确认，不猜 + image_clarity=vision.get("clarity"), + image_type=vision.get("image_type"), missing_fields=missing, ticket_draft=None if grade == "L1" else _build_ticket_draft(state, grade), safe_reply=_safe_reply(grade, symptom_summary, product_name), diff --git a/app/agent/nodes/reply.py b/app/agent/nodes/reply.py index 7bca22c..4c7486f 100644 --- a/app/agent/nodes/reply.py +++ b/app/agent/nodes/reply.py @@ -31,6 +31,7 @@ def reply_node(state: CustomerState) -> dict: evidence=_evidence_lines(state) or "暂无系统可核验的事实", suggested_actions=state.get("suggested_actions") or [], safe_reply=safe_reply, + timeline_summary=state.get("timeline_summary") or "无历史轨迹", format_instructions=format_instructions(ReplyDraft), ) diff --git a/app/agent/nodes/summary.py b/app/agent/nodes/summary.py new file mode 100644 index 0000000..b570cb4 --- /dev/null +++ b/app/agent/nodes/summary.py @@ -0,0 +1,40 @@ +"""跨会话轨迹摘要节点（AG-05）。 + +消费 CopilotRequest.events（后端按消费者预组装的历史事件）产出 +timeline_summary，供右侧「全轨迹」与回复草稿引用历史承诺 +/ 进度。 +""" + +from app.agent.llm import format_instructions, run_structured +from app.agent.mock import mock_summary +from app.agent.schemas.analysis import TrajectorySummary +from app.agent.state import CustomerState +from app.prompt.summary import TRAJECTORY_SUMMARY_PROMPT + + +def summary_node(state: CustomerState) -> dict: + prompt = TRAJECTORY_SUMMARY_PROMPT.format( + message=state.get("current_message", ""), + risk_level=state.get("risk_level", "L0"), + events=state.get("events") or [], + orders=state.get("orders") or [], + tickets=state.get("tickets") or [], + promises=state.get("promises") or [], + format_instructions=format_instructions(TrajectorySummary), + ) + + result, degraded = run_structured( + prompt=prompt, + schema=TrajectorySummary, + route="reasoning", + mode=state.get("mode", "auto"), + fallback=lambda: mock_summary( + state.get("events") or [], + state.get("messages") or [], + state.get("orders") or [], + state.get("tickets") or [], + state.get("promises") or [], + ), + ) + + return {"timeline_summary": result.summary, "degraded": degraded} diff --git a/app/agent/nodes/vision.py b/app/agent/nodes/vision.py new file mode 100644 index 0000000..904ae53 --- /dev/null +++ b/app/agent/nodes/vision.py @@ -0,0 +1,104 @@ +"""图片识别节点（AG-10）。 + +只在消息里存在图片时生效： +- 有真实图片数据（image_url / image_b64 / image_path）且模型可用 → 调 Vision +- 有图片但无数据、或模型不可用 → Mock 占位（明确「待识别」，交人工确认） +- 无图片 → 直接返回 {}，不产生任何调用 + +""" + +import base64 +import time + +from langchain_core.messages import HumanMessage + +from app.agent import cost +from app.agent.llm import format_instructions +from app.agent.mock import mock_vision +from app.agent.schemas.analysis import VisionExtraction +from app.agent.schemas.contract import EvidenceRef +from app.agent.state import CustomerState +from app.integrations.model.gateway import ModelUnavailableError, gateway +from app.prompt.vision import VISION_PROMPT + + +def _collect_images(state: CustomerState) -> list[dict]: + return [m for m in (state.get("messages") or []) if m.get("content_type") == "image"] + + +def _image_part(message: dict) -> dict | None: + """把消息里的图片数据转成 langchain 多模态 content 片段。""" + if message.get("image_url"): + return {"type": "image_url", "image_url": {"url": message["image_url"]}} + if message.get("image_b64"): + mime = message.get("image_mime") or "image/png" + return {"type": "image_url", "image_url": {"url": f"data:{mime};base64,{message['image_b64']}"}} + if message.get("image_path"): + try: + with open(message["image_path"], "rb") as fh: + b64 = base64.b64encode(fh.read()).decode() + return {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}} + except OSError: + return None + return None + + +def _run_vision(images: list[dict], prompt: str, mode: str) -> tuple[VisionExtraction, bool]: + model = gateway.model_name_for("vision") + parts = [part for part in (_image_part(m) for m in images) if part] + + if mode == "mock" or not parts: + cost.record_call(route="vision", model=model, mode=mode, latency_ms=0, degraded=True) + return mock_vision(images), True + + if not gateway.is_available("vision"): + if mode == "real": + raise ModelUnavailableError("mode=real 但 vision 路由未配置模型，无法执行") + cost.record_call(route="vision", model=model, mode=mode, latency_ms=0, error="route unavailable", degraded=True) + return mock_vision(images), True + + token_capture = cost.TokenCapture() + started = time.monotonic() + try: + llm = gateway.get("vision").bind(response_format={"type": "json_object"}) + message = HumanMessage(content=[{"type": "text", "text": prompt}, *parts]) + response = llm.invoke([message], config={"callbacks": [token_capture]}) + content = response.content if isinstance(response.content, str) else str(response.content) + result = VisionExtraction.model_validate_json(content) + latency_ms = int((time.monotonic() - started) * 1000) + cost.record_call( + route="vision", + model=model, + mode=mode, + latency_ms=latency_ms, + input_tokens=token_capture.input_tokens, + output_tokens=token_capture.output_tokens, + total_tokens=token_capture.total_tokens, + ) + return result, False + except Exception as exc: + latency_ms = int((time.monotonic() - started) * 1000) + if mode == "real": + raise + cost.record_call(route="vision", model=model, mode=mode, latency_ms=latency_ms, error=str(exc), degraded=True) + return mock_vision(images), True + + +def vision_node(state: CustomerState) -> dict: + images = _collect_images(state) + if not images: + return {} + + prompt = VISION_PROMPT.format(format_instructions=format_instructions(VisionExtraction)) + result, degraded = _run_vision(images, prompt, state.get("mode", "auto")) + + first = images[0] + quote = f"图片识别：{result.clarity} / {result.image_type}" if result.clarity != "待识别" else "图片待人工确认" + evidence = EvidenceRef( + source_type="image", + source_id=str(first.get("message_id") or "image"), + message_id=first.get("message_id"), + quote=quote, + ) + + return {"vision": result.model_dump(), "evidence": [evidence], "degraded": degraded} diff --git a/app/agent/run.py b/app/agent/run.py index 7250fab..a1a0a4c 100644 --- a/app/agent/run.py +++ b/app/agent/run.py @@ -17,6 +17,7 @@ import sys from typing import Any +from app.agent import cost from app.agent.graph import graph from app.agent.nodes.promise import extract_promise_candidate from app.agent.schemas.contract import ( @@ -56,6 +57,20 @@ } ], tickets=[], + events=[ + { + "event_id": "E-DEMO-001", + "event_type": "promise", + "occurred_at": "2026-09-10T18:00:00+08:00", + "title": "客服承诺 3 个工作日内回复退款结果", + }, + { + "event_id": "E-DEMO-002", + "event_type": "message", + "occurred_at": "2026-09-13T09:00:00+08:00", + "title": "消费者再次进线催办退款", + }, + ], ) }, "物流": { @@ -110,6 +125,20 @@ } ], tickets=[], + events=[ + { + "event_id": "E-DEMO-101", + "event_type": "order", + "occurred_at": "2026-09-01T10:00:00+08:00", + "title": "消费者下单舒缓修护面霜", + }, + { + "event_id": "E-DEMO-102", + "event_type": "message", + "occurred_at": "2026-10-03T09:00:00+08:00", + "title": "消费者反馈使用后脸红疼痛", + }, + ], ) }, } @@ -129,6 +158,7 @@ def _make_state(request: CopilotRequest) -> CustomerState: "orders": request.orders, "tickets": request.tickets, "promises": request.promises, + "events": request.events, "mode": request.mode, # 理解层 "intent": {}, @@ -153,9 +183,11 @@ def _make_state(request: CopilotRequest) -> CustomerState: "missing_fields": [], "suggested_actions": [], "adverse": None, + "vision": None, # 表达层 "reply_draft": "", "fact_check": {"status": "pass", "unverified_claims": [], "blocked": False}, + "timeline_summary": "", # 可观测性 "degraded": False, "model_route": "mock", @@ -164,7 +196,12 @@ def _make_state(request: CopilotRequest) -> CustomerState: def run_copilot(request: CopilotRequest) -> CopilotResult: """跑一遍副驾链路。不改变任何业务状态。""" - final = graph.invoke(_make_state(request)) + cost.set_session(request.session_id) + try: + final = graph.invoke(_make_state(request)) + finally: + cost.set_session(None) + intent = final.get("intent") or {} adverse = final.get("adverse") @@ -181,6 +218,7 @@ def run_copilot(request: CopilotRequest) -> CopilotResult: missing_fields=final.get("missing_fields") or [], suggested_actions=final.get("suggested_actions") or [], evidence=final.get("evidence") or [], + timeline_summary=final.get("timeline_summary") or None, model_route=final.get("model_route", "mock"), degraded=bool(final.get("degraded")), ), @@ -192,7 +230,11 @@ def run_copilot(request: CopilotRequest) -> CopilotResult: def extract_promises(req: PromiseExtractRequest) -> PromiseExtractResult: """抽取客服消息里的服务承诺候选。只给候选，建单由人工确认。""" - return extract_promise_candidate(req) + cost.set_session(req.session_id) + try: + return extract_promise_candidate(req) + finally: + cost.set_session(None) def _print_case(name: str, result: CopilotResult) -> None: diff --git a/app/agent/schemas/analysis.py b/app/agent/schemas/analysis.py index 3416777..21cbe4e 100644 --- a/app/agent/schemas/analysis.py +++ b/app/agent/schemas/analysis.py @@ -45,3 +45,16 @@ class PromiseExtraction(BaseModel): due_expression: str = Field(default="", description="时间的原始表达，如 '明天上午'、'3个工作日'；无则留空") owner_type: str = Field(default="agent", description="store / agent / team / system") confidence: float = Field(default=0.0, description="置信度 0 到 1") + + +class VisionExtraction(BaseModel): + image_type: str = Field(default="未知", description="图片类型：患处照片 / 门诊资料 / 产品批次图 / 快递面单 / 其他") + clarity: str = Field(default="待识别", description="清晰度：清晰 / 模糊 / 无法识别 / 待识别") + batch_no: str = Field(default="", description="图中可见的批次号，无则空") + visible_symptoms: list[str] = Field(default_factory=list, description="图中可见现象描述（仅描述，不诊断）") + extra_fields: dict[str, str] = Field(default_factory=dict, description="门诊资料/面单上可读的字段") + + +class TrajectorySummary(BaseModel): + summary: str = Field(description="跨会话轨迹摘要，只保留与当前任务相关的事件") + relevant_event_ids: list[str] = Field(default_factory=list, description="摘要引用的历史事件 ID，用于溯源") diff --git a/app/agent/schemas/contract.py b/app/agent/schemas/contract.py index a36d5f4..03a27f4 100644 --- a/app/agent/schemas/contract.py +++ b/app/agent/schemas/contract.py @@ -16,7 +16,7 @@ AdverseGrade = Literal["L1", "L2", "L3"] # L0 不进不良反应专项 ModelRoute = Literal["fast", "reasoning", "vision", "mock"] AgentMode = Literal["auto", "real", "mock"] -SourceType = Literal["chat", "order", "ticket", "rule", "action"] +SourceType = Literal["chat", "order", "ticket", "rule", "action", "image"] PromiseType = Literal["refund", "follow_up", "replenishment", "logistics", "other"] OwnerType = Literal["store", "agent", "team", "system"] @@ -40,6 +40,7 @@ class CopilotRequest(BaseModel): orders: list[dict[str, Any]] = Field(default_factory=list) tickets: list[dict[str, Any]] = Field(default_factory=list) promises: list[dict[str, Any]] = Field(default_factory=list) + events: list[dict[str, Any]] = Field(default_factory=list) mode: AgentMode = "auto" @@ -56,6 +57,7 @@ class CopilotInsight(BaseModel): missing_fields: list[str] = Field(default_factory=list) suggested_actions: list[str] = Field(default_factory=list) evidence: list[EvidenceRef] = Field(default_factory=list) + timeline_summary: str | None = None model_route: ModelRoute = "mock" degraded: bool = False @@ -71,6 +73,8 @@ class AdverseAssessment(BaseModel): symptom_summary: str = "" medical_visit: bool = False stopped_use: bool | None = None + image_clarity: str | None = None + image_type: str | None = None missing_fields: list[str] = Field(default_factory=list) ticket_draft: dict[str, Any] | None = None # CREATE_TICKET 的 draft_payload safe_reply: str | None = None diff --git a/app/agent/state.py b/app/agent/state.py index 5688e44..6b34b0d 100644 --- a/app/agent/state.py +++ b/app/agent/state.py @@ -33,6 +33,7 @@ class CustomerState(TypedDict): orders: list[dict] tickets: list[dict] promises: list[dict] + events: list[dict] mode: str # auto / real / mock # 理解层 @@ -56,10 +57,12 @@ class CustomerState(TypedDict): missing_fields: list[str] suggested_actions: list[str] adverse: dict | None + vision: dict | None # 表达层 reply_draft: str fact_check: FactCheck + timeline_summary: str # 可观测性：任一节点降级则整体标记降级，所以用 or_ 而不是覆盖 degraded: Annotated[bool, or_] diff --git a/app/integrations/model/gateway.py b/app/integrations/model/gateway.py index eb41f1c..ff50ef3 100644 --- a/app/integrations/model/gateway.py +++ b/app/integrations/model/gateway.py @@ -18,10 +18,13 @@ # 模型路由 -> (模型名环境变量, API Key 环境变量, Base URL 环境变量) ROUTE_ENV: dict[str, tuple[str, str, str]] = { "fast": ("MODEL_FAST", "DASHSCOPE_API_KEY", "DASHSCOPE_BASE_URL"), - "reasoning": ("MODEL_DS", "DEEPSEEK_API_KEY", "DEEPSEEK_BASE_URL"), + "reasoning": ("MODEL_REASONING", "DEEPSEEK_API_KEY", "DEEPSEEK_BASE_URL"), "vision": ("MODEL_VISION", "DASHSCOPE_API_KEY", "DASHSCOPE_BASE_URL"), } +# 兼容旧命名：早期用 MODEL_DS 指代推理模型 +_LEGACY_MODEL_ENV: dict[str, str] = {"reasoning": "MODEL_DS"} + DEFAULT_ROUTE = "fast" @@ -39,10 +42,22 @@ def __init__(self, temperature: float = 0.5) -> None: def _env_names(self, route: str) -> tuple[str, str, str]: return ROUTE_ENV.get(route, ROUTE_ENV[DEFAULT_ROUTE]) + def _resolve_model(self, route: str) -> str: + """返回路由实际使用的模型名；主名未配置时回退到旧命名。""" + model_env, _, _ = self._env_names(route) + value = os.getenv(model_env, "") + if not value and route in _LEGACY_MODEL_ENV: + value = os.getenv(_LEGACY_MODEL_ENV[route], "") + return value or "" + + def model_name_for(self, route: str = DEFAULT_ROUTE) -> str: + """只读模型名，供成本日志等观测使用，不抛异常。""" + return self._resolve_model(route) + def is_available(self, route: str = DEFAULT_ROUTE) -> bool: """模型名和 API Key 都配置了才算可用。""" - model_env, key_env, _ = self._env_names(route) - return bool(os.getenv(model_env)) and bool(os.getenv(key_env)) + _, key_env, _ = self._env_names(route) + return bool(self._resolve_model(route)) and bool(os.getenv(key_env)) def get(self, route: str = DEFAULT_ROUTE) -> ChatOpenAI: if not self.is_available(route): @@ -50,9 +65,9 @@ def get(self, route: str = DEFAULT_ROUTE) -> ChatOpenAI: raise ModelUnavailableError(f"模型路由 {route} 不可用：请在 .env 中同时配置 {model_env} 和 {key_env}") if route not in self._clients: - model_env, key_env, base_url_env = self._env_names(route) + _, key_env, base_url_env = self._env_names(route) self._clients[route] = ChatOpenAI( - model=os.getenv(model_env, ""), + model=self._resolve_model(route), api_key=SecretStr(os.getenv(key_env, "")), base_url=os.getenv(base_url_env) or None, temperature=self.temperature, diff --git a/app/prompt/reply.py b/app/prompt/reply.py index 23865da..a89533c 100644 --- a/app/prompt/reply.py +++ b/app/prompt/reply.py @@ -28,6 +28,9 @@ 处理建议： {suggested_actions} +历史轨迹摘要： +{timeline_summary} + 要求： 1. 回复必须基于提供的业务证据。 2. 不得编造订单状态、退款时间等不存在的信息。 @@ -61,6 +64,9 @@ 处理建议： {suggested_actions} +历史轨迹摘要： +{timeline_summary} + 处置基准话术（必须以此为骨架，不得改变其中的就医建议和升级承诺）： {safe_reply} diff --git a/app/prompt/summary.py b/app/prompt/summary.py new file mode 100644 index 0000000..f977f53 --- /dev/null +++ b/app/prompt/summary.py @@ -0,0 +1,35 @@ +"""跨会话轨迹摘要 Prompt（AG-05）。 + +只保留与当前任务相关的事件，不得编造事件里没有的金额、状态、时间。 +""" + +TRAJECTORY_SUMMARY_PROMPT = """你是美妆电商客服轨迹摘要助手。请把消费者跨会话、跨订单、跨工单的历史轨迹， +压缩成一句与当前任务相关的摘要。 + +当前用户消息： +{message} + +当前风险等级： +{risk_level} + +历史事件（按时间顺序）： +{events} + +关联订单： +{orders} + +关联工单： +{tickets} + +历史承诺： +{promises} + +要求： +1. 只保留与当前任务（退款 / 不良反应 / 物流等）相关的事件，无关事件丢弃。 +2. 摘要要能体现「之前发生了什么、当前进展到哪一步」，例如历史承诺是否已履约。 +3. 不得编造事件里没有的金额、状态、时间。 +4. relevant_event_ids 只填你在摘要里真正引用到的历史事件 ID。 + +请严格按照 json 格式输出。 + +{format_instructions}""" diff --git a/app/prompt/vision.py b/app/prompt/vision.py new file mode 100644 index 0000000..b5ba3f5 --- /dev/null +++ b/app/prompt/vision.py @@ -0,0 +1,19 @@ +"""图片识别 Prompt（AG-10）。 + +只描述图中确实可见的内容，不做诊断、不推断病因、不给用药建议。 +""" + +VISION_PROMPT = """你是美妆电商客服图片识别助手。请识别消费者上传的图片。 + +要求： +1. image_type：图片类型，从 患处照片 / 门诊资料 / 产品批次图 / 快递面单 / 其他 里选一个。 +2. clarity：清晰度，从 清晰 / 模糊 / 无法识别 里选一个。 +3. batch_no：图中可见的批次号，没有就留空。 +4. visible_symptoms：图中可见的现象（如 泛红、肿胀、脱皮），只描述现象，不做诊断。 +5. extra_fields：门诊资料或面单上可读的字段（如 日期、医院名、运单号）。 + +只描述图中确实可见的内容，不要推断，不要补充图中没有的信息。 + +请严格按照 json 格式输出（不要输出 json 以外的任何内容）。 + +{format_instructions}""" diff --git a/data/annotations.json b/data/annotations.json new file mode 100644 index 0000000..952278f --- /dev/null +++ b/data/annotations.json @@ -0,0 +1,66 @@ +{ + "version": 1, + "schema": { + "切分单位": "session_id：同一会话不会同时出现在 train 与 test", + "messages[].expected": "意图一级/二级、情绪、风险等级的人工金标准", + "promises[].expected": "承诺抽取的人工金标准" + }, + "sessions": { + "S-DEMO-001": { + "messages": [ + { + "message_id": "M-001", + "message_text": "我上周买的那瓶精华，退款怎么还没到账？都三天了", + "expected": { + "intent_primary": "订单服务", + "intent_secondary": "退款迟迟不到账", + "emotion": "焦虑", + "risk_level": "L0" + } + } + ] + }, + "S-DEMO-002": { + "messages": [ + { + "message_id": "M-101", + "message_text": "我的快递三天没动静了，到底什么时候能到？", + "expected": { + "intent_primary": "物流问题", + "intent_secondary": "物流停滞未更新", + "emotion": "焦虑", + "risk_level": "L0" + } + } + ] + }, + "S-DEMO-003": { + "messages": [ + { + "message_id": "M-201", + "message_text": "用了你们的面霜脸特别红，很疼，已经去医院了", + "expected": { + "intent_primary": "不良反应", + "intent_secondary": "过敏就医", + "emotion": "焦虑", + "risk_level": "L3" + } + } + ] + } + }, + "promises": [ + { + "message_text": "明天上午给您安排退款到账", + "expected": {"has_promise": true, "promise_type": "refund"} + }, + { + "message_text": "我们会尽快给您回访", + "expected": {"has_promise": true, "promise_type": "follow_up", "needs_confirmation": true} + }, + { + "message_text": "好的，感谢您的反馈", + "expected": {"has_promise": false, "promise_type": "other"} + } + ] +} diff --git a/scripts/eval_agent.py b/scripts/eval_agent.py new file mode 100644 index 0000000..f4e5338 --- /dev/null +++ b/scripts/eval_agent.py @@ -0,0 +1,144 @@ +"""Agent 离线评测脚本（EVA-01 / EVA-02）。 + +用法： + python scripts/eval_agent.py # 全量评测，mode=auto（无 Key 自动降级 Mock） + python scripts/eval_agent.py --mock # 强制 Mock，结果可复现 + python scripts/eval_agent.py --test-ratio 0.33 --seed 42 # 会话级切分，只在测试集上出指标 + +EVA-01 会话级切分：split_by_session 保证同一会话只落在 train 或 test 一侧，避免泄漏。 +EVA-02 指标：意图一级/二级、情绪、风险等级用准确率；承诺抽取用 Precision / Recall / F1。 +""" + +import argparse +import json +import random +import sys +from pathlib import Path +from typing import Any + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from app.agent.run import extract_promises, run_copilot +from app.agent.schemas.contract import AgentMode, CopilotRequest, PromiseExtractRequest + +DEFAULT_ANNOTATIONS = Path(__file__).resolve().parents[1] / "data" / "annotations.json" + + +def load_annotations(path: Path) -> dict[str, Any]: + with path.open(encoding="utf-8") as fh: + return json.load(fh) + + +def flatten(annotations: dict[str, Any]) -> list[dict[str, Any]]: + samples: list[dict[str, Any]] = [] + for session_id, session in annotations["sessions"].items(): + for msg in session["messages"]: + samples.append({"session_id": session_id, **msg}) + return samples + + +def split_by_session( + samples: list[dict[str, Any]], test_ratio: float, seed: int +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + """会话级切分：同一会话不会同时出现在 train 和 test 里，避免数据泄漏。""" + session_ids = sorted({s["session_id"] for s in samples}) + rng = random.Random(seed) + rng.shuffle(session_ids) + n_test = max(1, int(len(session_ids) * test_ratio)) if test_ratio > 0 else 0 + test_sessions = set(session_ids[:n_test]) + train = [s for s in samples if s["session_id"] not in test_sessions] + test = [s for s in samples if s["session_id"] in test_sessions] + return train, test + + +def predict_copilot(sample: dict[str, Any], mode: AgentMode) -> dict[str, Any]: + request = CopilotRequest(session_id=sample["session_id"], current_message=sample["message_text"], mode=mode) + result = run_copilot(request) + return { + "intent_primary": result.insight.intent_primary, + "intent_secondary": result.insight.intent_secondary, + "emotion": result.insight.emotion, + "risk_level": result.insight.risk_level, + } + + +def accuracy(preds: list[Any], golds: list[Any]) -> float: + if not golds: + return 0.0 + return sum(1 for p, g in zip(preds, golds, strict=False) if p == g) / len(golds) + + +def binary_metrics(preds: list[bool], golds: list[bool]) -> dict[str, float]: + tp = sum(1 for p, g in zip(preds, golds, strict=False) if p and g) + fp = sum(1 for p, g in zip(preds, golds, strict=False) if p and not g) + fn = sum(1 for p, g in zip(preds, golds, strict=False) if not p and g) + precision = tp / (tp + fp) if (tp + fp) else 0.0 + recall = tp / (tp + fn) if (tp + fn) else 0.0 + f1 = 2 * precision * recall / (precision + recall) if (precision + recall) else 0.0 + return {"precision": precision, "recall": recall, "f1": f1} + + +def evaluate_copilot(samples: list[dict[str, Any]], mode: AgentMode) -> dict[str, Any]: + preds = [predict_copilot(s, mode) for s in samples] + golds = [s["expected"] for s in samples] + return { + "intent_primary_acc": accuracy([p["intent_primary"] for p in preds], [g["intent_primary"] for g in golds]), + "intent_secondary_acc": accuracy( + [p["intent_secondary"] for p in preds], [g["intent_secondary"] for g in golds] + ), + "emotion_acc": accuracy([p["emotion"] for p in preds], [g["emotion"] for g in golds]), + "risk_level_acc": accuracy([p["risk_level"] for p in preds], [g["risk_level"] for g in golds]), + "n": len(samples), + } + + +def evaluate_promises(annotations: dict[str, Any], mode: AgentMode) -> dict[str, Any]: + preds_has: list[bool] = [] + golds_has: list[bool] = [] + preds_type: list[str] = [] + golds_type: list[str] = [] + + for item in annotations.get("promises", []): + result = extract_promises( + PromiseExtractRequest(session_id="eval", message_id="m", message_text=item["message_text"], mode=mode) + ) + cand = result.candidate + pred_has = cand is not None and cand.promise_type != "other" + preds_has.append(pred_has) + golds_has.append(bool(item["expected"].get("has_promise"))) + preds_type.append(cand.promise_type if cand else "other") + golds_type.append(item["expected"].get("promise_type", "other")) + + has_metrics = binary_metrics(preds_has, golds_has) + return { + "has_promise_precision": has_metrics["precision"], + "has_promise_recall": has_metrics["recall"], + "has_promise_f1": has_metrics["f1"], + "promise_type_acc": accuracy(preds_type, golds_type), + "n": len(golds_has), + } + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Agent 离线评测") + parser.add_argument("--mock", action="store_true", help="强制 Mock，结果可复现") + parser.add_argument("--test-ratio", type=float, default=0.0, help="测试集比例（>0 时按会话切分）") + parser.add_argument("--seed", type=int, default=42) + parser.add_argument("--annotations", default=str(DEFAULT_ANNOTATIONS)) + args = parser.parse_args(argv) + + mode: AgentMode = "mock" if args.mock else "auto" + annotations = load_annotations(Path(args.annotations)) + samples = flatten(annotations) + + target = split_by_session(samples, args.test_ratio, args.seed)[1] if args.test_ratio > 0 else samples + + print(f"== Agent 评测（mode={mode}, n={len(target)}）==") + print(json.dumps(evaluate_copilot(target, mode), ensure_ascii=False, indent=2)) + print("== 承诺抽取 ==") + print(json.dumps(evaluate_promises(annotations, mode), ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_agent.py b/tests/test_agent.py index 23d20f4..2da265d 100644 --- a/tests/test_agent.py +++ b/tests/test_agent.py @@ -74,7 +74,7 @@ def test_evidence_is_traceable(): assert result.insight.evidence for item in result.insight.evidence: assert item.source_id - assert item.source_type in {"chat", "order", "ticket", "rule", "action"} + assert item.source_type in {"chat", "order", "ticket", "rule", "action", "image"} def test_mock_mode_marks_degraded(): diff --git a/tests/test_agent_extras.py b/tests/test_agent_extras.py new file mode 100644 index 0000000..bf948bc --- /dev/null +++ b/tests/test_agent_extras.py @@ -0,0 +1,82 @@ +"""补齐后的 Agent 新能力测试：否定语、图片识别、轨迹摘要、成本日志、会话级切分。""" + +from app.agent import cost +from app.agent.mock import mock_risk, mock_vision +from app.agent.run import CASES, run_copilot + + +def test_negation_cancels_symptom(): + assert mock_risk("没有红肿，也没有瘙痒").adverse_reaction is False + assert mock_risk("我一点也不疼").adverse_reaction is False + + +def test_positive_symptom_still_hits(): + assert mock_risk("用了之后脸很红很疼").adverse_reaction is True + + +def test_negation_cancels_medical_visit(): + assert mock_risk("我还没去医院").medical_visit is False + assert mock_risk("已经去过医院了").medical_visit is True + + +def test_mock_vision_placeholder_pending(): + result = mock_vision([{"message_text": "[图片]", "content_type": "image"}]) + assert result.clarity == "待识别" + assert result.image_type == "未知" + + +def test_mock_vision_with_data_is_clear(): + result = mock_vision( + [{"message_text": "[图片]", "content_type": "image", "image_b64": "x", "batch_no_masked": "B2409**"}] + ) + assert result.clarity == "清晰" + assert result.image_type == "患处照片" + + +def test_run_copilot_produces_timeline_summary(): + result = run_copilot(CASES["退款"]["request"].model_copy(update={"mode": "mock"})) + assert result.insight.timeline_summary + assert "退款" in result.insight.timeline_summary or "承诺" in result.insight.timeline_summary + + +def test_cost_logs_recorded_in_mock_mode(): + cost.clear_call_logs() + run_copilot(CASES["退款"]["request"].model_copy(update={"mode": "mock"})) + logs = cost.get_call_logs() + assert logs + assert all(r.degraded for r in logs) + assert {r.route for r in logs} >= {"fast", "reasoning"} + # 退款 L0 无图片，不应有 vision 调用 + assert all(r.route != "vision" for r in logs) + + +def test_cost_session_totals(): + cost.clear_call_logs() + run_copilot(CASES["退款"]["request"].model_copy(update={"mode": "mock"})) + totals = cost.session_totals("S-DEMO-001") + assert totals["calls"] > 0 + assert totals["total_tokens"] == 0 # mock 无 token + + +def test_adverse_case_runs_vision_and_marks_pending(): + result = run_copilot(CASES["不良反应"]["request"].model_copy(update={"mode": "mock"})) + assert result.adverse is not None + assert result.adverse.image_clarity == "待识别" + assert "图片或门诊资料" in result.adverse.missing_fields + assert any(item.source_type == "image" for item in result.insight.evidence) + + +def test_split_by_session_no_leak(): + from scripts.eval_agent import split_by_session + + samples = [ + {"session_id": "A", "message_text": "x"}, + {"session_id": "A", "message_text": "y"}, + {"session_id": "B", "message_text": "z"}, + {"session_id": "C", "message_text": "w"}, + ] + train, test = split_by_session(samples, test_ratio=0.34, seed=42) + train_sessions = {s["session_id"] for s in train} + test_sessions = {s["session_id"] for s in test} + assert train_sessions.isdisjoint(test_sessions) + assert test