72 lines
2.3 KiB
Python
72 lines
2.3 KiB
Python
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)
|