feat(platform): 提取prompt补async_steps/sync_mode;apply_llm_config写profile的path/headers列+异步模型建后续步骤模板链登记query_profile_ids

This commit is contained in:
yumoqing 2026-09-04 14:00:16 +08:00
parent 92e7642bb3
commit 8bb9a0c682

View File

@ -20,6 +20,8 @@ import json
import logging import logging
import re import re
from appPublic.uniqueID import getID
from pipeline_core import ( from pipeline_core import (
ToolDefinition, ToolDefinition,
PipelineAbility, PipelineAbility,
@ -223,13 +225,20 @@ _EXTRACT_PROMPT = """你是大模型 API 配置专家。通读下面这份模型
"auth_header": "认证头格式说明(如 Bearer API_KEY)", "auth_header": "认证头格式说明(如 Bearer API_KEY)",
"request_fields": ["请求体字段名列表"], "request_fields": ["请求体字段名列表"],
"response_format": "响应格式说明(content 字段路径 + usage 字段路径)", "response_format": "响应格式说明(content 字段路径 + usage 字段路径)",
"async_steps": [{"purpose": "query|download", "path": "该步骤接口路径", "method": "GET|POST", "request_fields": ["请求字段名列表"], "response_format": "该步骤响应格式说明"}],
"models": [{"vendor_model_id": "供应商侧模型ID", "capability": "t2t|t2i|i2t|t2v|embedding|rerank|tts|asr", "models": [{"vendor_model_id": "供应商侧模型ID", "capability": "t2t|t2i|i2t|t2v|embedding|rerank|tts|asr",
"sync_mode": "sync|async(提交后需轮询查询结果的填 async)",
"price_input": 输入价元/千token或null, "price_output": 输出价元/千token或null, "price_input": 输入价元/千token或null, "price_output": 输出价元/千token或null,
"cost_input": 输入成本元/千token或null, "cost_output": 输出成本元/千token或null, "cost_input": 输入成本元/千token或null, "cost_output": 输出成本元/千token或null,
"description": "一句话说明"}], "description": "一句话说明"}],
"doc_notes": "文档中影响配置的关键注意点" "doc_notes": "文档中影响配置的关键注意点"
} }
异步模型规则:
- 文档描述「先提交任务、再轮询查询结果」的模型,sync_mode 填 async,并提取 async_steps:
至少一条 purpose=query(查询任务状态/结果);若文档另有独立下载/取文件接口,再加一条 purpose=download。
- 同步模型 async_steps 填空数组 []。
定价规则: 定价规则:
- 价格按文档原文记录,并保留文档标注的单位(在 description 中写明,如「元/百万tokens」) - 价格按文档原文记录,并保留文档标注的单位(在 description 中写明,如「元/百万tokens」)
- 禁止自行换算、推导或统一单位——单位是否吻合由上线测试阶段的记账核对验证 - 禁止自行换算、推导或统一单位——单位是否吻合由上线测试阶段的记账核对验证
@ -364,7 +373,9 @@ async def _h_apply_llm_config(sor, params, ctx):
await sor.C("llm_api_profile", { await sor.C("llm_api_profile", {
"id": pid, "name": "%s-%s-自动配置" % (protocol, cap), "id": pid, "name": "%s-%s-自动配置" % (protocol, cap),
"protocol": protocol, "capability": cap, "protocol": protocol, "capability": cap,
"request_template": req_tpl, "response_template": resp_tpl, "path": req_tpl["path"],
"headers": json.dumps(req_tpl["headers"], ensure_ascii=False),
"request_template": req_tpl["req"], "response_template": req_tpl["resp"],
"param_schema": json.dumps([{"name": "prompt", "label": "提示词", "param_schema": json.dumps([{"name": "prompt", "label": "提示词",
"uitype": "textarea", "required": True}], "uitype": "textarea", "required": True}],
ensure_ascii=False), ensure_ascii=False),
@ -379,6 +390,7 @@ async def _h_apply_llm_config(sor, params, ctx):
skipped.append("(缺 vendor_model_id)") skipped.append("(缺 vendor_model_id)")
continue continue
cap = m.get("capability") or "t2t" cap = m.get("capability") or "t2t"
sync_mode = "async" if str(m.get("sync_mode") or "").strip() == "async" else "sync"
raw_pi, raw_po = m.get("price_input"), m.get("price_output") raw_pi, raw_po = m.get("price_input"), m.get("price_output")
price_in, price_out = _f(raw_pi), _f(raw_po) price_in, price_out = _f(raw_pi), _f(raw_po)
cost_in = _f(m.get("cost_input"), price_in) cost_in = _f(m.get("cost_input"), price_in)
@ -387,6 +399,17 @@ async def _h_apply_llm_config(sor, params, ctx):
# 文档未给价(null)→ 落 0 并显式标注,避免「0=免费」歧义 # 文档未给价(null)→ 落 0 并显式标注,避免「0=免费」歧义
if raw_pi is None or raw_po is None: if raw_pi is None or raw_po is None:
desc = "⚠️文档未列定价(暂落0,需人工补录)。" + desc desc = "⚠️文档未列定价(暂落0,需人工补录)。" + desc
# 异步模型:后续步骤模板链(提交后顺序执行 query→[download])
query_ids = []
if sync_mode == "async":
steps = spec.get("async_steps") or []
if not any((s.get("purpose") or "") == "query" for s in steps):
desc = "⚠️异步模型但文档未提取到查询步骤(query_profile_ids 需人工补录)。" + desc
for s in steps:
qid = await _ensure_step_profile(sor, protocol, cap, vmid, s, spec)
if qid:
query_ids.append(qid)
query_ids_json = json.dumps(query_ids) if query_ids else ""
recs = await sor.sqlExe( recs = await sor.sqlExe(
"SELECT id FROM llm_model WHERE vendor_id=${v}$ AND vendor_model_id=${m}$", "SELECT id FROM llm_model WHERE vendor_id=${v}$ AND vendor_model_id=${m}$",
{"v": vendor_id, "m": vmid}) {"v": vendor_id, "m": vmid})
@ -395,10 +418,11 @@ async def _h_apply_llm_config(sor, params, ctx):
mid = getattr(recs[0], "id", "") mid = getattr(recs[0], "id", "")
await sor.sqlExe( await sor.sqlExe(
"UPDATE llm_model SET price_input=${pi}$, price_output=${po}$, " "UPDATE llm_model SET price_input=${pi}$, price_output=${po}$, "
"cost_input=${ci}$, cost_output=${co}$, description=${d}$, updated_at=NOW() " "cost_input=${ci}$, cost_output=${co}$, description=${d}$, "
"sync_mode=${sm}$, query_profile_ids=${q}$, updated_at=NOW() "
"WHERE id=${i}$", "WHERE id=${i}$",
{"pi": price_in, "po": price_out, "ci": cost_in, "co": cost_out, {"pi": price_in, "po": price_out, "ci": cost_in, "co": cost_out,
"d": desc, "i": mid}) "d": desc, "sm": sync_mode, "q": query_ids_json, "i": mid})
await sor.sqlExe("COMMIT", {}) await sor.sqlExe("COMMIT", {})
updated.append(vmid) updated.append(vmid)
continue continue
@ -406,7 +430,8 @@ async def _h_apply_llm_config(sor, params, ctx):
await sor.C("llm_model", { await sor.C("llm_model", {
"id": mid, "vendor_id": vendor_id, "account_id": "", "id": mid, "vendor_id": vendor_id, "account_id": "",
"name": vmid, "vendor_model_id": vmid, "capability": cap, "name": vmid, "vendor_model_id": vmid, "capability": cap,
"sync_mode": "sync", "profile_id": profile_ids.get(cap, ""), "sync_mode": sync_mode, "profile_id": profile_ids.get(cap, ""),
"query_profile_ids": query_ids_json,
"ppid": "", "price_input": price_in, "price_output": price_out, "ppid": "", "price_input": price_in, "price_output": price_out,
"cost_input": cost_in, "cost_output": cost_out, "cost_input": cost_in, "cost_output": cost_out,
"default_params": "{}", "status": "active", "default_params": "{}", "status": "active",
@ -424,18 +449,17 @@ async def _h_apply_llm_config(sor, params, ctx):
}, ensure_ascii=False) }, ensure_ascii=False)
def _default_templates(protocol: str, capability: str, spec: dict) -> tuple: def _default_templates(protocol: str, capability: str, spec: dict) -> dict:
"""生成适配模板默认值(headers/data/response)。 """生成适配模板默认值(path/headers 独立列 + data/response 模板)。
OpenAI 兼容协议用平台标准模板;其他协议按文档格式生成骨架供人工微调。 OpenAI 兼容协议用平台标准模板;其他协议按文档格式生成骨架供人工微调。
返回 {"path", "headers", "req", "resp"}。
""" """
chat_path = (spec.get("chat_path") or "/chat/completions").strip() chat_path = (spec.get("chat_path") or "/chat/completions").strip()
headers = {"Authorization": "Bearer {{api_key}}",
"Content-Type": "application/json"}
if protocol == "openai_compat": if protocol == "openai_compat":
req = json.dumps({ req = json.dumps({
"path": chat_path, "method": "POST",
"headers": {"Authorization": "Bearer {{api_key}}",
"Content-Type": "application/json"},
"params": {},
"data": {"model": "{{model}}", "messages": "{{messages}}", "data": {"model": "{{model}}", "messages": "{{messages}}",
"temperature": "{{temperature}}", "stream": False}, "temperature": "{{temperature}}", "stream": False},
}, ensure_ascii=False) }, ensure_ascii=False)
@ -444,13 +468,9 @@ def _default_templates(protocol: str, capability: str, spec: dict) -> tuple:
"usage": {"prompt_tokens": "usage.prompt_tokens", "usage": {"prompt_tokens": "usage.prompt_tokens",
"completion_tokens": "usage.completion_tokens"}, "completion_tokens": "usage.completion_tokens"},
}, ensure_ascii=False) }, ensure_ascii=False)
return req, resp return {"path": chat_path, "headers": headers, "req": req, "resp": resp}
# 非标准协议:骨架模板(字段路径来自文档提取),标注需人工核对 # 非标准协议:骨架模板(字段路径来自文档提取),标注需人工核对
req = json.dumps({ 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 []}, "data": {"__from_doc__": spec.get("request_fields") or []},
"__note__": "非 openai_compat 协议骨架模板,需按文档人工核对", "__note__": "非 openai_compat 协议骨架模板,需按文档人工核对",
}, ensure_ascii=False) }, ensure_ascii=False)
@ -458,7 +478,43 @@ def _default_templates(protocol: str, capability: str, spec: dict) -> tuple:
"content": spec.get("response_format", "") or "待按文档填写", "content": spec.get("response_format", "") or "待按文档填写",
"__note__": "需人工核对", "__note__": "需人工核对",
}, ensure_ascii=False) }, ensure_ascii=False)
return req, resp return {"path": chat_path, "headers": headers, "req": req, "resp": resp}
async def _ensure_step_profile(sor, protocol, cap, vmid, step, spec):
"""异步模型的后续步骤模板(query/download):按名幂等创建/复用 profile。
step: {"purpose": "query|download", "path", "method", "request_fields", "response_format"}
返回 profile id。
"""
purpose = (step.get("purpose") or "query").strip() or "query"
pname = "%s-%s" % (vmid, purpose)
recs = await sor.sqlExe(
"SELECT id FROM llm_api_profile WHERE name=${n}$", {"n": pname})
await sor.sqlExe("COMMIT", {})
if recs:
return getattr(recs[0], "id", "")
headers = {"Authorization": "Bearer {{api_key}}",
"Content-Type": "application/json"}
method = (step.get("method") or "POST").strip().upper() or "POST"
req = json.dumps({
"method": method,
"data": {"__from_doc__": step.get("request_fields") or [],
"__task_id__": "{{task_id}}"},
"__note__": "异步%s步骤骨架,需按文档人工核对" % purpose,
}, ensure_ascii=False)
resp = json.dumps({
"content": step.get("response_format", "") or "待按文档填写",
"__note__": "需人工核对",
}, ensure_ascii=False)
pid = getID()
await sor.C("llm_api_profile", {
"id": pid, "name": pname, "protocol": protocol, "capability": cap,
"path": (step.get("path") or "").strip(),
"headers": json.dumps(headers, ensure_ascii=False),
"request_template": req, "response_template": resp,
"param_schema": "", "status": "active"})
return pid
# ────────────────────── 工具 5:模块/应用信息查询 ────────────────────── # ────────────────────── 工具 5:模块/应用信息查询 ──────────────────────