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

"""拍照节点回归：LLM 提议阶段与正文，运行时校验并保存上下文。

fake 输出仅验证接线与校验，不代替真实模型对提示词服从性的评测。
"""

import pytest

from src.agent.collect import CollectReplyError
from src.agent.graph import GraphDeps, build_graph, workflow_fake_llm
from src.agent.llm import FakeLLMService


async def invoke(graph, thread_id, text, llm=None, **initial):
    return await graph.ainvoke(
        {"session_id": thread_id, "input_text": text, **initial},
        config={"configurable": {"thread_id": thread_id}},
        context=GraphDeps(llm=llm or workflow_fake_llm()),
    )


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("source", "target"),
    [("2000", "2001"), ("2001", "2002"), ("2002", "2003"), ("2003", "2005"),
     ("2010", "2011"), ("2011", "2012"), ("2012", "2013"),
     ("2013", "2014"), ("2014", "2015")],
)
async def test_photo_applies_model_proposal(source, target):
    raw = f"<state>{target}</state>模型生成的本轮拍照引导。"
    llm = FakeLLMService(lambda s, u: raw)
    result = await invoke(build_graph(), "photo", "【拍摄完成】", llm=llm,
                          stage_code=source, client_mode="direct")
    assert result["stage_code"] == target
    assert result["response_text"] == raw
    assert len(llm.calls) == 1
    assert "【拍摄完成】" in llm.calls[0]["user_message"]


@pytest.mark.asyncio
@pytest.mark.parametrize("text", ["【拍摄完成】", "【客户端连续3次拍摄识别失败：图片过于模糊】", "【用户无回复】"])
async def test_event_alone_does_not_force_stage(text):
    # 模型保持阶段时，运行时不会仅凭事件强制跳转。
    llm = FakeLLMService(lambda s, u: "<state>2012</state>请确认当前操作。")
    result = await invoke(build_graph(), "hold", text, llm=llm,
                          stage_code="2012", client_mode="direct")
    assert result["stage_code"] == "2012"
    assert len(llm.calls) == 1


@pytest.mark.asyncio
async def test_failure_stage_also_comes_from_model():
    llm = FakeLLMService(lambda s, u: "<state>0005</state>正在转接人工。")
    result = await invoke(build_graph(), "fail", "【客户端连续3次拍摄识别失败：图片过于模糊】", llm=llm,
                          stage_code="2012", client_mode="direct")
    assert result["stage_code"] == "0005"
    assert len(llm.calls) == 1


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("source", "mode", "target", "node"),
    [("2005", "direct", "3001", "单车拍照"),
     ("2005", "browser_addon", "0000", "单车拍照#6"),
     ("2015", "direct", "3002", "双车拍照"),
     ("2015", "browser_addon", "0000", "双车拍照#7")],
)
async def test_mode_selects_workflow_prompt(source, mode, target, node):
    from src.agent.prompts import load_prompt
    from src.agent.photo import PHOTO_PROMPTS

    kind = "single" if source == "2005" else "dual"
    expected_prompt = load_prompt(PHOTO_PROMPTS[(kind, mode)])
    expected_body = expected_prompt.split("--- 原文开始 ---", 1)[1].split("--- 原文结束 ---", 1)[0].strip()
    raw = f"<state>{target}</state>对应模式的结束话术。"
    llm = FakeLLMService(lambda s, u: raw)
    text = "车子左后方" if source == "2005" else "【拍摄完成】"
    result = await invoke(build_graph(), "end", text, llm=llm,
                          stage_code=source, client_mode=mode)
    assert result["stage_code"] == target
    assert len(llm.calls) == 1
    assert f"节点「{node}」" in expected_prompt
    # 模型收到原文，预填变量已经渲染，来源头部不进入模型上下文。
    assert "{{$" not in llm.calls[0]["system_prompt"]
    assert expected_body.splitlines()[0] in llm.calls[0]["system_prompt"]


@pytest.mark.asyncio
async def test_photo_preserves_guidance_and_handoff_history():
    replies = iter([
        "<state>2015</state>您可以点击手机界面中间的拍照按钮。",
        "<state>3002</state>请问第一辆车是由您驾驶的吗？",
    ])
    llm = FakeLLMService(lambda s, u: next(replies))
    graph = build_graph()
    first = await invoke(graph, "history", "拍照按钮在哪里？", llm=llm,
                         stage_code="2015", client_mode="direct")
    second = await invoke(graph, "history", "【拍摄完成】", llm=llm)
    assert "手机界面中间" in first["response_text"]
    assert llm.calls[1]["history"][0] == {"role": "user", "content": "拍照按钮在哪里？"}
    assert llm.calls[1]["user_message"] == "【拍摄完成】"
    assert second["dialogue"][-1]["content"] == second["response_text"]
    assert "驾驶" in second["dialogue"][-1]["content"]


@pytest.mark.asyncio
@pytest.mark.parametrize(("source", "mode", "target"), [
    ("2000", "direct", "2003"),
    ("2015", "browser_addon", "3002"),
    ("2005", "direct", "0000"),
])
async def test_illegal_proposal_fails_after_one_repair(source, mode, target):
    llm = FakeLLMService(lambda s, u: f"<state>{target}</state>非法提议")
    with pytest.raises(CollectReplyError):
        await invoke(build_graph(), "illegal", "继续", llm=llm,
                     stage_code=source, client_mode=mode)
    assert len(llm.calls) == 2