From 75cfd52c21d68a1a294b46ff4014c83ecb284bf8 Mon Sep 17 00:00:00 2001 From: ymq Date: Wed, 2 Sep 2026 15:07:23 +0800 Subject: [PATCH] =?UTF-8?q?feat(platform=5Fability):=20=E5=B9=B3=E5=8F=B0?= =?UTF-8?q?=E5=86=85=E9=83=A8agent=E8=83=BD=E5=8A=9B=E5=8C=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 5工具: 模型治理状态/抓取文档/提取规格/写入配置(幂等)/模块清单 - 权限: 代码层硬门禁仅owner.superuser|admin,不依赖prompt - 模型自动配置: 通读API文档→生成供应商端点/适配模板/模型四价 - 定价只来自文档原文,禁止编造;单位统一元/千tokens --- pipeline_service/__init__.py | 1 + pipeline_service/platform_ability.py | 521 +++++++++++++++++++++++++++ 2 files changed, 522 insertions(+) create mode 100644 pipeline_service/platform_ability.py diff --git a/pipeline_service/__init__.py b/pipeline_service/__init__.py index d53dfc5..53420b7 100644 --- a/pipeline_service/__init__.py +++ b/pipeline_service/__init__.py @@ -35,6 +35,7 @@ from .agent_loop_v2 import AgentExecutor, run_agent # 产线能力包注册（import 即注册到 pipeline_core 的 PipelineAbility 注册表） from . import sdlc_ability # noqa: F401 +from . import platform_ability # noqa: F401（平台内部 agent：模型自动配置等） # 通用 slash 命令注册（import 即注册） from . import slash_commands # noqa: F401 diff --git a/pipeline_service/platform_ability.py b/pipeline_service/platform_ability.py new file mode 100644 index 0000000..40fd582 --- /dev/null +++ b/pipeline_service/platform_ability.py @@ -0,0 +1,521 @@ +""" +pipeline-service: platform_ability — 平台内部 agent 能力包（pipeline_id=platform_general） + +产线平台自身的运维/管理 agent：拥有平台应用和各模块的技能， +核心能力是「通读模型 API 文档 → 自动生成模型治理配置（供应商/适配模板/模型/定价）」。 + +权限模型（代码层硬门禁，不依赖 prompt）： + 所有工具 handler 入口校验调用者的 RBAC 角色 ∈ {owner.superuser, owner.admin}。 + 内部 agent 只服务管理员——普通用户即使打开页面也调不动任何工具。 + +配置生成链路（用户给定 API 文档 → 完成配置）： + ① fetch_model_doc：抓取文档页面（SSRF 防护：仅 http(s) 公网域名） + ② extract_llm_api_spec：LLM 从文档提取端点/协议/请求响应格式/定价 + ③ apply_llm_config：写库——复用 pipeline-llm 模块的 CRUD 函数 + （llm_vendor.endpoints + llm_api_profile 模板 + llm_model 含四价） +""" + +import json +import logging +import re + +from pipeline_core import ( + ToolDefinition, + PipelineAbility, + register_ability, +) + +logger = logging.getLogger("pipeline.platform_ability") + +PLATFORM_PIPELINE_ID = "platform_general" + +# 管理角色：内部 agent 仅管理员可用 +ADMIN_ROLES = ("owner.superuser", "owner.admin") + +# 当前运行时调用链支持的协议（llm_bridge 固定 OpenAI 兼容路径）。 +# 其他协议的适配模板会存入 llm_api_profile 备查，但运行时暂不渲染。 +RUNTIME_PROTOCOLS = ("openai_compat",) + + +# ────────────────────── 权限门禁（代码层） ────────────────────── + +async def _require_admin(sor, ctx) -> str: + """校验当前用户是管理员。返回 ''=通过，否则错误信息。""" + uid = ctx.get("user_id", "") or "" + if not uid: + return "无法识别当前用户身份（未登录），内部 agent 仅管理员可用" + recs = await sor.sqlExe( + "SELECT r.orgtypeid, r.name FROM userrole ur JOIN role r ON ur.roleid=r.id " + "WHERE ur.userid=${u}$", {"u": uid}) + await sor.sqlExe("COMMIT", {}) + roles = set() + for r in (recs or []): + o = getattr(r, "orgtypeid", "") or "" + n = getattr(r, "name", "") or "" + if o and n: + roles.add("%s.%s" % (o, n)) + if roles & set(ADMIN_ROLES): + return "" + return "权限不足：内部 agent 工具仅管理员（%s）可用，当前角色 %s" % ( + "/".join(ADMIN_ROLES), sorted(roles) or "无") + + +def _row(r): + """sqlor row → dict（容忍属性访问）。""" + if hasattr(r, "__dict__"): + return {k: v for k, v in r.__dict__.items() if not k.startswith("_")} + return {} + + +# ────────────────────── 工具 1：模型治理状态 ────────────────────── + +async def _h_platform_llm_status(sor, params, ctx): + """供应商/账号/模型/用量概览（管理员诊断配置用）。""" + err = await _require_admin(sor, ctx) + if err: + return err + out = {} + for tbl in ("llm_vendor", "llm_account", "llm_model"): + recs = await sor.sqlExe( + "SELECT status, COUNT(*) AS c FROM " + tbl + " GROUP BY status", {}) + await sor.sqlExe("COMMIT", {}) + out[tbl] = {getattr(r, "status", ""): int(getattr(r, "c", 0)) for r in (recs or [])} + accs = await sor.sqlExe( + "SELECT name, balance, status FROM llm_account ORDER BY balance DESC LIMIT 20", {}) + await sor.sqlExe("COMMIT", {}) + out["accounts"] = [ + {"name": getattr(r, "name", ""), "balance": float(getattr(r, "balance", 0) or 0), + "status": getattr(r, "status", "")} for r in (accs or [])] + models = await sor.sqlExe( + "SELECT name, vendor_model_id, capability, price_input, price_output, status " + "FROM llm_model ORDER BY created_at DESC LIMIT 30", {}) + await sor.sqlExe("COMMIT", {}) + out["models"] = [ + {"name": getattr(r, "name", ""), "vendor_model_id": getattr(r, "vendor_model_id", ""), + "capability": getattr(r, "capability", ""), + "price_input": float(getattr(r, "price_input", 0) or 0), + "price_output": float(getattr(r, "price_output", 0) or 0), + "status": getattr(r, "status", "")} for r in (models or [])] + return json.dumps(out, ensure_ascii=False) + + +# ────────────────────── 工具 2：抓取模型 API 文档 ────────────────────── + +_MAX_DOC_CHARS = 60000 + + +def _validate_doc_url(url: str) -> str: + """SSRF 防护：仅允许 http(s) + 公网域名。返回 ''=通过，否则错误。""" + if not re.match(r"^https?://", url or ""): + return "文档 URL 必须是 http/https 链接" + host = re.sub(r"^https?://", "", url).split("/")[0].split(":")[0].lower() + if not re.match(r"^[a-z0-9][a-z0-9.-]+\.[a-z]{2,}$", host): + return "文档 URL 域名非法（不接受 IP/内网地址）" + if host in ("localhost",) or host.endswith((".local", ".internal", ".localhost")): + return "文档 URL 不允许指向内网/本地地址" + return "" + + +def _html_to_text(html: str) -> str: + """粗暴去标签提取文本（文档页面用，不追求完美排版）。""" + txt = re.sub(r"(?is)<(script|style|noscript)[^>]*>.*?", " ", html or "") + txt = re.sub(r"(?is)<[^>]+>", " ", txt) + txt = re.sub(r" ", " ", txt) + txt = re.sub(r"<", "<", txt) + txt = re.sub(r">", ">", txt) + txt = re.sub(r"&", "&", txt) + txt = re.sub(r""", '"', txt) + txt = re.sub(r"[ \t]+", " ", txt) + txt = re.sub(r"\n\s*\n+", "\n", txt) + return txt.strip() + + +async def _h_fetch_model_doc(sor, params, ctx): + """抓取模型 API 文档页面 → 纯文本（供 LLM 提取配置规格）。""" + err = await _require_admin(sor, ctx) + if err: + return err + url = (params.get("url") or "").strip() + verr = _validate_doc_url(url) + if verr: + return verr + import aiohttp + try: + timeout = aiohttp.ClientTimeout(total=30) + async with aiohttp.ClientSession(timeout=timeout) as sess: + async with sess.get(url, headers={"User-Agent": "Mozilla/5.0"}, + allow_redirects=True, ssl=False) as resp: + if resp.status != 200: + return "抓取失败：HTTP %d" % resp.status + ctype = resp.headers.get("Content-Type", "") + raw = await resp.text(errors="replace") + except Exception as e: + return "抓取失败：%s" % str(e)[:200] + if "html" in ctype.lower() or raw.lstrip()[:15].lower().startswith((" _MAX_DOC_CHARS: + text = text[:_MAX_DOC_CHARS] + "\n[文档过长已截断，共 %d 字符]" % len(text) + return text + + +# ────────────────────── 工具 3：提取配置规格（LLM） ────────────────────── + +_EXTRACT_PROMPT = """你是大模型 API 配置专家。通读下面这份模型 API 文档，提取配置规格。 + +只输出一个 JSON 对象（不要 markdown 代码块），字段： +{ + "vendor_name": "供应商名称", + "base_url": "API 基础地址（如 https://dashscope.aliyuncs.com/compatible-mode/v1）", + "protocol": "openai_compat | dashscope_async | custom（能走 /chat/completions 的填 openai_compat）", + "endpoints": [{"base_url": "...", "region": "domestic|international", "timeout": 60}], + "chat_path": "对话接口路径（如 /chat/completions）", + "auth_header": "认证头格式说明（如 Bearer API_KEY）", + "request_fields": ["请求体字段名列表"], + "response_format": "响应格式说明（content 字段路径 + usage 字段路径）", + "models": [{"vendor_model_id": "供应商侧模型ID", "capability": "t2t|t2i|i2t|t2v|embedding|rerank|tts|asr", + "price_input": 输入价元/千token或null, "price_output": 输出价元/千token或null, + "cost_input": 输入成本元/千token或null, "cost_output": 输出成本元/千token或null, + "description": "一句话说明"}], + "doc_notes": "文档中影响配置的关键注意点" +} + +定价规则： +- 文档给的单位若是「元/百万tokens」，换算为「元/千tokens」= 原价/1000，保留6位小数 +- 文档没写价格的模型：价格字段填 null（禁止编造价格），在 description 注明 +- cost_* 未知时与 price_* 相同（无供应商折扣时成本=售价） + +文档内容： +""" + + +async def _h_extract_llm_api_spec(sor, params, ctx): + """LLM 通读文档文本 → 结构化配置规格（JSON）。""" + err = await _require_admin(sor, ctx) + if err: + return err + doc_text = (params.get("doc_text") or "").strip() + if len(doc_text) < 100: + return "文档文本太短（<100字符），无法提取配置——先用 fetch_model_doc 抓取" + try: + from .llm_bridge import llm_call_msgs + raw = await llm_call_msgs( + [{"role": "system", "content": _EXTRACT_PROMPT}, + {"role": "user", "content": doc_text[:48000]}], + temperature=0.1, org_id="0", user_id=ctx.get("user_id", "")) + except Exception as e: + return "LLM 提取失败：%s" % str(e)[:200] + txt = (raw or "").strip() + if txt.startswith("```"): + txt = txt.split("\n", 1)[1] if "\n" in txt else txt + txt = txt.rsplit("```", 1)[0] + try: + spec = json.loads(txt) + except Exception: + m = re.search(r"\{.*\}", txt, re.S) + if not m: + return "LLM 输出不是合法 JSON：%s" % txt[:300] + try: + spec = json.loads(m.group(0)) + except Exception: + return "LLM 输出不是合法 JSON：%s" % txt[:300] + return json.dumps(spec, ensure_ascii=False) + + +# ────────────────────── 工具 4：写入配置（幂等） ────────────────────── + +def _f(v, default=0.0): + try: + return float(v) + except (TypeError, ValueError): + return default + + +async def _h_apply_llm_config(sor, params, ctx): + """按提取的规格写库：供应商（复用/新建）+ 适配模板 + 模型（含定价）。 + + 幂等：供应商按名称复用；模型按 (vendor, vendor_model_id) 复用—— + 已存在则更新定价，不重复创建。 + """ + err = await _require_admin(sor, ctx) + if err: + return err + try: + spec = json.loads(params.get("spec") or "{}") + except Exception: + return "spec 不是合法 JSON" + vendor_name = (spec.get("vendor_name") or "").strip() + if not vendor_name: + return "spec 缺少 vendor_name" + endpoints = spec.get("endpoints") or [] + if not endpoints and spec.get("base_url"): + endpoints = [{"base_url": spec["base_url"], "region": "domestic", "timeout": 60}] + if not endpoints: + return "spec 缺少 endpoints/base_url" + protocol = (spec.get("protocol") or "openai_compat").strip() or "openai_compat" + models = spec.get("models") or [] + if not models: + return "spec.models 为空——文档里没有可配置的模型？" + + from appPublic.uniqueID import getID + + # 1. 供应商：按名称复用 + recs = await sor.sqlExe( + "SELECT id, endpoints FROM llm_vendor WHERE name=${n}$", {"n": vendor_name}) + await sor.sqlExe("COMMIT", {}) + if recs: + vendor_id = getattr(recs[0], "id", "") + eps_old = [] + try: + eps_old = json.loads(getattr(recs[0], "endpoints", "") or "[]") + except Exception: + eps_old = [] + merged = list(eps_old) + added_eps = [] + for ep in endpoints: + bu = (ep.get("base_url") or "").rstrip("/") + if bu and bu not in [(e.get("base_url") or "").rstrip("/") for e in merged]: + merged.append({"base_url": bu, "region": ep.get("region") or "domestic", + "timeout": int(_f(ep.get("timeout"), 60))}) + added_eps.append(bu) + await sor.sqlExe( + "UPDATE llm_vendor SET endpoints=${e}$, protocol=${p}$, updated_at=NOW() " + "WHERE id=${i}$", + {"e": json.dumps(merged, ensure_ascii=False), "p": protocol, "i": vendor_id}) + await sor.sqlExe("COMMIT", {}) + vendor_action = "复用供应商 %s（新增端点 %d 个）" % (vendor_id, len(added_eps)) + else: + vendor_id = getID() + eps_norm = [{"base_url": (ep.get("base_url") or "").rstrip("/"), + "region": ep.get("region") or "domestic", + "timeout": int(_f(ep.get("timeout"), 60))} for ep in endpoints] + await sor.C("llm_vendor", { + "id": vendor_id, "name": vendor_name, "protocol": protocol, + "endpoints": json.dumps(eps_norm, ensure_ascii=False), + "description": spec.get("doc_notes", "") or "", + "status": "active", "org_id": "0"}) + vendor_action = "新建供应商 %s（%s）" % (vendor_id, vendor_name) + + # 2. 适配模板：按 (协议×能力) 复用，缺则新建 + profile_ids = {} + caps = sorted(set((m.get("capability") or "t2t") for m in models)) + for cap in caps: + recs = await sor.sqlExe( + "SELECT id FROM llm_api_profile WHERE protocol=${p}$ AND capability=${c}$ " + "AND status='active' ORDER BY created_at DESC LIMIT 1", + {"p": protocol, "c": cap}) + await sor.sqlExe("COMMIT", {}) + if recs: + profile_ids[cap] = getattr(recs[0], "id", "") + continue + req_tpl, resp_tpl = _default_templates(protocol, cap, spec) + pid = getID() + await sor.C("llm_api_profile", { + "id": pid, "name": "%s-%s-自动配置" % (protocol, cap), + "protocol": protocol, "capability": cap, + "request_template": req_tpl, "response_template": resp_tpl, + "param_schema": json.dumps([{"name": "prompt", "label": "提示词", + "uitype": "textarea", "required": True}], + ensure_ascii=False), + "status": "active"}) + profile_ids[cap] = pid + + # 3. 模型：按 (vendor, vendor_model_id) 幂等 + created, updated, skipped = [], [], [] + for m in models: + vmid = (m.get("vendor_model_id") or "").strip() + if not vmid: + skipped.append("(缺 vendor_model_id)") + continue + cap = m.get("capability") or "t2t" + price_in, price_out = _f(m.get("price_input")), _f(m.get("price_output")) + cost_in = _f(m.get("cost_input"), price_in) + cost_out = _f(m.get("cost_output"), price_out) + recs = await sor.sqlExe( + "SELECT id FROM llm_model WHERE vendor_id=${v}$ AND vendor_model_id=${m}$", + {"v": vendor_id, "m": vmid}) + await sor.sqlExe("COMMIT", {}) + if recs: + mid = getattr(recs[0], "id", "") + await sor.sqlExe( + "UPDATE llm_model SET price_input=${pi}$, price_output=${po}$, " + "cost_input=${ci}$, cost_output=${co}$, description=${d}$, updated_at=NOW() " + "WHERE id=${i}$", + {"pi": price_in, "po": price_out, "ci": cost_in, "co": cost_out, + "d": m.get("description", "") or "", "i": mid}) + await sor.sqlExe("COMMIT", {}) + updated.append(vmid) + continue + mid = getID() + await sor.C("llm_model", { + "id": mid, "vendor_id": vendor_id, "account_id": "", + "name": vmid, "vendor_model_id": vmid, "capability": cap, + "sync_mode": "sync", "profile_id": profile_ids.get(cap, ""), + "ppid": "", "price_input": price_in, "price_output": price_out, + "cost_input": cost_in, "cost_output": cost_out, + "default_params": "{}", "status": "active", + "description": m.get("description", "") or "", "org_id": "0"}) + created.append(vmid) + + rt_note = "" + if protocol not in RUNTIME_PROTOCOLS: + rt_note = ("⚠️ 协议「%s」的适配模板已存档，但当前运行时调用链仅支持 %s——" + "该供应商模型暂不可被产线直接调用" % (protocol, "/".join(RUNTIME_PROTOCOLS))) + return json.dumps({ + "vendor": vendor_action, "profiles": profile_ids, + "models_created": created, "models_updated": updated, "models_skipped": skipped, + "runtime_note": rt_note, + }, ensure_ascii=False) + + +def _default_templates(protocol: str, capability: str, spec: dict) -> tuple: + """生成适配模板默认值（headers/data/response）。 + + OpenAI 兼容协议用平台标准模板；其他协议按文档格式生成骨架供人工微调。 + """ + chat_path = (spec.get("chat_path") or "/chat/completions").strip() + if protocol == "openai_compat": + req = json.dumps({ + "path": chat_path, "method": "POST", + "headers": {"Authorization": "Bearer {{api_key}}", + "Content-Type": "application/json"}, + "params": {}, + "data": {"model": "{{model}}", "messages": "{{messages}}", + "temperature": "{{temperature}}", "stream": False}, + }, ensure_ascii=False) + resp = json.dumps({ + "content": "choices[0].message.content", + "usage": {"prompt_tokens": "usage.prompt_tokens", + "completion_tokens": "usage.completion_tokens"}, + }, ensure_ascii=False) + return req, resp + # 非标准协议：骨架模板（字段路径来自文档提取），标注需人工核对 + req = json.dumps({ + "path": chat_path, "method": "POST", + "headers": {"Authorization": "Bearer {{api_key}}", + "Content-Type": "application/json"}, + "params": {}, + "data": {"__from_doc__": spec.get("request_fields") or []}, + "__note__": "非 openai_compat 协议骨架模板，需按文档人工核对", + }, ensure_ascii=False) + resp = json.dumps({ + "content": spec.get("response_format", "") or "待按文档填写", + "__note__": "需人工核对", + }, ensure_ascii=False) + return req, resp + + +# ────────────────────── 工具 5：模块/应用信息查询 ────────────────────── + +async def _h_platform_modules(sor, params, ctx): + """列出平台已装载的业务模块（内部 agent 了解平台构成用）。""" + err = await _require_admin(sor, ctx) + if err: + return err + mods = [] + for name, title in [ + ("pipeline_core", "产线核心（会话/技能/项目）"), + ("pipeline_service", "执行引擎（任务链/角色/治理）"), + ("pipeline-llm", "模型治理（供应商/账号/模型/定价/记账）"), + ("pipeline-bidding", "投标产线"), + ("pipeline-opportunity", "商机产线"), + ("pipeline-sdlc", "开发产线前端"), + ("pipeline-ops", "运维"), + ("pipeline-dist", "分销"), + ("pipeline-task", "任务中心"), + ]: + try: + __import__(name.replace("-", "_")) + loaded = True + except Exception: + loaded = False + mods.append({"module": name, "title": title, "loaded": loaded}) + return json.dumps(mods, ensure_ascii=False) + + +# ────────────────────── 工具定义 ────────────────────── + +PLATFORM_TOOLS = [ + ToolDefinition( + name="platform_llm_status", + description="查看模型治理状态：供应商/账号余额/模型定价/各状态分布。用户问「模型配置状态/账号余额」时调用。仅管理员可用。", + parameters={}, + category="platform", + ), + ToolDefinition( + name="fetch_model_doc", + description="抓取大模型供应商的 API 文档页面（返回纯文本）。配置新模型前先用此抓取官方文档。仅管理员可用。", + parameters={"url": "文档页面 URL（必须是公网 http/https）"}, + category="platform", + ), + ToolDefinition( + name="extract_llm_api_spec", + description="通读文档文本，LLM 提取 API 配置规格（端点/协议/请求响应格式/定价）。配合 fetch_model_doc 使用。仅管理员可用。", + parameters={"doc_text": "fetch_model_doc 返回的文档文本"}, + category="platform", + ), + ToolDefinition( + name="apply_llm_config", + description="按提取的规格写入模型治理配置：供应商/端点/适配模板/模型（含定价）。幂等——已有模型只更新定价。仅管理员可用。", + parameters={"spec": "extract_llm_api_spec 返回的 JSON 规格"}, + category="platform", + requires_confirmation=True, + ), + ToolDefinition( + name="platform_modules", + description="列出平台已装载的业务模块清单。用户问「平台有哪些模块」时调用。仅管理员可用。", + parameters={}, + category="platform", + ), +] + +PLATFORM_PROMPT = """ +你是产线平台的内部运维 agent，服务对象是平台管理员。 + +## 模型自动配置工作流（用户给你文档 URL 要求配置模型时） +1. `fetch_model_doc` 抓取官方文档页面 +2. `extract_llm_api_spec` 通读文档提取配置规格——**定价只允许来自文档原文**，文档没写的价格一律 null，禁止编造 +3. 把提取结果摘要给用户确认（尤其定价和端点），用户确认后 +4. `apply_llm_config` 写入（幂等：已有模型只更新定价） + +## 硬规则 +- 定价单位统一「元/千tokens」（文档给百万价要 ÷1000） +- cost_*（供应商成本）未知时与 price_* 相同 +- 非 openai_compat 协议的模型：模板会存档但运行时暂不可调用，必须如实告知 +- 文档抓取失败/内容不足时如实说明，不要凭记忆编造 API 格式 +- 所有工具仅管理员可用，权限报错时如实转告用户 + +## 平台知识 +平台模块清单用 `platform_modules` 查询；技能库中「平台」相关技能（skills_library/all/ 与 +pipelines/platform_general/）包含各模块的开发规范与踩坑记录，需要时用 load_skill 加载。 +""" + +PLATFORM_HANDLERS = { + "platform_llm_status": _h_platform_llm_status, + "fetch_model_doc": _h_fetch_model_doc, + "extract_llm_api_spec": _h_extract_llm_api_spec, + "apply_llm_config": _h_apply_llm_config, + "platform_modules": _h_platform_modules, +} + + +def register_platform_ability(): + """注册平台内部 agent 能力包（幂等）。""" + ability = PipelineAbility( + pipeline_id=PLATFORM_PIPELINE_ID, + name="平台内部 agent", + tools=PLATFORM_TOOLS, + system_prompt=PLATFORM_PROMPT, + handlers=PLATFORM_HANDLERS, + roles=[], + menus=[ + {"label": "📊 模型治理", "icon": "", "url": "/pipeline-llm", + "type": "tab"}, + ], + ) + register_ability(ability) + return ability + + +register_platform_ability()