mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 23:22:00 +00:00
ThearchyHelios
03c18c4d78
一、文案（前端约 1600 处，后端传到界面的约 880 处） - 术语统一：工作流层面用「运行」（继续运行、重新运行、试运行），节点和工具层面用「执行」，不再用「跑」； 总时长 / 执行时长 / 等待审批；服务端；助手；Agent；token；工作流；固定版本；模型接入；系统提示；批准 / 驳回；受管级别 - 口语改书面：取不到、没取回来、读不懂、连不上、多半、照样、兜底、揉成这类说法改成「无法 / 未能 / 失败 / 请」；去掉拟人写法 - 不再露出实现细节：界面上不写 checkpoint、同一拍、哈希（凭证处除外），不写改版历史（「以前…」「老数据」）； 报错、校验、门禁、自动修复里裸写的配置键名和枚举值换成界面上的中文标签，由新的 backend/app/engine/labels.py 统一提供； 发布弹窗里结论句的选项不再是 off / require_citation - 同一件事只留一种说法：生成阶段文案、出具档位说明、画布锁定提示各合成一份 - 去掉界面上的工业场景示例（MES、ERP、「张工」「王工」）和真实产品名，换成通用示例 二、说错的提示（都有测试证明） - 数据库测连接失败时的专用提示以前从不出现（按报错文字全等比较，永远比不上）：改为按错误类型判断，不再拼出重复的前缀 - 「设置 → 供应商」改指「设置 → 模型接入」；「在设置里建数据源」改指「数据」页 - 缺少 API Key、模型接入已停用的报错不再夹英文，写明去哪里填 - 两处原因说错（「这种引用在后续版本支持」「这一版还不支持」）改成真实原因 - 节点说明里不准确的说法改正：代码节点在 microVM 或系统沙箱中执行；报告撰写、口径卡的说明不再说过头 - 导出的证据 CSV 里两列同名，第三列改为「片段编号」 三、前后端对齐 - 前端按后端原文识别日志、报错的地方，新旧两种写法都认得（历史运行里存的是旧文字），逐条补了新旧原文的核对 - 后端交回模型的原指令放进单独的 for_model 字段，前端在解析响应时去掉，不会显示到界面上 四、检查 - 检查脚本里按原文匹配的断言改成新文案；改完会变成永远成立的反向断言逐条修正，并做了变异验证 - 新增 check-copy（静态扫描前端界面字符串和后端上界面的文案）：再出现「跑」「后端」「多半」「令牌」、 界面上的 Copilot、裸写的配置键名、半角标点等就报错；确需保留的写进 scripts/copy-allow.json 并注明理由
497 lines
20 KiB
Python
497 lines
20 KiB
Python
"""MCP 和自定义工具的信任三档：等审批 / 始终允许 · 门控把关 / 始终允许。
|
|
|
|
以前这两类工具在任何审批模式下都不问人：审批关卡只认内置工具的 dangerous 和数据源的
|
|
dangerous_if，它们两样都没有。现在默认每次等人批；「始终允许」可以交给门控模型逐次把关。
|
|
|
|
要守住的几件事：
|
|
- 默认等审批，审批请求带 trust_key，审批卡才有「始终允许」；
|
|
- 门控出任何意外（没答上来、答不成格式）都按交给人工处理，不能变成放行；
|
|
- 点了「始终允许」：写进设置，本节点后面的同一个工具改由门控把关，不再问人；
|
|
- 一次运行用发起时的快照；正式运行不看三档；升级前的运行（快照为 None）照旧不问。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
from langchain_core.messages import AIMessage, SystemMessage, ToolMessage
|
|
from langchain_core.tools import StructuredTool
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import select
|
|
|
|
from app.db.base import SessionLocal
|
|
from app.db.models import Approval, CustomTool, Run, RunEvent, Setting
|
|
from app.engine.runner import run_manager
|
|
from app.tools import custom as custom_tools
|
|
from app.tools.trust import (
|
|
FORMAL_MARK, GATE_SYSTEM, SETTING_KEY, asks_trustable, call_policy, load_trust, parse_verdict,
|
|
set_trust,
|
|
)
|
|
|
|
TOOL = "crm_lookup"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
async def engine_up():
|
|
await run_manager.setup()
|
|
yield
|
|
await run_manager.shutdown()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
async def clean_trust():
|
|
async with SessionLocal() as session:
|
|
row = await session.get(Setting, SETTING_KEY)
|
|
if row:
|
|
await session.delete(row)
|
|
await session.commit()
|
|
yield
|
|
|
|
|
|
@pytest.fixture
|
|
async def crm(monkeypatch):
|
|
"""一个自定义 http 工具。真正的请求换成记账：调了几次、参数是什么。"""
|
|
calls: list[dict] = []
|
|
|
|
async def fake_http(row, args):
|
|
calls.append(dict(args))
|
|
return {"customer": args.get("name"), "level": "gold"}
|
|
|
|
monkeypatch.setattr(custom_tools, "_run_http", fake_http)
|
|
async with SessionLocal() as session:
|
|
session.add(CustomTool(name=TOOL, kind="http", description="按名字查客户等级",
|
|
parameters={"type": "object", "properties": {"name": {"type": "string"}},
|
|
"required": ["name"]},
|
|
config={"url": "https://example.com/crm"}))
|
|
await session.commit()
|
|
yield calls
|
|
async with SessionLocal() as session:
|
|
for row in (await session.execute(select(CustomTool).where(CustomTool.name == TOOL))).scalars():
|
|
await session.delete(row)
|
|
await session.commit()
|
|
|
|
|
|
def node(nid, ntype, **config):
|
|
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0},
|
|
"data": {"label": nid, "config": config}}
|
|
|
|
|
|
def chain(*nodes):
|
|
return {"nodes": list(nodes),
|
|
"edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
|
|
|
|
|
|
def agent_graph(**extra):
|
|
return chain(
|
|
node("start", "input", fields=[{"name": "question"}]),
|
|
node("bot", "agent", prompt="查两位客户的等级", tools=[TOOL], max_steps=5, **extra),
|
|
node("out", "output", fields=[{"name": "结果", "value": "{{ nodes.bot.text }}"}]),
|
|
)
|
|
|
|
|
|
def tool_graph(**extra):
|
|
return chain(
|
|
node("start", "input", fields=[{"name": "question"}]),
|
|
node("t", "tool", tool=TOOL, args={"name": "甲公司"}, **extra),
|
|
node("out", "output", fields=[{"name": "结果", "value": "{{ nodes.t.result }}"}]),
|
|
)
|
|
|
|
|
|
def script(monkeypatch, *, gate=None, names=("甲公司", "乙公司"), team=False):
|
|
"""agent 依次查 names 里的客户；gate 是门控模型的回复（字符串、可调用，或 None 表示不该被问到）。"""
|
|
from app.providers import mock_model
|
|
|
|
gate_calls: list[str] = []
|
|
turns = {"route": 0}
|
|
|
|
def _decide(self, messages):
|
|
if messages and isinstance(messages[0], SystemMessage) and messages[0].content == GATE_SYSTEM:
|
|
gate_calls.append(str(messages[-1].content))
|
|
if gate is None:
|
|
raise AssertionError("不该问门控")
|
|
return AIMessage(content=gate(messages) if callable(gate) else gate)
|
|
if team and self.response_format:
|
|
turns["route"] += 1
|
|
plan = ({"assignments": [{"agent": "sales", "instruction": "查甲公司"}]}
|
|
if turns["route"] == 1 else {"assignments": [], "done": True})
|
|
return AIMessage(content=json.dumps(plan, ensure_ascii=False))
|
|
if self.response_format:
|
|
return AIMessage(content="{}")
|
|
if not self.tools:
|
|
return AIMessage(content="收尾")
|
|
done = sum(isinstance(m, ToolMessage) for m in messages)
|
|
if done < len(names):
|
|
return AIMessage(content="查一下。", tool_calls=[
|
|
{"name": TOOL, "args": {"name": names[done]}, "id": f"call_{done}"}])
|
|
return AIMessage(content="都查完了。")
|
|
|
|
monkeypatch.setattr(mock_model.MockChatModel, "_decide", _decide)
|
|
return gate_calls
|
|
|
|
|
|
ALLOW = '{"verdict": "allow", "reason": "只读查询，和任务一致"}'
|
|
ESCALATE = '{"verdict": "escalate", "reason": "拿不准"}'
|
|
|
|
|
|
async def wait(run_id: str, statuses=("interrupted", "failed", "succeeded"), timeout=20.0) -> Run:
|
|
deadline = asyncio.get_event_loop().time() + timeout
|
|
while asyncio.get_event_loop().time() < deadline:
|
|
async with SessionLocal() as session:
|
|
row = await session.get(Run, run_id)
|
|
if row and row.status in statuses:
|
|
return row
|
|
await asyncio.sleep(0.05)
|
|
raise AssertionError(f"等超时了：{run_id}")
|
|
|
|
|
|
async def events(run_id: str, kind: str | None = None) -> list[RunEvent]:
|
|
async with SessionLocal() as session:
|
|
rows = list((await session.execute(
|
|
select(RunEvent).where(RunEvent.run_id == run_id).order_by(RunEvent.seq))).scalars())
|
|
return [e for e in rows if kind is None or e.type == kind]
|
|
|
|
|
|
async def pending(run_id: str) -> list[Approval]:
|
|
async with SessionLocal() as session:
|
|
return list((await session.execute(select(Approval).where(
|
|
Approval.run_id == run_id, Approval.status == "pending"))).scalars())
|
|
|
|
|
|
async def start(graph, **kw) -> str:
|
|
return (await run_manager.start(graph=graph, input_payload={"question": "查"}, **kw)).id
|
|
|
|
|
|
# ---------------------------------------------------------------- 判定规则
|
|
|
|
|
|
class _NoArgs(BaseModel):
|
|
pass
|
|
|
|
|
|
def _tool(key: str | None, name: str = "x") -> StructuredTool:
|
|
async def run(**_): return ""
|
|
return StructuredTool(name=name, description="", coroutine=run, func=None, args_schema=_NoArgs,
|
|
metadata={"trust_key": key} if key else {})
|
|
|
|
|
|
def test_policy_for_trustable_tools():
|
|
t = _tool("mcp:demo/search", "search")
|
|
assert call_policy(t, "search", {}, {}) == "ask"
|
|
assert call_policy(t, "search", {}, {"mcp:demo/search": "gated"}) == "gate"
|
|
assert call_policy(t, "search", {}, {"mcp:demo/search": "always"}) == "safe"
|
|
# 正式运行：档位一律不看
|
|
assert call_policy(t, "search", {}, {FORMAL_MARK: "1", "mcp:demo/search": "always"}) == "ask"
|
|
# 升级前发起的运行：照旧不问（它们可能正停在别的审批上，多问一次答复就对错号了）
|
|
assert call_policy(t, "search", {}, None) == "safe"
|
|
|
|
|
|
def test_policy_leaves_builtin_tools_alone():
|
|
"""内置工具不看三档：calculator 安全，shell_exec 危险，和以前一样。"""
|
|
assert call_policy(_tool(None, "calculator"), "calculator", {}, {}) == "safe"
|
|
assert call_policy(_tool(None, "shell_exec"), "shell_exec", {}, {"shell_exec": "always"}) == "ask"
|
|
|
|
|
|
def test_always_button_only_where_it_takes_effect():
|
|
t = _tool("crm_lookup")
|
|
assert asks_trustable(t, {}) == "crm_lookup"
|
|
assert asks_trustable(t, {FORMAL_MARK: "1"}) is None
|
|
assert asks_trustable(t, None) is None
|
|
assert asks_trustable(_tool(None), {}) is None
|
|
|
|
|
|
@pytest.mark.parametrize("text, expected", [
|
|
('{"verdict": "allow", "reason": "只读"}', (True, "只读")),
|
|
('好的：{"verdict":"escalate","reason":"要删数据"}', (False, "要删数据")),
|
|
('{"verdict": "allow"}', (True, "（门控未说明理由）")),
|
|
('{"verdict": "yes"}', None),
|
|
("放行", None),
|
|
("", None),
|
|
('{"verdict": "allow", ', None),
|
|
])
|
|
def test_parse_verdict(text, expected):
|
|
assert parse_verdict(text) == expected
|
|
|
|
|
|
# ---------------------------------------------------------------- 接口
|
|
|
|
|
|
@pytest.fixture
|
|
async def client():
|
|
# 不走 TestClient：它会跑应用的启动流程、按环境变量种 provider，后面的运行就不再用 mock
|
|
from app.main import app
|
|
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
|
yield c
|
|
|
|
|
|
async def _row(client) -> dict:
|
|
return next(t for t in (await client.get("/api/tools")).json() if t["id"] == TOOL)
|
|
|
|
|
|
async def test_api_lists_and_sets_trust(crm, client):
|
|
row = await _row(client)
|
|
assert (row["trust"], row["trust_key"], row["runtime_approval"]) == ("ask", TOOL, True)
|
|
|
|
assert (await client.put("/api/tools/trust", json={"key": TOOL, "trust": "always"})).json() \
|
|
== {"key": TOOL, "trust": "always"}
|
|
row = await _row(client)
|
|
assert (row["trust"], row["runtime_approval"]) == ("always", False)
|
|
|
|
await client.put("/api/tools/trust", json={"key": TOOL, "trust": "gated"})
|
|
row = await _row(client)
|
|
assert (row["trust"], row["runtime_approval"]) == ("gated", True)
|
|
|
|
# 改回等审批：从表里删掉，不留一条 ask
|
|
await client.put("/api/tools/trust", json={"key": TOOL, "trust": "ask"})
|
|
stored = (await client.get("/api/settings")).json().get(SETTING_KEY) or {}
|
|
assert TOOL not in stored.get("tools", {})
|
|
|
|
assert (await client.put("/api/tools/trust", json={"key": TOOL, "trust": "yes"})).status_code == 422
|
|
assert (await client.put("/api/tools/trust", json={"key": "", "trust": "ask"})).status_code == 422
|
|
|
|
|
|
async def test_settings_page_cannot_overwrite_trust(crm, client):
|
|
"""设置页整组回写时带着一份旧的三档，不能把工具页刚改的冲掉。"""
|
|
await client.put("/api/tools/trust", json={"key": TOOL, "trust": "always"})
|
|
await client.put("/api/settings", json={"values": {SETTING_KEY: {"tools": {}}, "run": {"stream_tokens": True}}})
|
|
assert (await _row(client))["trust"] == "always"
|
|
|
|
|
|
async def test_gate_model_settings_have_defaults(client):
|
|
run = (await client.get("/api/settings")).json()["run"]
|
|
assert run["tool_gate_provider"] is None and run["tool_gate_model"] is None
|
|
|
|
|
|
# ---------------------------------------------------------------- agent 节点
|
|
|
|
|
|
async def test_custom_tool_waits_for_approval_by_default(crm, monkeypatch):
|
|
script(monkeypatch, names=("甲公司",))
|
|
run_id = await start(agent_graph())
|
|
run = await wait(run_id)
|
|
assert run.status == "interrupted", f"自定义工具没等审批就执行了：{run.status} {run.error}"
|
|
assert crm == []
|
|
[approval] = await pending(run_id)
|
|
assert approval.payload["trust_key"] == TOOL
|
|
[asked] = await events(run_id, "human.requested")
|
|
assert asked.data["trust_key"] == TOOL
|
|
|
|
await run_manager.resume(run_id, {"approved": True})
|
|
run = await wait(run_id, ("failed", "succeeded"))
|
|
assert run.status == "succeeded", run.error
|
|
assert crm == [{"name": "甲公司"}]
|
|
|
|
|
|
async def test_always_runs_without_asking(crm, monkeypatch):
|
|
await _set(TOOL, "always")
|
|
script(monkeypatch, gate=None)
|
|
run = await wait(await start(agent_graph()))
|
|
assert run.status == "succeeded", run.error
|
|
assert [c["name"] for c in crm] == ["甲公司", "乙公司"]
|
|
assert not await events(run.id, "human.requested")
|
|
|
|
|
|
async def test_gated_and_allowed_runs_without_asking(crm, monkeypatch):
|
|
await _set(TOOL, "gated")
|
|
asked = script(monkeypatch, gate=ALLOW)
|
|
run = await wait(await start(agent_graph()))
|
|
assert run.status == "succeeded", run.error
|
|
assert len(asked) == 2 and "甲公司" in asked[0] and "查两位客户的等级" in asked[0]
|
|
gated = await events(run.id, "tool.gated")
|
|
assert [(e.data["verdict"], e.data["reason"]) for e in gated] == [("allow", "只读查询，和任务一致")] * 2
|
|
assert not await events(run.id, "human.requested")
|
|
assert len(crm) == 2
|
|
|
|
|
|
@pytest.mark.parametrize("reply", [ESCALATE, "我觉得可以", '{"verdict": "maybe"}'])
|
|
async def test_gate_that_does_not_clearly_allow_hands_over_to_a_person(crm, monkeypatch, reply):
|
|
await _set(TOOL, "gated")
|
|
script(monkeypatch, gate=reply, names=("甲公司",))
|
|
run = await wait(await start(agent_graph()))
|
|
assert run.status == "interrupted", "门控没放行，却没交给人"
|
|
assert crm == []
|
|
[gated] = await events(run.id, "tool.gated")
|
|
assert gated.data["verdict"] == "escalate"
|
|
|
|
|
|
async def test_gate_that_crashes_hands_over_to_a_person(crm, monkeypatch):
|
|
await _set(TOOL, "gated")
|
|
|
|
def boom(_):
|
|
raise RuntimeError("网关断开")
|
|
|
|
script(monkeypatch, gate=boom, names=("甲公司",))
|
|
run = await wait(await start(agent_graph()))
|
|
assert run.status == "interrupted"
|
|
[gated] = await events(run.id, "tool.gated")
|
|
assert gated.data["verdict"] == "escalate" and "门控模型未给出判断" in gated.data["reason"]
|
|
|
|
|
|
async def test_gate_verdict_is_not_asked_again_when_the_node_replays(crm, monkeypatch):
|
|
"""门控判了交给人 → 批准 → 节点重放。重放时不能再问门控：这次可能判放行，interrupt 就错号了。"""
|
|
await _set(TOOL, "gated")
|
|
replies = iter([ESCALATE, ALLOW, ALLOW])
|
|
asked = script(monkeypatch, gate=lambda _: next(replies))
|
|
run_id = await start(agent_graph())
|
|
await wait(run_id)
|
|
await run_manager.resume(run_id, {"approved": True})
|
|
run = await wait(run_id, ("failed", "succeeded", "interrupted"))
|
|
assert run.status == "succeeded", run.error
|
|
assert len(asked) == 2, f"门控被问了 {len(asked)} 次，重放时又问了一遍"
|
|
assert [c["name"] for c in crm] == ["甲公司", "乙公司"]
|
|
|
|
|
|
async def test_always_button_persists_and_covers_the_rest_of_the_node(crm, monkeypatch):
|
|
asked = script(monkeypatch, gate=ALLOW)
|
|
run_id = await start(agent_graph())
|
|
await wait(run_id)
|
|
await run_manager.resume(run_id, {"approved": True, "always": True})
|
|
run = await wait(run_id, ("failed", "succeeded", "interrupted"))
|
|
assert run.status == "succeeded", f"点了始终允许，第二次调用还在等审批：{run.status}"
|
|
assert len(await events(run_id, "human.requested")) == 1
|
|
# 第一次是人批的，第二次改由门控把关
|
|
assert len(asked) == 1 and "乙公司" in asked[0]
|
|
resolved = (await events(run_id, "human.resolved"))[0]
|
|
assert resolved.data.get("always") is True
|
|
async with SessionLocal() as session:
|
|
assert (await load_trust(session)) == {TOOL: "gated"}
|
|
|
|
|
|
async def test_always_is_ignored_where_there_is_no_button(monkeypatch):
|
|
"""内置工具的审批带了 always 也不理：没有可改的档位。"""
|
|
graph = chain(
|
|
node("start", "input", fields=[{"name": "question"}]),
|
|
node("t", "tool", tool="calculator", args={"expression": "1+1"}, approval="always"),
|
|
node("out", "output", fields=[]),
|
|
)
|
|
run_id = await start(graph)
|
|
await wait(run_id)
|
|
[approval] = await pending(run_id)
|
|
assert "trust_key" not in approval.payload
|
|
await run_manager.resume(run_id, {"approved": True, "always": True})
|
|
await wait(run_id, ("failed", "succeeded"))
|
|
async with SessionLocal() as session:
|
|
assert await load_trust(session) == {}
|
|
|
|
|
|
async def test_a_run_keeps_the_trust_it_started_with(crm, monkeypatch):
|
|
"""运行中途在工具页改了档位，这次运行不受影响：审批恢复时的判定必须和第一次一样。"""
|
|
script(monkeypatch, gate=None)
|
|
run_id = await start(agent_graph())
|
|
await wait(run_id)
|
|
await _set(TOOL, "always")
|
|
await run_manager.resume(run_id, {"approved": True})
|
|
run = await wait(run_id, ("failed", "succeeded", "interrupted"))
|
|
assert run.status == "interrupted", "快照被中途的改动顶掉了：第二次调用没再问"
|
|
async with SessionLocal() as session:
|
|
assert (await session.get(Run, run_id)).tool_trust == {}
|
|
|
|
|
|
async def test_formal_runs_ignore_trust(crm, monkeypatch):
|
|
await _set(TOOL, "always")
|
|
script(monkeypatch, gate=None, names=("甲公司",))
|
|
run_id = await start(agent_graph(), run_class="formal")
|
|
run = await wait(run_id)
|
|
assert run.status == "interrupted", "正式运行吃了用户的「始终允许」"
|
|
[approval] = await pending(run_id)
|
|
assert "trust_key" not in approval.payload
|
|
started = (await events(run_id, "run.started"))[0]
|
|
assert started.data["tool_trust"] == "formal"
|
|
|
|
|
|
async def test_run_started_records_the_snapshot(crm, monkeypatch):
|
|
await _set(TOOL, "always")
|
|
await _set("mcp:demo/search", "gated")
|
|
script(monkeypatch, gate=None)
|
|
run = await wait(await start(agent_graph()))
|
|
started = (await events(run.id, "run.started"))[0]
|
|
assert started.data["tool_trust"] == {TOOL: "always", "mcp:demo/search": "gated"}
|
|
|
|
|
|
async def test_node_approval_never_skips_the_gate(crm, monkeypatch):
|
|
await _set(TOOL, "gated")
|
|
script(monkeypatch, gate=None)
|
|
run = await wait(await start(agent_graph(approval="never")))
|
|
assert run.status == "succeeded", run.error
|
|
assert not await events(run.id, "tool.gated")
|
|
|
|
|
|
# ---------------------------------------------------------------- 工具节点
|
|
|
|
|
|
async def test_tool_node_gated_allow_runs(crm, monkeypatch):
|
|
await _set(TOOL, "gated")
|
|
script(monkeypatch, gate=ALLOW)
|
|
run = await wait(await start(tool_graph()))
|
|
assert run.status == "succeeded", run.error
|
|
assert crm == [{"name": "甲公司"}]
|
|
assert [e.data["verdict"] for e in await events(run.id, "tool.gated")] == ["allow"]
|
|
|
|
|
|
async def test_tool_node_escalation_asks_with_the_always_button(crm, monkeypatch):
|
|
await _set(TOOL, "gated")
|
|
asked = script(monkeypatch, gate=ESCALATE)
|
|
run_id = await start(tool_graph())
|
|
assert (await wait(run_id)).status == "interrupted"
|
|
[approval] = await pending(run_id)
|
|
assert approval.payload["trust_key"] == TOOL
|
|
await run_manager.resume(run_id, {"approved": True, "always": True})
|
|
run = await wait(run_id, ("failed", "succeeded"))
|
|
assert run.status == "succeeded", run.error
|
|
assert len(asked) == 1, "重放时又问了一遍门控"
|
|
assert len(await events(run_id, "tool.gated")) == 1
|
|
assert crm == [{"name": "甲公司"}]
|
|
|
|
|
|
async def test_tool_node_default_waits(crm, monkeypatch):
|
|
script(monkeypatch, gate=None)
|
|
run = await wait(await start(tool_graph()))
|
|
assert run.status == "interrupted" and crm == []
|
|
|
|
|
|
# ---------------------------------------------------------------- 协作团队
|
|
|
|
|
|
def team_graph():
|
|
return chain(
|
|
node("start", "input", fields=[{"name": "question"}]),
|
|
node("team", "supervisor", goal="查客户", max_rounds=3,
|
|
agents=[{"name": "sales", "system": "你是销售助理", "tools": [TOOL], "max_steps": 3}]),
|
|
node("out", "output", fields=[]),
|
|
)
|
|
|
|
|
|
async def test_team_member_does_not_run_an_unapproved_custom_tool(crm, monkeypatch):
|
|
script(monkeypatch, gate=None, names=("甲公司",), team=True)
|
|
run = await wait(await start(team_graph()))
|
|
assert run.status == "succeeded", run.error
|
|
assert crm == []
|
|
assert "tool_needs_approval" in [e.data.get("code") for e in await events(run.id, "log")]
|
|
|
|
|
|
async def test_team_member_gated(crm, monkeypatch):
|
|
await _set(TOOL, "gated")
|
|
script(monkeypatch, gate=ALLOW, names=("甲公司",), team=True)
|
|
run = await wait(await start(team_graph()))
|
|
assert run.status == "succeeded", run.error
|
|
assert crm == [{"name": "甲公司"}]
|
|
[gated] = await events(run.id, "tool.gated")
|
|
assert gated.data["agent"] == "sales" and gated.data["verdict"] == "allow"
|
|
|
|
|
|
async def test_team_member_gate_escalation_is_not_executed(crm, monkeypatch):
|
|
await _set(TOOL, "gated")
|
|
script(monkeypatch, gate=ESCALATE, names=("甲公司",), team=True)
|
|
run = await wait(await start(team_graph()))
|
|
assert run.status == "succeeded", run.error
|
|
assert crm == []
|
|
ends = [e.data.get("preview", "") for e in await events(run.id, "tool.end")]
|
|
assert any("门控模型未批准（拿不准）" in p for p in ends), ends
|
|
|
|
|
|
async def _set(key: str, level: str) -> None:
|
|
async with SessionLocal() as session:
|
|
await set_trust(session, key, level)
|