from __future__ import annotations from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker from sqlalchemy.pool import StaticPool from app.db.base import Base from app.schemas.ontology import OntologyParseRequest from app.schemas.user_agent import UserAgentRequest from app.services.ontology import SemanticOntologyService from app.services.user_agent import UserAgentService def build_session_factory() -> sessionmaker[Session]: engine = create_engine( "sqlite+pysqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) Base.metadata.create_all(bind=engine) return sessionmaker(bind=engine, autoflush=False, autocommit=False) def test_user_agent_query_returns_readable_answer_and_actions() -> None: session_factory = build_session_factory() with session_factory() as db: ontology = SemanticOntologyService(db).parse( OntologyParseRequest( query="张三 4 月差旅报销金额是多少", user_id="pytest", ) ) response = UserAgentService(db).respond( UserAgentRequest( run_id=ontology.run_id, user_id="pytest", message="张三 4 月差旅报销金额是多少", ontology=ontology, tool_payload={"record_count": 2, "total_amount": 8800.0}, ) ) assert "8800.00" in response.answer assert len(response.suggested_actions) >= 1 def test_user_agent_prefers_runtime_model_answer_when_available(monkeypatch) -> None: session_factory = build_session_factory() with session_factory() as db: ontology = SemanticOntologyService(db).parse( OntologyParseRequest( query="张三 4 月差旅报销金额是多少", user_id="pytest", ) ) service = UserAgentService(db) monkeypatch.setattr( service, "_generate_answer_with_model", lambda *args, **kwargs: "这是模型回答", ) response = service.respond( UserAgentRequest( run_id=ontology.run_id, user_id="pytest", message="张三 4 月差旅报销金额是多少", ontology=ontology, tool_payload={"record_count": 2, "total_amount": 8800.0}, ) ) assert response.answer == "这是模型回答" def test_user_agent_sanitizes_model_thinking_blocks() -> None: session_factory = build_session_factory() with session_factory() as db: service = UserAgentService(db) assert ( service._sanitize_model_answer("内部推理\n最终答复") == "最终答复" ) def test_user_agent_guides_generic_expense_request() -> None: session_factory = build_session_factory() with session_factory() as db: ontology = SemanticOntologyService(db).parse( OntologyParseRequest( query="我要报销", user_id="pytest", ) ) response = UserAgentService(db).respond( UserAgentRequest( run_id=ontology.run_id, user_id="pytest", message="我要报销", ontology=ontology, tool_payload={"record_count": 9, "total_amount": 12345.0}, ) ) assert "补充费用类型" in response.answer assert "上传票据" in response.answer def test_user_agent_guides_implicit_expense_draft_request() -> None: session_factory = build_session_factory() with session_factory() as db: ontology = SemanticOntologyService(db).parse( OntologyParseRequest( query="我今天去客户现场，招待了客户，花销了1000元", user_id="pytest", ) ) response = UserAgentService(db).respond( UserAgentRequest( run_id=ontology.run_id, user_id="pytest", message="我今天去客户现场，招待了客户，花销了1000元", ontology=ontology, tool_payload={"draft_only": True}, ) ) assert "1000元" in response.answer assert "票据附件" in response.answer assert "报销草稿" in response.answer def test_user_agent_risk_response_includes_rule_citations() -> None: session_factory = build_session_factory() with session_factory() as db: ontology = SemanticOntologyService(db).parse( OntologyParseRequest( query="检查重复报销风险", user_id="pytest", ) ) response = UserAgentService(db).respond( UserAgentRequest( run_id=ontology.run_id, user_id="pytest", message="检查重复报销风险", ontology=ontology, tool_payload={"risk_flags": ["duplicate_expense"]}, ) ) assert response.risk_flags == ["duplicate_expense"] assert any(item.source_type == "rule" for item in response.citations) assert "duplicate_expense" in response.answer def test_user_agent_draft_returns_structured_payload() -> None: session_factory = build_session_factory() with session_factory() as db: ontology = SemanticOntologyService(db).parse( OntologyParseRequest( query="帮我生成张三4月差旅报销草稿", user_id="pytest", ) ) response = UserAgentService(db).respond( UserAgentRequest( run_id=ontology.run_id, user_id="pytest", message="帮我生成张三4月差旅报销草稿", ontology=ontology, tool_payload={"draft_only": True}, ) ) assert response.draft_payload is not None assert response.draft_payload.confirmation_required is True assert "待人工确认" in response.answer def test_user_agent_builds_review_payload_for_multi_document_expense_flow() -> None: session_factory = build_session_factory() with session_factory() as db: ontology = SemanticOntologyService(db).parse( OntologyParseRequest( query="我昨天去上海出差，还请客户A吃饭，帮我生成报销草稿", user_id="pytest", context_json={ "attachment_names": ["机票行程单.png", "餐饮发票.jpg"], "attachment_count": 2, "ocr_documents": [ { "filename": "机票行程单.png", "summary": "机票行程单 上海-北京 金额 680 元", "text": "机票行程单 上海-北京 金额 680 元", "avg_score": 0.93, "warnings": [], }, { "filename": "餐饮发票.jpg", "summary": "餐饮发票 客户招待 金额 320 元", "text": "餐饮发票 客户招待 金额 320 元", "avg_score": 0.91, "warnings": [], }, ], }, ) ) response = UserAgentService(db).respond( UserAgentRequest( run_id=ontology.run_id, user_id="pytest", message="我昨天去上海出差，还请客户A吃饭，帮我生成报销草稿", ontology=ontology, context_json={ "name": "张三", "attachment_names": ["机票行程单.png", "餐饮发票.jpg"], "attachment_count": 2, "ocr_documents": [ { "filename": "机票行程单.png", "summary": "机票行程单 上海-北京 金额 680 元", "text": "机票行程单 上海-北京 金额 680 元", "avg_score": 0.93, "warnings": [], }, { "filename": "餐饮发票.jpg", "summary": "餐饮发票 客户招待 金额 320 元", "text": "餐饮发票 客户招待 金额 320 元", "avg_score": 0.91, "warnings": [], }, ], }, tool_payload={"draft_only": True, "claim_no": "EXP-202605-009", "status": "draft"}, ) ) assert response.review_payload is not None assert len(response.review_payload.document_cards) == 2 assert len(response.review_payload.claim_groups) == 2 assert any( item.action_type == "split_claims" for item in response.review_payload.confirmation_actions )