aps-agent/tests/pi_primary_adapter.py

from __future__ import annotations

from server.contracts import AgentReply, IntentResult


def install_fake_pi_tool_adapter(monkeypatch) -> None:
    """让历史网关测试用确定性 Pi 替身选择工具，不恢复规则意图识别。"""
    from server.agent_core import fallback_lane
    from server.agent_core.tool_runtime import run_tool_async

    async def fake_pi(
        store,
        session_id: str,
        intent: IntentResult,
        *,
        actor: str = "planner",
        **_kwargs,
    ) -> AgentReply:
        query = str(intent.params.get("query") or "")
        if ("根据这些数据排产" in query or "立即排产" in query
                or "手动触发排产" in query):
            selected = IntentResult(
                intent="folder.schedule",
                params={"query": query},
                confidence=1.0,
                source="LLM",
            )
        elif query.strip() in ("排产还缺什么", "查看缺少资料"):
            selected = IntentResult(
                intent="data.analyze",
                params={"query": query},
                confidence=1.0,
                source="LLM",
            )
        elif "分析" in query and ("数据" in query or "文件" in query):
            selected = IntentResult(
                intent="data.analyze",
                params={"query": query},
                confidence=1.0,
                source="LLM",
            )
        elif "检查点" in query:
            selected = IntentResult(
                intent="checkpoint.create",
                params={},
                confidence=1.0,
                source="LLM",
            )
        elif "试排" in query:
            params = {}
            if "交期优先" in query:
                params["strategy"] = "DELIVERY_FIRST"
            elif "产能均衡" in query:
                params["strategy"] = "CAPACITY_BALANCE"
            selected = IntentResult(
                intent="schedule.run",
                params=params,
                confidence=1.0,
                source="LLM",
            )
        else:
            return AgentReply(text="测试 Pi 未选择任何工具。")

        return await run_tool_async(
            store,
            session_id,
            selected,
            actor=actor,
        )

    monkeypatch.setattr(fallback_lane, "propose_reply", fake_pi)