Eric Wang
e72d9c9bc0
- merge main (b0ff2e0..c7ca781): workflow/20261022 moves dynamic context out of fixed prompts - collection/extraction prompts carry no inline vars; reference time and current form travel in the user message <turn_context> block; photo/confirm data kept at prompt tails - prompts: all 8 files regenerated from 20261022 (byte-identity tests point at the new export) - collect: user message = <turn_context>(本轮参考时间) + <user_input>; extraction message adds 当前表单; single trusted clock per turn threads through reply/extraction/repair - photo: raw input only (20261022 photo nodes use system_text, no wrapper) - workflow fake llm: unwraps <user_input> before event parsing - fields: accident_time normalized to extraction format (YYYY年M月D日H点m分, strip zero padding) - fixes form assertion mismatch with padded output - evals/run.py: trust_env=False - bypass Windows system proxy (registry proxy at 192.168.0.4:7897 was intercepting localhost and returning 503) - verified: 172 unit tests; ALL golden suites pass on real model (46/46): collection_skip 12/12, photo_skip 4/4, no_false 2/2, button_guidance 12/12, non_motor_vehicle 14/14, time_confirmation 2/2
297 lines
12 KiB
Python
297 lines
12 KiB
Python
"""采集节点 LLM 接入（作业 4）。
|
||
|
||
职责边界：
|
||
- LLM 一次调用同时产出 ``<state>XXXX</state>正文``——
|
||
不得为复判状态再调用第二次（计划作业 4-3）；
|
||
- 代码校验前缀合法性、阶段枚举与白名单守卫，
|
||
非法提议不写状态（受控修复重试：前缀/非法提议最多各修复一次）；
|
||
- 字段候选经 fields.validate_candidate 校验后进入
|
||
accident_fields / confirm_status / turn_patch（三态语义）；
|
||
- 澄清/无回复计数是代码侧簿记，只作为 LLM 判断的输入，
|
||
转不转人工由模型按提示词决定（代码不代替模型决定）。
|
||
|
||
失败语义（计划作业 4-10）：前缀缺失/非法且修复失败时，抛出
|
||
``CollectReplyError``——不提交业务状态、不泄漏原始模型内容；
|
||
调用方（API 层）将其转换为可识别错误响应。
|
||
"""
|
||
|
||
from collections.abc import Callable
|
||
from datetime import datetime
|
||
import json
|
||
import re
|
||
|
||
from .fields import (
|
||
FORM_FIELDS,
|
||
validate_form_update,
|
||
)
|
||
from .llm import LLMService
|
||
from .prompts import COLLECTION, EXTRACTION, render_prompt
|
||
from .state import AgentState
|
||
from .time_context import observation_time, time_evidence, valid_time_reply
|
||
from .transitions import COLLECT_STAGE, is_legal_transition
|
||
|
||
STATE_PREFIX_RE = re.compile(r"^\s*<state>\s*(\d{4})\s*</state>")
|
||
_STATE_TAG_RE = re.compile(r"</?state>")
|
||
|
||
# 采集阶段澄清阈值（来源：prompts/collection.md）
|
||
COLLECT_FAILURE_LIMIT = 5
|
||
|
||
# 受控修复指令（计划作业 4-10）。追加在系统提示词之后，
|
||
# 明确要求输出格式，供模型自我修正。
|
||
_REPAIR_SUFFIX = (
|
||
"\n\n【输出修复】你上一次的输出不合规：必须且只能以"
|
||
" <state>四位状态码</state> 开头（如 <state>1002</state>），"
|
||
"正文部分不得再出现 <state> 或 </state> 标签。"
|
||
"请重新输出一条完全合规的回复。"
|
||
)
|
||
|
||
|
||
class CollectReplyError(Exception):
|
||
"""采集回复不可用：格式、转换或时间依据校验且修复失败。
|
||
|
||
不携带原始模型内容，避免向上层泄漏。
|
||
"""
|
||
|
||
|
||
def parse_state_reply(text: str) -> tuple[str, str]:
|
||
"""解析 ``<state>XXXX</state>`` 前缀，返回 (stage_code, 正文)。
|
||
|
||
严格模式：无有效前缀、或正文再出现 state 标签（泄漏），
|
||
都视为不合规并抛 ValueError，交由 run_collect 决定是否修复。
|
||
"""
|
||
match = STATE_PREFIX_RE.match(text)
|
||
if not match:
|
||
raise ValueError("回复缺少有效的 <state> 前缀")
|
||
body = text[match.end():].strip()
|
||
if _STATE_TAG_RE.search(body):
|
||
raise ValueError("正文中再次出现 state 标签")
|
||
return match.group(1), body
|
||
|
||
|
||
def _try_parse(text: str) -> tuple[str, str] | None:
|
||
"""宽松入口：不合规返回 None 而不是抛错（供修复流程判断）。"""
|
||
try:
|
||
return parse_state_reply(text)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
async def validated_reply(
|
||
llm: LLMService,
|
||
system_prompt: str,
|
||
user_message: str,
|
||
current: str,
|
||
*,
|
||
validator: Callable[[str, str], bool] = is_legal_transition,
|
||
history: list[dict[str, str]] | None = None,
|
||
reply_validator: Callable[[str, str], bool] | None = None,
|
||
) -> tuple[str, str]:
|
||
"""模型同时产出下一阶段和正文；格式与转换共用一次修复预算。
|
||
|
||
格式、阶段转换与可选的回复依据校验共用一次修复预算。不按输入长度
|
||
或关键字替模型选择动作。非法输出不写回图状态；修复仍失败则抛出
|
||
不含模型原文的错误。
|
||
"""
|
||
for attempt in range(2):
|
||
prompt = system_prompt
|
||
if attempt:
|
||
prompt += (
|
||
_REPAIR_SUFFIX
|
||
+ f"\n当前阶段为 {current}。请重新检查该阶段允许的转换"
|
||
"和当前 clientMode，不得跳步或切换到另一模式的出口。"
|
||
)
|
||
raw = await llm.complete(prompt, user_message, history=history)
|
||
parsed = _try_parse(raw)
|
||
if (parsed is not None and validator(current, parsed[0])
|
||
and (reply_validator is None or reply_validator(parsed[0], parsed[1]))):
|
||
return raw, parsed[0]
|
||
raise CollectReplyError("模型回复格式、阶段转换或时间依据不合规（一次修复后仍失败）")
|
||
|
||
|
||
# 工作流 chatNode 的 history 配置为 50；保留独立消息及其角色。
|
||
_HISTORY_MESSAGES = 50
|
||
|
||
|
||
def conversation_history(state: AgentState) -> list[dict[str, str]]:
|
||
"""历史不拼入 user 消息，不剥除历史 assistant 的状态前缀。"""
|
||
return [dict(message) for message in (state.get("dialogue") or [])[-_HISTORY_MESSAGES:]]
|
||
|
||
|
||
def context_prompt(state: AgentState, *, time_context: dict | None = None) -> str:
|
||
"""运行时上下文随 system 提示提供；业务判断仍由模型负责。"""
|
||
return "\n\n【当前会话上下文】\n" + json.dumps({
|
||
"stage_code": state.get("stage_code") or COLLECT_STAGE,
|
||
"client_mode": state.get("client_mode") or "direct",
|
||
"person_index": state.get("person_index") or 1,
|
||
"accident_fields": state.get("accident_fields") or {},
|
||
"person_fields": state.get("person_fields") or {},
|
||
"no_reply_count_before_this_turn": state.get("no_reply_count") or 0,
|
||
**({"time_evidence": time_context} if time_context is not None else {}),
|
||
}, ensure_ascii=False)
|
||
|
||
|
||
def _format_turn_context(moment: datetime, extra: dict | None = None) -> str:
|
||
"""workflow@20261022 的 <turn_context> 块：动态上下文经 user 消息传递。"""
|
||
lines = [f"本轮参考时间：{moment.strftime('%Y-%m-%d %H:%M:%S %z')}"]
|
||
for key, value in (extra or {}).items():
|
||
lines.append(f"{key}：{value}")
|
||
return "<turn_context>\n" + "\n".join(lines) + "\n</turn_context>"
|
||
|
||
|
||
def _build_user_message(state: AgentState, *, now: datetime) -> str:
|
||
"""对齐 workflow@20261022 采集节点 userChatInput：
|
||
<turn_context>（本轮参考时间）+ <user_input>（原始输入）。
|
||
"""
|
||
return (
|
||
f"{_format_turn_context(now)}\n\n"
|
||
f"<user_input>\n{state.get('input_text', '')}\n</user_input>"
|
||
)
|
||
|
||
|
||
|
||
# 采集阶段表单状态使用工作流的外部表单键（has_injury 等，见 fields.FORM_FIELDS）；
|
||
# 内部键（ywrysw 等）属于 get_info 分组，作业 6 再建立映射。
|
||
|
||
# 提取调用的系统提示标记（不以 <state> 开头，不会被误当对话回复解析）
|
||
_EXTRACTION_MARKER = "【字段提取】"
|
||
|
||
|
||
async def _extract_turn_fields(
|
||
state: AgentState, llm: LLMService, *, now: datetime, time_context: dict,
|
||
) -> dict:
|
||
"""独立提取调用（计划作业 4-4：调用计数可审计——每轮在对话调用之外多一次）。
|
||
|
||
提示词为工作流「文本内容提取事故信息」原文（20261022 起时间/表单
|
||
经 content 的 <turn_context> 传递）；模型只输出 JSON；
|
||
校验见 fields.validate_form_update。提取失败不阻塞对话（返回 {}）。
|
||
"""
|
||
system = (
|
||
_EXTRACTION_MARKER
|
||
+ "\n按下列系统提示词执行提取，只输出 JSON 对象，不输出解释。\n\n"
|
||
+ render_prompt(EXTRACTION, state, now=now)
|
||
+ context_prompt(state, time_context=time_context)
|
||
)
|
||
message = (
|
||
f"{_format_turn_context(now, {'当前表单': json.dumps(state.get('accident_fields') or {}, ensure_ascii=False)})}\n\n"
|
||
f"<user_input>\n{state.get('input_text', '')}\n</user_input>"
|
||
)
|
||
try:
|
||
raw = await llm.complete(
|
||
system,
|
||
message,
|
||
history=conversation_history(state),
|
||
)
|
||
data = json.loads(raw.strip())
|
||
except Exception:
|
||
return {}
|
||
if not isinstance(data, dict):
|
||
return {}
|
||
if time_context["historical_reference_missing"]:
|
||
# An unanchored historical relative expression cannot support a date.
|
||
# Keep all other independently supported fields from this turn.
|
||
data.pop("accident_time", None)
|
||
return validate_form_update(data)
|
||
|
||
|
||
def _has_meaningful_value(value) -> bool:
|
||
"""对应「更新表单」的 hasMeaningfulValue：None/空串无效（0 有效）。"""
|
||
return value is not None and value != ""
|
||
|
||
|
||
def _is_same_value(old, new) -> bool:
|
||
"""对应「更新表单」的 isSameValue：数字与数字字符串视为相等。"""
|
||
if old == new:
|
||
return True
|
||
if old is None or new is None:
|
||
return False
|
||
if old == "" or new == "":
|
||
return False
|
||
try:
|
||
return float(old) == float(new)
|
||
except (TypeError, ValueError):
|
||
return str(old).strip() == str(new).strip()
|
||
|
||
|
||
def _merge_form(state: AgentState, update: dict) -> tuple[dict, dict]:
|
||
"""按「更新表单」语义合并：白名单内、有效、有变化的字段进入差量。
|
||
|
||
返回 (合并后的完整表单, 本轮差量 turn_patch)。
|
||
"""
|
||
form = dict(state.get("accident_fields") or {})
|
||
patch: dict = {}
|
||
for key in FORM_FIELDS:
|
||
if key not in update:
|
||
continue
|
||
value = update[key]
|
||
if not _has_meaningful_value(value):
|
||
continue
|
||
if _is_same_value(form.get(key), value):
|
||
continue
|
||
form[key] = value
|
||
patch[key] = value
|
||
return form, patch
|
||
|
||
|
||
def _bookkeeping(state: AgentState) -> tuple[int, int]:
|
||
"""代码侧上下文簿记：无回复计数 + 澄清计数。
|
||
|
||
返回 (no_reply_count, clarify_count)。注意这只是给 LLM 的
|
||
判断输入，是否转人工由模型按提示词决定。
|
||
|
||
澄清失败次数由模型结合语义与历史判断；这里透传已有值，不把
|
||
每次普通输入都当作一次失败。
|
||
"""
|
||
event = state.get("client_event") or ""
|
||
no_reply_count = state.get("no_reply_count") or 0
|
||
if event == "no_reply":
|
||
no_reply_count += 1
|
||
elif (state.get("input_text") or "").strip():
|
||
no_reply_count = 0
|
||
|
||
return no_reply_count, state.get("clarify_count") or 0
|
||
|
||
|
||
async def run_collect(state: AgentState, llm: LLMService, *, now: datetime | None = None) -> dict:
|
||
"""采集节点核心：一轮对话的 LLM 接入（参考实现）。
|
||
|
||
流程：一次模型调用 → 前缀校验（不合规修复一次）→
|
||
白名单守卫（非法提议修复一次）→ 字段候选落库 → 返回状态增量。
|
||
任何一步修复后仍不合规都抛 CollectReplyError：不提交业务状态、
|
||
错误信息不携带原始模型内容。
|
||
"""
|
||
moment = observation_time(now)
|
||
evidence = time_evidence(state, moment)
|
||
system_prompt = render_prompt(COLLECTION, state, now=moment) + context_prompt(state, time_context=evidence)
|
||
user_message = _build_user_message(state, now=moment)
|
||
current = state.get("stage_code") or COLLECT_STAGE
|
||
raw, stage = await validated_reply(
|
||
llm, system_prompt, user_message, current, history=conversation_history(state),
|
||
reply_validator=lambda code, body: valid_time_reply(code, body, evidence),
|
||
)
|
||
|
||
|
||
# 4) 字段提取与表单差量（提取调用在对话调用之外，见 _extract_turn_fields）
|
||
extracted = await _extract_turn_fields(state, llm, now=moment, time_context=evidence)
|
||
accident_fields, patch = _merge_form(state, extracted)
|
||
no_reply_count, clarify_count = _bookkeeping(state)
|
||
|
||
# 5) 状态增量：response_text 保留完整前缀文本，API 层负责解析；
|
||
# dialogue 追加本轮消息，供下一轮历史注入（计划作业 4-5）
|
||
dialogue = list(state.get("dialogue") or [])
|
||
timestamps = dict(state.get("dialogue_timestamps") or {})
|
||
timestamps[str(len(dialogue))] = moment.isoformat()
|
||
dialogue.append({"role": "user", "content": state.get("input_text", "")})
|
||
dialogue.append({"role": "assistant", "content": raw})
|
||
return {
|
||
"turn_count": (state.get("turn_count") or 0) + 1,
|
||
"no_reply_count": no_reply_count,
|
||
"clarify_count": clarify_count,
|
||
"response_text": raw,
|
||
"stage_code": stage,
|
||
"raw_status_code": stage,
|
||
"accident_fields": accident_fields,
|
||
"turn_patch": patch,
|
||
"dialogue": dialogue,
|
||
"dialogue_timestamps": timestamps,
|
||
}
|