ZNJJ-api-server/test/agent/test_workflow_prompts.py

"""20261020 工作流的提示词移植与模型决策回归。"""

from datetime import datetime, timezone
import json
from pathlib import Path

import pytest

from src.agent.collect import CollectReplyError, run_collect
from src.agent.llm import FakeLLMService
from src.agent.prompts import COLLECTION, load_prompt, render_prompt

WORKFLOW = Path(__file__).resolve().parents[2] / "workflow/20261022/事故信息采集2025118.json"
NODES = {
    "collection": "事故信息收集",
    "single_photo": "单车拍照",
    "dual_photo": "双车拍照",
    "single_photo_addon": "单车拍照#6",
    "dual_photo_addon": "双车拍照#7",
    "single_confirm": "单车信息确认",
    "dual_confirm": "双车信息确认",
    "extraction": "文本内容提取事故信息",
}
_INPUT_KEYS = {"extraction": "description"}


@pytest.mark.parametrize(("prompt", "node_name"), NODES.items())
def test_archived_prompt_matches_workflow(prompt, node_name):
    workflow = json.loads(WORKFLOW.read_text(encoding="utf-8-sig"))
    node = next(n for n in workflow["nodes"] if n["name"] == node_name)
    input_key = _INPUT_KEYS.get(prompt, "systemPrompt")
    original = next(i["value"] for i in node["inputs"] if i["key"] == input_key)
    archived = load_prompt(prompt).split("--- 原文开始 ---", 1)[1].split("--- 原文结束 ---", 1)[0]
    assert archived.strip() == original.strip()


def test_render_replaces_time_and_person_variables():
    # 20261022：collection/extraction 提示词本体不含变量（动态上下文走
    # user 消息的 <turn_context>）；拍照/确认提示词尾部仍渲染人员变量。
    rendered = render_prompt(COLLECTION, {})
    assert "{{$" not in rendered
    rendered = render_prompt("dual_confirm", {
        "person_fields": {"hphm1": "测试A", "hphm2": "测试B", "sfzmwh1": "1234"}
    })
    assert "测试A" in rendered and "测试B" in rendered
    assert "1234" in rendered
    assert "{{$" not in rendered
    assert "未提供" in rendered


def test_user_message_carries_turn_context():
    from src.agent.collect import _build_user_message
    from datetime import datetime, timezone
    now = datetime(2026, 10, 22, 8, 0, tzinfo=timezone.utc)
    message = _build_user_message({"input_text": "两辆"}, now=now)
    assert "<turn_context>" in message
    assert "本轮参考时间：2026-10-22 08:00:00 +0000" in message
    assert "<user_input>" in message
    assert "两辆" in message


@pytest.mark.asyncio
async def test_short_answer_does_not_override_model_routing():
    # “在”可能是明确回答；运行时不得按字数一律拒绝分流。
    llm = FakeLLMService(lambda s, u: "<state>2000</state>信息已齐，请拍摄车辆前方。")
    result = await run_collect({
        "stage_code": "1002", "input_text": "在", "clarify_count": 0,
        "dialogue": [{"role": "assistant", "content": "您还在事故现场吗？"}],
    }, llm)
    assert result["stage_code"] == "2000"
    assert result["clarify_count"] == 0
    assert len(llm.calls) == 2  # 对话 + 提取


@pytest.mark.asyncio
async def test_format_and_stage_errors_share_one_repair_budget():
    replies = iter(["缺少前缀", "<state>9999</state>非法阶段"])
    llm = FakeLLMService(lambda s, u: next(replies))
    with pytest.raises(CollectReplyError):
        await run_collect({"stage_code": "1002", "input_text": "追尾了"}, llm)
    assert len(llm.calls) == 2


@pytest.mark.asyncio
async def test_collect_sends_history_as_messages():
    history = [
        {"role": "user", "content": "没有人受伤"},
        {"role": "assistant", "content": "<state>1002</state>请问有几辆车？"},
    ]
    llm = FakeLLMService(lambda s, u: "<state>1002</state>请问您还在现场吗？")
    await run_collect({
        "stage_code": "1002", "input_text": "两辆", "dialogue": history,
    }, llm)
    assert llm.calls[0]["history"] == history
    # 20261022 格式：<turn_context> + <user_input>
    assert "<turn_context>" in llm.calls[0]["user_message"]
    assert "<user_input>\n两辆\n</user_input>" in llm.calls[0]["user_message"]
    assert "没有人受伤" not in llm.calls[0]["user_message"]


@pytest.mark.asyncio
async def test_openai_adapter_preserves_message_roles():
    from types import SimpleNamespace
    from unittest.mock import AsyncMock
    from src.agent.llm import OpenAILLMService

    from langchain_core.messages import AIMessage

    create = AsyncMock(return_value=AIMessage(content="<state>1002</state>收到。"))
    llm = OpenAILLMService(model="test", api_key="test", max_retries=0)
    llm._client = SimpleNamespace(ainvoke=create)
    history = [{"role": "assistant", "content": "之前的问题"}]
    await llm.complete("系统提示词", "本轮输入", history=history)
    assert create.call_args.args[0] == [
        {"role": "system", "content": "系统提示词"},
        {"role": "assistant", "content": "之前的问题"},
        {"role": "user", "content": "本轮输入"},
    ]