280 lines
13 KiB
Python
280 lines
13 KiB
Python
"""创建「商业情报收集链」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())
|