151 lines
5.3 KiB
Python
151 lines
5.3 KiB
Python
"""更新「商业QB助理」智能体并绑定 3q-0 配置位。
|
|
|
|
SOUL 从当前 DeerFlow 智能体目录读取，避免维护两份行为规则。运行前应先部署
|
|
``skills/custom/task-context``、``skills/custom/task-detail-update`` 和智能体 SOUL。
|
|
|
|
用法（backend 目录）：PYTHONPATH=. uv run python scripts/setup_biz_detail_assistant.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
if hasattr(sys.stdout, "reconfigure"):
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
|
|
BASE = os.environ.get("DEERFLOW_SETUP_BASE", "http://localhost:8001").rstrip("/")
|
|
USERNAME = os.environ.get("DEERFLOW_SETUP_USERNAME", "admin")
|
|
PASSWORD = os.environ.get("DEERFLOW_SETUP_PASSWORD", "123ewq")
|
|
AGENT_ID = "biz-detail-assistant"
|
|
AGENT_NAME = "商业QB助理（全量检索版）"
|
|
MAPPING_CODE = "3q-0"
|
|
SKILLS = ["task-context", "task-detail-update"]
|
|
TOOL_GROUPS = ["web", "file:read", "file:write", "bash"]
|
|
DESCRIPTION = (
|
|
"任务详情页右侧问答智能体：结合用户输入、首轮任务四要素与相关上传文件进行分析或检索；"
|
|
"“态势研判”固定覆盖态势、风险、战机三模块。情报摘要基于任务信息、相关文件证据与六个"
|
|
"现有模块的压缩证据生成敌情、我情、环境三条快照。方案确认后原子写回，任务衔接不在范围内。"
|
|
)
|
|
|
|
AGENT_DIR = Path(__file__).resolve().parents[1] / ".deer-flow" / "agents" / AGENT_ID
|
|
SOUL_PATH = AGENT_DIR / "SOUL.md"
|
|
|
|
|
|
def call(method: str, path: str, body: dict | None, token: str | None, *, form: dict | None = None):
|
|
if body is not None and form is not None:
|
|
raise ValueError("请求不能同时包含 JSON 与表单数据")
|
|
headers = {"Accept": "application/json"}
|
|
if form is not None:
|
|
headers["Content-Type"] = "application/x-www-form-urlencoded"
|
|
data = urllib.parse.urlencode(form).encode()
|
|
else:
|
|
headers["Content-Type"] = "application/json"
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
if token:
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
request = urllib.request.Request(
|
|
BASE + path,
|
|
data=data,
|
|
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:
|
|
raw = error.read().decode("utf-8", "replace")
|
|
try:
|
|
return error.code, json.loads(raw or b"{}")
|
|
except json.JSONDecodeError:
|
|
return error.code, {"raw": raw}
|
|
|
|
|
|
def main() -> int:
|
|
if not SOUL_PATH.is_file():
|
|
print(f"缺少当前智能体 SOUL，拒绝用旧模板覆盖：{SOUL_PATH}")
|
|
return 1
|
|
soul = SOUL_PATH.read_text(encoding="utf-8")
|
|
|
|
status, login = call(
|
|
"POST",
|
|
"/api/v1/auth/login/local",
|
|
None,
|
|
None,
|
|
form={"username": USERNAME, "password": PASSWORD},
|
|
)
|
|
# 与两个任务详情技能保持一致：旧部署关闭密码登录、但仍开启 SSO 兼容直登时，
|
|
# 仅对策略拒绝（403）回退；凭据错误、限流等失败不能绕过账号密码校验。
|
|
if status == 403:
|
|
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")
|
|
|
|
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 SKILLS:
|
|
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
|
|
|
|
body = {
|
|
"id": AGENT_ID,
|
|
"name": AGENT_NAME,
|
|
"description": DESCRIPTION,
|
|
"soul": soul,
|
|
"published": True,
|
|
"skills": SKILLS,
|
|
"tool_groups": TOOL_GROUPS,
|
|
}
|
|
status, _existing = call("GET", f"/api/agents/{AGENT_ID}", None, token)
|
|
if status == 200:
|
|
status, agent = call("PUT", f"/api/agents/{AGENT_ID}", body, token)
|
|
action = "更新"
|
|
else:
|
|
status, agent = call("POST", "/api/agents", body, token)
|
|
action = "创建"
|
|
print(f"{action}智能体 {AGENT_ID}: HTTP {status}, skills={agent.get('skills')}")
|
|
if status not in (200, 201):
|
|
print(agent)
|
|
return 1
|
|
|
|
status, mapping = call(
|
|
"PUT",
|
|
f"/api/business-mapping/{MAPPING_CODE}",
|
|
{"agent_id": AGENT_ID, "agent_name": AGENT_NAME},
|
|
token,
|
|
)
|
|
print(f"绑定配置位 {MAPPING_CODE}: HTTP {status}, agent_id={mapping.get('agent_id')}")
|
|
if status not in (200, 201):
|
|
print(mapping)
|
|
return 1
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|