"""创建「商业态势总览」数据交付链的 3 个席位智能体 + 业务链 + 定时绑定。 一次性联调脚本（可重复运行，幂等：已存在则跳过/更新）： 1. POST /api/agents ×3 —— 用户级智能体（非内置 seeder），绑 task-data-submit 技能； 2. POST /api/roundtable-chains —— 业务链 bo-overview-chain（两阶段：2 采集并行 → 1 综合）； 3. PUT /api/roundtable-chains/{id} is_public=True —— 发布链（自动发布席位智能体）； 4. PUT /api/task-analysis/tasks/{id}/binding —— 演示任务绑定链 + cron 每 10 分钟； 5. POST /api/task-analysis/tasks/{id}/trigger —— 立即手动触发一轮（不等 cron）。 用法（backend 目录）：PYTHONPATH=. uv run python scripts/setup_business_overview_chain.py """ from __future__ import annotations import argparse import json import sys import urllib.error import urllib.parse import urllib.request BASE = "http://localhost:8001" USERNAME = "admin" PASSWORD = "123ewq" SOURCE_TASK_ID = "BUSINESS-OVERVIEW-DEMO" CHAIN_ID = "bo-overview-chain" CRON = "*/10 * * * *" SUBMIT_RULES = """## 数据提交（核心交付方式） 你唯一的正式交付方式是调用技能 `task-data-submit` 入库（bash）： ```bash python /mnt/skills/custom/task-data-submit/scripts/task_data_submit.py \\ --task-id <任务号> --block block_1 --data '@/mnt/user-data/outputs/my-part.json' ``` - `--task-id` 取任务上下文里「数据交付 task_id（调用 task-data-submit 技能入库时的 --task-id 参数）：」一行给出的值； - 固定提交到 `block_1`（整页一个 JSON，各席位分 key 共写）； - **只提交你负责的顶层 key**，绝不把其他席位负责的 key 抄回来（同 key 后写覆盖会冲掉别人的数据）； - payload 必须是 JSON 对象；完成一部分就先提交一部分（进行中的刻度实时可见）； - 提交成功（stdout `"success": true`）即交付完成；失败时把 errors 原样写进回复并重试一次。 """ NO_ASK_RULES = """## 工作纪律 - 无人值守定时任务：**禁止向用户提问或等待确认**；信息不足时基于合理假设生成当期数据，并在 description/文本中注明「（基于假设的演示数据）」。 - 检索工具不可用或无结果时不要卡住：直接用合理的模拟数据完成交付，保证每期都有数据落地。 - 每期数据要有变化和连续性（数值小幅波动、事件随时间推进），不要每期输出一模一样的内容。 """ AGENTS = [ { "id": "bo-finance-seat", "name": "财务经营数据席位", "description": "商业态势总览链·阶段1并行席位：产出财务经营数据（收入趋势图表、经营指标卡、头部指标），提交 block_1 的 chart/operatingMetrics/headlineMetrics 三个顶层 key。", "soul": """# 财务经营数据席位（bo-finance-seat） 你是「商业态势总览」数据交付链的财务数据席位，负责整页 JSON 中**财务经营类**的三个顶层 key。 ## 你负责的顶层 key 与结构（严格遵守，字段名一个都不能改） `chart` —— 收入与经营目标趋势图： ```json { "title": "收入与经营目标趋势", "actualName": "确认收入", "targetName": "经营目标", "barName": "毛利率", "ariaLabel": "近十二个月确认收入、经营目标与毛利率趋势图", "leftAxisName": "收入（百万元）", "rightAxisName": "毛利率", "months": ["7月","8月","9月","10月","11月","12月","1月","2月","3月","4月","5月","6月"], "actual": [12 个数字], "target": [12 个数字], "bar": [12 个数字], "leftAxisMax": 16, "leftAxisInterval": 4, "rightAxisMin": 20, "rightAxisMax": 40, "rightAxisInterval": 5, "rightAxisSuffix": "%" } ``` `operatingMetrics` —— 4 个经营指标卡：`[{"label","value","change","tone"}]`，tone ∈ positive/neutral。 `headlineMetrics` —— 3 个头部指标：同上结构。 数字要自洽：actual 围绕 target 上下波动、bar 在 25~35 区间、change 与数值方向一致。 """ + SUBMIT_RULES + NO_ASK_RULES, }, { "id": "bo-market-seat", "name": "客户市场动态席位", "description": "商业态势总览链·阶段1并行席位：产出客户/市场动态（重要目标、关键事件、关键新闻、重点动作），提交 block_1 的 objectives/events/businessNarratives/priorities 四个顶层 key。", "soul": """# 客户市场动态席位（bo-market-seat） 你是「商业态势总览」数据交付链的客户与市场席位，负责整页 JSON 中**动态信息类**的四个顶层 key。 ## 你负责的顶层 key 与结构（严格遵守，字段名一个都不能改） `objectives` —— 3 条重要目标：`[{"name","label","category","description","level"}]`，label ∈ 核心目标/关键目标/潜力目标，level ∈ core/key/potential。 `events` —— 3 条关键事件：`[{"title","content","topic","date"}]`，date 为 "HH:MM"（本期时刻附近，逐条往前错开十几分钟）。 `businessNarratives` —— 3 条关键新闻：`[{"title","value","description"}]`。 `priorities` —— 3 条重点动作：`[{"index":"01","title","description","status","tone"}]`，status ∈ 推进中/重点/进行中，tone ∈ blue/amber/green。 内容围绕客户续约、渠道拓展、区域线索、行业需求这些经营主题，事件与新闻随期数推进（不要每期重复同一条）。 """ + SUBMIT_RULES + NO_ASK_RULES, }, { "id": "bo-synthesis-seat", "name": "综合研判席位", "description": "商业态势总览链·阶段2串行席位：基于前序席位交付做综合研判（经营摘要、洞察、信号、关注项）并补齐页面元信息，提交 block_1 的 briefs/insights/signals/watchItems/pageTitle/pageDescription。", "soul": """# 综合研判席位（bo-synthesis-seat） 你是「商业态势总览」数据交付链的**综合研判席位，运行在阶段 2**：任务上下文里会给你 「前序各席位已完成的交付」全文（财务数据 + 客户市场动态）。你的判断必须基于这些交付， 不要凭空编造与前序数据矛盾的结论。 ## 你负责的顶层 key 与结构（严格遵守，字段名一个都不能改） `briefs` —— 3 条经营摘要：`[{"title","label","description","tone"}]`，title 固定为 当前判断/协同重点/需持续关注，tone ∈ blue/green/amber。 `insights` —— 2 条经营洞察：`[{"label","highlight","description"}]`。 `signals` —— 3 条经营信号：`[{"title","description","tone"}]`，tone ∈ positive/blue。 `watchItems` —— 2 条关注项：`[{"title","description","level","tone"}]`，level ∈ 中风险/关注，tone ∈ medium/low。 `pageTitle` —— 固定 "商业态势总览"。 `pageDescription` —— 一句话本期研判结论，格式："｛本期核心结论一句｝ 当前展示 ｛MM-DD HH:mm｝ 经营快照。"（时间用本期时刻）。 """ + SUBMIT_RULES + NO_ASK_RULES, }, ] CHAIN = { "id": CHAIN_ID, "title": "商业态势总览采集链", "description": "定时数据交付演示链：阶段1 财务/市场两席位并行采集，阶段2 综合研判席位基于前序交付出结论；全部经 task-data-submit 技能提交到任务 block_1。", "seats": [ {"agent_id": "bo-finance-seat", "name": "财务经营数据", "description": "产出 chart/operatingMetrics/headlineMetrics（财务经营类顶层 key）并入库"}, {"agent_id": "bo-market-seat", "name": "客户市场动态", "description": "产出 objectives/events/businessNarratives/priorities（动态信息类顶层 key）并入库"}, {"agent_id": "bo-synthesis-seat", "name": "综合研判", "description": "基于前序交付产出 briefs/insights/signals/watchItems 与页面元信息并入库"}, ], "stages": [["bo-finance-seat", "bo-market-seat"], ["bo-synthesis-seat"]], "stage_goals": [ "并行采集：财务席位交 chart/operatingMetrics/headlineMetrics，市场席位交 objectives/events/businessNarratives/priorities，各自用 task-data-submit 提交 block_1", "综合研判：基于前序两席位交付产出 briefs/insights/signals/watchItems + pageTitle/pageDescription 并提交", ], "coordinator_prompt": ( "派活时明确每个席位的顶层 key 边界：财务席位只交 chart/operatingMetrics/headlineMetrics，" "市场席位只交 objectives/events/businessNarratives/priorities，综合席位只交 " "briefs/insights/signals/watchItems/pageTitle/pageDescription。所有席位用 " "task-data-submit 技能提交到 block_1，--task-id 取任务上下文里的「数据交付 task_id」。" "无人值守，禁止澄清。" ), "enabled": True, } def call(method: str, path: str, body: dict | None, token: str): headers = {"Content-Type": "application/json", "Accept": "application/json"} if token: headers["Authorization"] = f"Bearer {token}" request = urllib.request.Request( BASE + path, data=json.dumps(body).encode() if body is not None else None, headers=headers, method=method) try: with urllib.request.urlopen(request, timeout=30) as response: return response.status, json.loads(response.read() or b"{}") except urllib.error.HTTPError as error: return error.code, json.loads(error.read() or b"{}") def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--skip-trigger", action="store_true", help="只建智能体/链/绑定，不手动触发执行") args = parser.parse_args() query = urllib.parse.urlencode({"password": PASSWORD}) status, login = call("POST", f"/api/v1/auth/login/username?{query}", {"username": USERNAME}, None) if status != 200: print("登录失败", status, login) return 1 token = login["access_token"] print("登录 OK") # 1. 智能体（幂等：存在则跳过） for agent in AGENTS: status, existing = call("GET", f"/api/agents/{agent['id']}", None, token) if status == 200: print(f"智能体已存在: {agent['id']}") continue body = {**agent, "published": True, "skills": ["task-data-submit"], "tool_groups": ["web", "file:read", "file:write", "bash"]} status, created = call("POST", "/api/agents", body, token) print(f"创建智能体 {agent['id']}: HTTP {status}", created.get("id") or created) if status not in (200, 201): return 1 # 2. 业务链（幂等：存在则跳过） status, existing_chain = call("GET", f"/api/roundtable-chains/{CHAIN_ID}", None, token) if status == 200: print(f"业务链已存在: {CHAIN_ID}") else: status, created = call("POST", "/api/roundtable-chains", CHAIN, token) print(f"创建业务链: HTTP {status}", created.get("id") or created) if status not in (200, 201): return 1 # 3. 发布链（is_public=True；已公开则跳过） status, chain = call("GET", f"/api/roundtable-chains/{CHAIN_ID}", None, token) if status == 200 and not chain.get("is_public"): status, updated = call("PUT", f"/api/roundtable-chains/{CHAIN_ID}", {"is_public": True}, token) print(f"发布业务链: HTTP {status}, is_public={updated.get('is_public')}, 自动发布席位={updated.get('published_seat_agents')}") else: print("业务链已是公开状态") # 4. 演示任务绑定链 + 每 10 分钟 status, task = call("GET", f"/api/task-analysis/external/tasks/{SOURCE_TASK_ID}/detail", None, token) if status != 200: print("演示任务不存在", status, task) return 1 internal_id = task["task"]["task_id"] binding = {"chain_id": CHAIN_ID, "cron": CRON, "timezone": "Asia/Shanghai", "enabled": True, "model": None, "coordinator_prompt": None} status, bound = call("PUT", f"/api/task-analysis/tasks/{internal_id}/binding", binding, token) print(f"绑定定时（{CRON}）: HTTP {status}, next_run_at={bound.get('binding', {}).get('next_run_at')}") if status not in (200, 201): print(bound) return 1 # 5. 立即触发一轮 if not args.skip_trigger: status, run = call("POST", f"/api/task-analysis/tasks/{internal_id}/trigger", {}, token) print(f"手动触发: HTTP {status}, run_id={run.get('run', {}).get('id')}, status={run.get('run', {}).get('status')}") print(f"\n内部任务 id: {internal_id}") print("查询执行: GET /api/task-analysis/tasks/{id}/runs") return 0 if __name__ == "__main__": sys.exit(main())