111 lines
4.9 KiB
Python
111 lines
4.9 KiB
Python
"""后台挂起「分层 DAG」实测脚本：带一个并行 stage 的业务链。
|
||
|
||
POST 启动一个 dag 作业（stage1 两席位并行 → stage2 → stage3），轮询打印每帧的
|
||
dispatchChain 节点状态，直观看到：①stage1 两席位**同时 active**（并行）；②stage2/3
|
||
在前一 stage 全 done 后才 active（串行）；③orchestrationPlan 在响应里回得来（持久化生效）。
|
||
|
||
用法：PYTHONPATH=. .venv/Scripts/python scripts/probe_roundtable_dag_job.py
|
||
环境：GATEWAY_URL 默认 http://127.0.0.1:8001；AUTH_COOKIE 可选。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import os
|
||
import time
|
||
|
||
import httpx
|
||
|
||
GATEWAY_URL = os.getenv("GATEWAY_URL", "http://127.0.0.1:8001").rstrip("/")
|
||
AUTH_COOKIE = os.getenv("AUTH_COOKIE", "")
|
||
AUTH_BEARER = os.getenv("AUTH_BEARER", "")
|
||
|
||
# 4 个已 seed 的圆桌席位。stage1 = 风险审查 ‖ 方案设计（并行），stage2 = 执行规划，stage3 = 总结。
|
||
AGENTS = [
|
||
{"agentId": "roundtable-risk-review", "name": "风险审查"},
|
||
{"agentId": "roundtable-solution-design", "name": "方案设计"},
|
||
{"agentId": "roundtable-execution-plan", "name": "执行规划"},
|
||
{"agentId": "roundtable-summary", "name": "方案总结"},
|
||
]
|
||
PLAN = {
|
||
"mode": "dag",
|
||
"stages": [
|
||
{"id": "s0", "agentIds": ["roundtable-risk-review", "roundtable-solution-design"], "goal": "并行评估风险并产出初步方案"},
|
||
{"id": "s1", "agentIds": ["roundtable-execution-plan"], "goal": "据风险与方案制定执行规划"},
|
||
{"id": "s2", "agentIds": ["roundtable-summary"]},
|
||
],
|
||
"finalSynthesis": True,
|
||
"coordinatorPrompt": "先并行评估、再规划执行、最后总结，逐阶段承接。",
|
||
}
|
||
INTENT = "为一家中型制造企业设计一套数字化转型的整体方案"
|
||
TERMINAL = {"done", "error", "cancelled"}
|
||
|
||
|
||
def _headers() -> dict[str, str]:
|
||
h = {"Content-Type": "application/json"}
|
||
if AUTH_COOKIE:
|
||
h["Cookie"] = AUTH_COOKIE
|
||
if AUTH_BEARER:
|
||
h["Authorization"] = f"Bearer {AUTH_BEARER}"
|
||
return h
|
||
|
||
|
||
def _chain_str(nodes: list[dict]) -> str:
|
||
sym = {"pending": ".", "active": ">RUN", "done": "DONE"}
|
||
return " ".join(f"{n['name']}={sym.get(n['state'], '?')}" for n in nodes)
|
||
|
||
|
||
async def main() -> None:
|
||
async with httpx.AsyncClient(timeout=30.0, trust_env=False) as client:
|
||
print(f"[probe] POST {GATEWAY_URL}/api/roundtable-jobs mode=dag (stage1 并行)")
|
||
resp = await client.post(
|
||
f"{GATEWAY_URL}/api/roundtable-jobs",
|
||
headers=_headers(),
|
||
json={"intent": INTENT, "agents": AGENTS, "orchestrationMode": "dag", "orchestrationPlan": PLAN},
|
||
)
|
||
if resp.status_code != 201:
|
||
print(f"[probe] start failed: HTTP {resp.status_code}: {resp.text[:500]}")
|
||
return
|
||
job = resp.json()
|
||
job_id = job["id"]
|
||
# 验证 plan 在响应里回得来（持久化 + _to_response 生效）
|
||
plan_back = job.get("orchestrationPlan")
|
||
print(f"[probe] started {job_id} orchestrationMode={job.get('orchestrationMode')} "
|
||
f"orchestrationPlan stages={len((plan_back or {}).get('stages') or [])}")
|
||
|
||
last = ""
|
||
t0 = time.perf_counter()
|
||
saw_parallel = False
|
||
while True:
|
||
await asyncio.sleep(1.0)
|
||
r = await client.get(f"{GATEWAY_URL}/api/roundtable-jobs/{job_id}", headers=_headers())
|
||
if r.status_code != 200:
|
||
print(f"[probe] get failed: HTTP {r.status_code}: {r.text[:300]}")
|
||
return
|
||
j = r.json()
|
||
nodes = j.get("dispatchChain") or []
|
||
active_seats = [n["name"] for n in nodes if n["state"] == "active"]
|
||
if len(active_seats) >= 2:
|
||
saw_parallel = True
|
||
line = f"status={j['status']} phase={j['phase']} consensus={j['consensusPercentage']}% [{_chain_str(nodes)}]"
|
||
if line != last:
|
||
print(f"[probe] +{time.perf_counter() - t0:5.1f}s {line}")
|
||
last = line
|
||
if j["status"] in TERMINAL:
|
||
print(f"\n[probe] terminal: {j['status']}")
|
||
if j["status"] == "error":
|
||
print(f"[probe] error detail: {j.get('error')}")
|
||
# 对话顺序证明 stage 串行
|
||
seats = [d for d in (j.get("dialogues") or []) if d["role"] == "seat"]
|
||
print(f"[probe] 席位交付顺序: {[d['agentId'] for d in seats]}")
|
||
leaders = [d for d in (j.get("dialogues") or []) if d["role"] == "leader"]
|
||
print(f"[probe] 总控发言轮数: {len(leaders)}（3 stage 派活 + 1 收口 = 4）")
|
||
print(f"[probe] 是否观察到 stage 内并行（≥2 席位同时 active）: {'是 ✅' if saw_parallel else '否 ❌'}")
|
||
sample = next((d["content"] for d in seats if d.get("content")), "")
|
||
print(f"[probe] 席位交付样例: {sample[:80]}")
|
||
return
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|