zhifu-platform/tests/agent_model_acceptance.py

import os
from uuid import UUID

from sqlalchemy import func, select
from test_after_sales import scenario  # noqa: F401

from zhifu.modules.after_sales.models import Approval, Ticket
from zhifu.modules.agent_runtime.config import AgentSettings
from zhifu.modules.agent_runtime.models import AgentRun, ModelAttempt
from zhifu.modules.agent_runtime.worker import execute_one
from zhifu.modules.integrations.models import ExternalOperation


async def test_real_model_queries_order_and_creates_proposal(scenario):  # noqa: F811
    settings = AgentSettings()
    assert settings.model_name and settings.model_url, "需要配置实际模型名称和地址"
    assert os.environ.get(settings.credential_env), "需要配置实际模型凭据"
    body = {
        "store_id": str(scenario["store"]),
        "message": (
            f"请查询订单 {scenario['order_id']}，商品 keyboard-1 的一个键盘按键失灵。"
            "请为我创建数量为 1 的换货方案，创建后告诉我真实方案标识和下一步操作。"
        ),
    }
    response = await scenario["call"](
        "POST",
        "/openapi/v1/agent/runs",
        payload=body,
        key="model-acceptance",
    )
    assert response.status_code == 202, response.text
    run_id = UUID(response.json()["run_id"])
    repeated = await scenario["call"](
        "POST", "/openapi/v1/agent/runs", payload=body, key="model-acceptance"
    )
    assert repeated.json()["run_id"] == str(run_id)
    conflict = await scenario["call"](
        "POST",
        "/openapi/v1/agent/runs",
        payload={**body, "message": "其他问题"},
        key="model-acceptance",
    )
    assert conflict.status_code == 409
    app = scenario["app"]
    assert await execute_one(app.state.sessions, app.state.state, settings)
    response = await scenario["call"]("GET", f"/openapi/v1/agent/runs/{run_id}")
    assert response.json()["status"] == "COMPLETED", response.text
    async with app.state.sessions() as session:
        run = await session.get(AgentRun, run_id.bytes)
        calls = {
            call["name"]
            for message in run.messages
            for call in message["data"].get("tool_calls", [])
        }
        assert {"get_order", "create_exchange_proposal"} <= calls
        attempts = (
            await session.scalars(select(ModelAttempt).where(ModelAttempt.run_id == run_id.bytes))
        ).all()
        assert attempts and all(attempt.status == "RECEIVED" for attempt in attempts)
        assert all(attempt.usage is not None for attempt in attempts)
        tickets = (
            await session.scalars(
                select(Ticket).where(Ticket.tenant_id == scenario["tenant"].bytes)
            )
        ).all()
        assert len(tickets) == 1 and tickets[0].status == "DRAFT"
        for model in (Approval, ExternalOperation):
            count = await session.scalar(
                select(func.count())
                .select_from(model)
                .where(model.tenant_id == scenario["tenant"].bytes)
            )
            assert count == 0
        print(
            {
                "model": settings.model_name,
                "model_calls": len(attempts),
                "total_tokens": sum(attempt.usage["total_tokens"] for attempt in attempts),
                "ticket_status": tickets[0].status,
            }
        )