"""创建「商业情报收集链」3 个自定义席位智能体、业务链与定时绑定。 一次性联调脚本（可重复运行，幂等：已存在则更新/跳过）： 1. POST/PUT /api/agents ×3 —— 用户级智能体（非内置 seeder），各绑本区块信息技能； 2. POST /api/roundtable-chains —— 业务链 biz-intel-chain（三阶段串行，避免详情并发覆盖）； 3. PUT /api/roundtable-chains/{id} is_public=True —— 发布链（自动发布席位智能体）； 4. POST /api/task-analysis/external/tasks —— 建任务 003（商业情报定时收集）； 5. PUT /api/task-analysis/tasks/{id}/binding —— 绑定链 + cron 每小时整点； 6. 可选 `--trigger`：立即手动触发一轮（默认只配置，不触发生产接口）。 用法（backend 目录）：PYTHONPATH=. uv run python scripts/setup_force_situation_chain.py """ from __future__ import annotations import argparse import json import sys import urllib.error import urllib.parse import urllib.request if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") BASE = "http://localhost:8001" USERNAME = "admin" PASSWORD = "123ewq" SOURCE_TASK_ID = "003" CHAIN_ID = "biz-intel-chain" CRON = "0 * * * *" AGENTS = [ { "id": "biz-insight-seat", "name": "商业情报接口采集智能体", "description": "商业情报链阶段1：读取独立任务详情，调用四类监测接口并由脚本转换、原子合并六个来源区块。", "skills": [ "fs-collect", "fs-task-detail-info", "fs-timeline-info", "fs-insight-info", "fs-target-info", "fs-connection-info", ], "soul": """# 商业情报接口采集智能体（biz-insight-seat） 你是商业情报收集链阶段 1 的自定义接口采集智能体。只执行下面一条命令： ```bash python /mnt/skills/custom/fs-collect/scripts/fs_collect.py --group source --task-id <任务号> ``` `--task-id` 取任务上下文里「数据交付 task_id（调用 task-data-submit 技能入库时的 --task-id 参数）」一行。 - 成功（stdout `"success": true`）后报告已写入的 submitted_keys； - 失败时重试一次，仍失败把 errors 原样回复； - 禁止向用户提问，禁止改写接口数据。请求参数、放宽条件和原始响应由脚本直接显示。 """, }, { "id": "biz-plan-seat", "name": "当前判断接口采集智能体", "description": "商业情报链阶段2：通过独立当前判断技能调用威胁事件接口，转换并原子合并 currentJudgments。", "skills": ["fs-collect", "fs-task-detail-info", "fs-current-judgment-info"], "soul": """# 当前判断接口采集智能体（biz-plan-seat） 你是商业情报收集链阶段 2 的自定义接口智能体。只执行： ```bash python /mnt/skills/custom/fs-collect/scripts/fs_collect.py --group judgment --task-id <任务号> ``` 脚本通过独立技能调用 GET eventMonitorByTime，固定传 discriminant=威胁，逐步放宽条件， 转换并原子合并 currentJudgments。成功后报告 submitted_keys；失败重试一次后原样报告 errors。 禁止提问、禁止改写接口结果、禁止回退到固定池或网络检索。 """, }, { "id": "biz-assembler-seat", "name": "情报摘要与校验智能体", "description": "商业情报链阶段3：基于前两阶段数据由脚本生成三类摘要，校验八个区块和关键来源标识。", "skills": ["fs-collect", "fs-brief-info", "fs-detail-read"], "soul": """# 情报摘要与校验智能体（biz-assembler-seat） 你是商业情报收集链阶段 3 的自定义摘要与校验智能体。前序阶段已串行完成，先执行： ```bash python /mnt/skills/custom/fs-collect/scripts/fs_collect.py --group summary --task-id <任务号> ``` 该命令基于真实采集结果和当前判断生成情报摘要并原子合并入库。随后读取详情校验。 ## 8 个顶层 key `situationBriefs` / `currentJudgments` / `importantTargets` / `importantTimeNodes` / `insightTopics` / `insightRisks` / `insightOpportunities` / `taskConnections` ## 流程 1. 读取详情：`python /mnt/skills/custom/fs-detail-read/scripts/fs_detail_read.py --task-id <任务号>` 2. 比对 8 个 key（区块为空数组/缺失都算缺）； 3. 校验关键来源标识：时间节点保留 recUuid，三类事件保留 id/eventId，目标保留 eid，衔接任务保留 taskId； 4. 任一项缺失时明确报告，不伪造补齐；全部齐全后回复“校验通过（8/8 区块）”。 """, }, ] CHAIN = { "id": CHAIN_ID, "title": "商业情报收集链", "description": "商业态势页定时信息收集：来源接口采集、当前判断接口采集、情报摘要与校验三阶段串行，避免并发覆盖详情。", "seats": [ {"agent_id": "biz-insight-seat", "name": "接口数据采集", "description": "读取独立任务详情并采集时间节点、三类事件、重要目标和衔接任务"}, {"agent_id": "biz-plan-seat", "name": "当前判断采集", "description": "独立调用威胁事件接口并转换 currentJudgments"}, {"agent_id": "biz-assembler-seat", "name": "情报摘要与校验", "description": "由脚本生成三类摘要并校验八个区块及关键来源标识"}, ], "stages": [["biz-insight-seat"], ["biz-plan-seat"], ["biz-assembler-seat"]], "stage_goals": [ "接口采集：读取独立任务详情，依次调用监测接口并由脚本转换，六个来源区块一次原子合并入库", "当前判断：独立技能调用 eventMonitorByTime（discriminant=威胁），转换为 currentJudgments 后合并入库", "摘要与校验：基于全部已采集信息生成 situationBriefs，校验八个区块和关键来源 id", ], "coordinator_prompt": ( "严格按三个阶段串行执行：阶段 1 用 fs-collect --group source 采集接口数据；阶段 2 用 fs-collect --group judgment 采集当前判断；" "阶段 3 用 fs-collect --group summary 生成摘要并校验。接口数据由脚本转换，模型禁止改写。" "所有席位无人值守，禁止澄清。" ), "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(description="配置商业情报自定义智能体、业务链与定时绑定") parser.add_argument("--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") # 手工落盘的新技能保持 custom 分类；登记到当前用户并发布后才能绑定共享业务链席位。 status, reconciled = call("POST", "/api/skills/reconcile-custom", None, token) print(f"登记自定义技能: HTTP {status}, adopted={reconciled.get('adopted', [])}") if status not in (200, 201): return 1 for skill_name in ("fs-task-detail-info", "fs-current-judgment-info"): status, published = call( "PUT", f"/api/skills/custom/{skill_name}/published", {"published": True}, token, ) print(f"发布自定义技能 {skill_name}: HTTP {status}, published={published.get('published')}") if status not in (200, 201): return 1 # 0. 仅清理早期旧命名资源；不删除当前任务和执行历史。 for old_agent in ("fs-insight-seat", "fs-plan-seat", "fs-assembler-seat"): status, _ = call("DELETE", f"/api/agents/{old_agent}", None, token) print(f"清理旧智能体 {old_agent}: HTTP {status}") status, _ = call("DELETE", "/api/roundtable-chains/fs-situation-chain", None, token) print(f"清理旧业务链 fs-situation-chain: HTTP {status}") # 1. 智能体（幂等：存在则 PUT 同步，不存在则 POST 创建） for agent in AGENTS: body = { **{k: agent[k] for k in ("id", "name", "description", "soul")}, "published": True, "skills": agent["skills"], "tool_groups": ["web", "file:read", "file:write", "bash"], } status, existing = call("GET", f"/api/agents/{agent['id']}", None, token) if status == 200: status, updated = call("PUT", f"/api/agents/{agent['id']}", body, token) print(f"更新智能体 {agent['id']}: HTTP {status}") else: status, created = call("POST", "/api/agents", body, token) print(f"创建智能体 {agent['id']}: HTTP {status}") if status not in (200, 201): print(body if status != 200 else updated) return 1 # 2. 业务链（幂等：存在则 PUT 同步阶段目标等编排字段，不存在则创建） status, existing_chain = call("GET", f"/api/roundtable-chains/{CHAIN_ID}", None, token) if status == 200: status, updated_chain = call( "PUT", f"/api/roundtable-chains/{CHAIN_ID}", { "title": CHAIN["title"], "description": CHAIN["description"], "seats": CHAIN["seats"], "stages": CHAIN["stages"], "stage_goals": CHAIN["stage_goals"], "coordinator_prompt": CHAIN["coordinator_prompt"], "enabled": True, }, token, ) print(f"更新业务链: HTTP {status}") if status not in (200, 201): print(updated_chain) return 1 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. 发布链 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}, 自动发布席位={updated.get('published_seat_agents')}") else: print("业务链已是公开状态") # 4. 任务 003（幂等：已存在跳过） status, detail = call("GET", f"/api/task-analysis/external/tasks/{SOURCE_TASK_ID}/detail", None, token) if status == 200: internal_id = detail["task"]["task_id"] print(f"任务已存在: {SOURCE_TASK_ID} → {internal_id}") else: status, created = call( "POST", "/api/task-analysis/external/tasks", {"source_task_id": SOURCE_TASK_ID, "name": "商业情报定时收集", "description": "商业态势页定时信息收集任务（每小时）"}, token, ) print(f"创建任务: HTTP {status}, task_id={created.get('task_id')}") if status not in (200, 201): print(created) return 1 internal_id = created["task_id"] # 5. 绑定链 + 每小时整点 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 # 6. 默认不触发，避免真实监测接口尚未上线时产生失败执行。 if args.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')}") else: print("未触发执行；需要立即运行时追加 --trigger") print(f"\n内部任务 id: {internal_id}") print(f"查询执行: GET /api/task-analysis/tasks/{internal_id}/runs") print(f"详情验证: GET /api/force-situation/tasks/{SOURCE_TASK_ID}/detail") return 0 if __name__ == "__main__": sys.exit(main())