yumoqing
b399663863
1. tool_sources.py(新): global/org/pipeline/role/project 五级工具作用域解析
- 策略表 pipeline_tool_policies allow/deny(配套 models+CRUD json)
- capability 两层语义: 映射手册(all/概念技能,含global层) + 声明注入
(仅org/pipeline/role/project/user层)——防 generic 会话经 global 概念
技能拿到~70个产线工具,击穿七层隔离(单测 29/29 含此回归)
- role 白名单是过滤器非添加器,且作用于 base+capability 全集
2. agent_config: GENERAL_TOOLS 加 patch_file(定点替换,唯一性校验);
write_file 描述引导改文件优先用 patch_file
3. upload_tools: is_image + build_image_parts(OpenAI多模态data URL,
8MB/张+4张/条上限,超限显式告知不静默丢)
4. agent_chat/agent_chat_generic: 图片分流不进文本上下文,
image_paths 传 gateway
5. load_path: 策略管理页注册
226 lines
12 KiB
Plaintext
226 lines
12 KiB
Plaintext
# agent_chat.dspy - 通用会话 Agent（pipeline-core，产线无关）
|
||
# 使用 AgentExecutor v2（gateway 统一入口），是 Web 端复刻 Hermes CLI 的通用交互能力
|
||
# 各产线/模块可在其上叠加专属能力，本 dspy 不依赖任何产线专属表
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import zipfile
|
||
from ahserver.filestorage import FileStorage
|
||
|
||
|
||
def _extract_text(path, name):
|
||
"""[已废弃] 保留签名兼容；真实逻辑在 pipeline_core.upload_tools.extract_text。"""
|
||
from pipeline_core.upload_tools import extract_text
|
||
preview, _total, _trunc = extract_text(path, name)
|
||
return preview
|
||
|
||
|
||
action = (params_kw or {}).get('action', 'send_message')
|
||
msg = ''
|
||
if action != 'send_message':
|
||
p = params_kw or {}
|
||
msg = (p.get('message_text') or '').strip()
|
||
if not msg:
|
||
inner = p.get('params', {})
|
||
if isinstance(inner, dict):
|
||
msg = (inner.get('prompt') or inner.get('message_text') or '').strip()
|
||
if msg:
|
||
action = 'send_message'
|
||
|
||
dbname = get_module_dbname('pipeline_core')
|
||
|
||
if action == 'send_message':
|
||
# 从 params 中提取 prompt
|
||
prompt = (params_kw or {}).get('prompt', '') or msg
|
||
prompt = prompt.strip()
|
||
|
||
# 处理用户上传的文件（multipart file 字段 → web_path）：
|
||
# 统一走 pipeline_core.upload_tools——截断显式告知、落盘位置=agent 可读位置。
|
||
from pipeline_core.upload_tools import extract_text, resolve_upload_dir, save_uploads, build_file_context, is_image
|
||
_items = [] # build_file_context 入参
|
||
_uploads = [] # [(src_abs, filename)]
|
||
_img_uploads = [] # 图片上传（原生视觉，2026-09-10）：不进文本上下文，走多模态注入
|
||
_fval = (params_kw or {}).get('file')
|
||
_fpaths = _fval if isinstance(_fval, list) else ([_fval] if _fval else [])
|
||
for _fp in _fpaths:
|
||
try:
|
||
_abs = FileStorage().realPath(_fp)
|
||
_name = os.path.basename(_abs)
|
||
_uploads.append((_abs, _name))
|
||
except Exception:
|
||
pass
|
||
|
||
# ── 意图识别：停止类指令 ──
|
||
# 注意（2026-08-31 修复）：会话 agent 新增了项目管理能力（终止/删除项目走
|
||
# pause_project/delete_project 工具，代码层校验人类 owner）。「终止项目」「停止项目」
|
||
# 含停止词但语义是项目管理指令，不能拦——只对纯停止指令（或不含项目词的）拦截。
|
||
stop_keywords = ['停止', '取消', '停下', '终止', '停', 'stop', 'cancel', 'abort']
|
||
_pl = prompt.strip().lower()
|
||
_is_project_mgmt = any(w in _pl for w in ('项目', '工程', 'project'))
|
||
if any(kw == _pl or (kw in _pl and not _is_project_mgmt) for kw in stop_keywords):
|
||
async def _stop_stream():
|
||
yield json.dumps({"content": "已停止当前任务。"}, ensure_ascii=False) + '\n'
|
||
yield json.dumps({"widgettype": "Text", "options": {"otext": "## 完毕 ##", "text": "## 完毕 ##", "i18n": True, "halign": "left", "css": "agent-done", "color": "#94a3b8", "marginBottom": "6px"}}, ensure_ascii=False) + '\n'
|
||
return await stream_response(request, _stop_stream, 'text/plain; charset=utf-8')
|
||
|
||
uid = await get_user()
|
||
if not uid:
|
||
uid = 'user-01' # 测试兼容
|
||
|
||
# ── 会话内唯一标识（web 多 tab 独立会话）：前端每个 tab 传独立 session_id ──
|
||
session_id = (params_kw or {}).get('session_id', '') or ''
|
||
# ── 产线默认能力（入口指定，如 bidding_general）：无当前项目时装载该产线能力 ──
|
||
pipeline_id = (params_kw or {}).get('pipeline_id', '') or ''
|
||
|
||
# ── 订阅门禁（2026-09-08 用户要求：未购买不能进入对话模式）──
|
||
# 产线对话须持有该产线有效（未到期）订阅；平台侧角色（owner./reseller.）
|
||
# 豁免；通用会话（空/_generic）豁免。桌面/手机所有入口在此统一拦截，
|
||
# 前端遮罩只做引导。DB 异常时 fail-open（可用性优先，门禁是商业限制非安全边界）。
|
||
if pipeline_id and pipeline_id != '_generic':
|
||
_gate_ok = True
|
||
try:
|
||
_roles = await get_user_roles(uid)
|
||
_is_ops = any(str(r).startswith('owner.') or str(r).startswith('reseller.')
|
||
for r in (_roles or []))
|
||
except Exception:
|
||
_is_ops = False
|
||
if not _is_ops:
|
||
_orgid = await get_userorgid()
|
||
try:
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
_sub = await sor.sqlExe(
|
||
"SELECT ps.id FROM product_subscription ps"
|
||
" JOIN product pr ON ps.product_id = pr.id"
|
||
" WHERE ps.user_org_id=${org}$ AND ps.status='1'"
|
||
" AND ps.end_date >= CURDATE()"
|
||
" AND pr.product_type='pipeline'"
|
||
" AND pr.resource_ref_id=${pid}$ LIMIT 1",
|
||
{"org": _orgid, "pid": pipeline_id})
|
||
await sor.sqlExe("COMMIT", {})
|
||
_gate_ok = bool(_sub)
|
||
except Exception:
|
||
_gate_ok = True
|
||
if not _gate_ok:
|
||
async def _gate_stream():
|
||
yield json.dumps({"error": "尚未购买该产线或订阅已到期，请到产品商城购买后再开始对话"},
|
||
ensure_ascii=False) + '\n'
|
||
return await stream_response(request, _gate_stream, 'text/plain; charset=utf-8')
|
||
|
||
# ── 上传文件落盘（有项目→项目根；无项目→workspace 会话目录）+ 生成显式截断上下文 ──
|
||
file_ctx = ''
|
||
if _uploads:
|
||
try:
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
_udir, _rel = await resolve_upload_dir(sor, uid, session_id)
|
||
_saved = save_uploads(_udir, _uploads)
|
||
_byname = {os.path.basename(src): n for (src, _n), (n, _p) in zip(_uploads, _saved)}
|
||
for _src, _name in _uploads:
|
||
if is_image(_name):
|
||
# 图片走原生视觉（gateway.build_image_parts），不进文本上下文——
|
||
# 避免注入「二进制无法读取」与「图片已注入」矛盾提示。
|
||
# 仍随 save_uploads 落盘（agent 可再用 invoke_model i2t 处理）
|
||
_img_uploads.append((_src, _name))
|
||
continue
|
||
_preview, _total, _trunc = extract_text(_src, _name)
|
||
_isbin = (not _preview and _total == 0)
|
||
_relp = (_rel + _byname.get(_name, _name)) if _name in _byname else ''
|
||
_items.append((_name, _relp, _preview, _total, _trunc, _isbin))
|
||
file_ctx = build_file_context(_items)
|
||
except Exception:
|
||
file_ctx = ''
|
||
if file_ctx:
|
||
prompt = file_ctx + "\n用户指令：" + prompt
|
||
|
||
# ── 走 gateway 统一入口（Web AgentIO 通道）──
|
||
# model_id：前端模型下拉选中的模型 → gateway 校验后持久化到项目（项目模型一经设置
|
||
# 即生效，直到用户再次选择），空 = 沿用项目已设模型。
|
||
model_id = (params_kw or {}).get('model_id', '') or ''
|
||
from pipeline_service.gateway import get_gateway
|
||
gateway = get_gateway()
|
||
|
||
async def agent_stream():
|
||
_base = entire_url("/")
|
||
async for chunk in gateway.run_message("web", uid, prompt, base_url=_base, session_id=session_id, pipeline_id=pipeline_id, model_id=model_id, image_paths=_img_uploads or None):
|
||
data = json.loads(chunk)
|
||
t = data.get('type', '')
|
||
|
||
if t == 'progress':
|
||
yield json.dumps({"reasoning_content": data.get('message', '') + "\n"}, ensure_ascii=False) + '\n'
|
||
elif t == 'auto_tool':
|
||
yield json.dumps({"reasoning_content": "🔄 " + data.get('message', '') + "\n"}, ensure_ascii=False) + '\n'
|
||
elif t == 'debug':
|
||
continue # skip debug in UI
|
||
elif t == 'tool_call':
|
||
tool = data.get('tool', '')
|
||
params = json.dumps(data.get('params', {}), ensure_ascii=False)
|
||
yield json.dumps({"content": "**🔧 调用: " + tool + "**\n```\n" + params + "\n```\n\n"}, ensure_ascii=False) + '\n'
|
||
elif t == 'tool_result':
|
||
result = data.get('result', '')
|
||
yield json.dumps({"content": result + "\n\n"}, ensure_ascii=False) + '\n'
|
||
elif t == 'reply':
|
||
msg = data.get('message', '')
|
||
if isinstance(msg, dict) and msg.get('widgettype'):
|
||
# widget JSON：顶层透传，前端 AgentOut 检测 widgettype 渲染（如 /task 的 PopupWindow）
|
||
yield json.dumps(msg, ensure_ascii=False) + '\n'
|
||
else:
|
||
yield json.dumps({"content": msg}, ensure_ascii=False) + '\n'
|
||
elif t == 'ask_user':
|
||
yield json.dumps({"content": "❓ " + data.get('message', '')}, ensure_ascii=False) + '\n'
|
||
elif t == 'confirm':
|
||
_tool = data.get('tool', '')
|
||
_params = data.get('params', {}) or {}
|
||
# 确认框摘要（2026-09-05 修复）：旧实现只取 command 字段（run_command 专用），
|
||
# 其他工具（apply_llm_config 等参数叫 spec）显示为空命令，用户被迫盲确认。
|
||
# 改为显示工具名 + 关键参数摘要（use_last_extract/overrides/url/model_name 等），
|
||
# 大字段（spec/doc_text）只显长度不显内容。
|
||
_brief = []
|
||
if isinstance(_params, dict):
|
||
for _k in ('command', 'url', 'model_name', 'use_last_extract', 'overrides', 'task_ref'):
|
||
if _k in _params and str(_params.get(_k) or '') != '':
|
||
_v = _params.get(_k)
|
||
_vs = _v if isinstance(_v, str) else json.dumps(_v, ensure_ascii=False)
|
||
_brief.append("%s=%s" % (_k, _vs[:80]))
|
||
for _k in ('spec', 'doc_text'):
|
||
if _params.get(_k):
|
||
_brief.append("%s=<%d字符>" % (_k, len(str(_params.get(_k)))))
|
||
_summary = '，'.join(_brief) if _brief else '(无参数)'
|
||
yield json.dumps({"content": "⚠️ 需要你确认执行工具：**" + str(_tool) + "**\n参数：" + _summary[:300] + "\n请回复「确认」执行、「全部确认」本会话不再询问，或「取消」放弃。"}, ensure_ascii=False) + '\n'
|
||
elif t == 'error':
|
||
yield json.dumps({"error": data.get('message', '')}, ensure_ascii=False) + '\n'
|
||
else:
|
||
yield json.dumps({"content": chunk}, ensure_ascii=False) + '\n'
|
||
# 应答输出结束后追加一行「## 完毕 ##」：Text 控件 otext+i18n，前端按当前语言翻译
|
||
# （词条见 pipeline-app/i18n/*/i18n.json；msg.txt 以 # 开头的键会被当注释，故只放 i18n.json）
|
||
yield json.dumps({"widgettype": "Text", "options": {"otext": "## 完毕 ##", "text": "## 完毕 ##", "i18n": True, "halign": "left", "css": "agent-done", "color": "#94a3b8", "marginBottom": "6px"}}, ensure_ascii=False) + '\n'
|
||
|
||
return await stream_response(request, agent_stream, 'text/plain; charset=utf-8')
|
||
|
||
elif action == 'list_messages':
|
||
uid = await get_user()
|
||
if not uid:
|
||
return json.dumps({"error": "请先登录"}, ensure_ascii=False)
|
||
# 产线隔离（2026-09-02）：必须按 user + session + pipeline 过滤。
|
||
# 旧实现无过滤返回全表——跨产线/跨用户会话历史泄漏。
|
||
_sid = (params_kw or {}).get('session_id', '') or ''
|
||
_pl = (params_kw or {}).get('pipeline_id', '') or ''
|
||
_sql = ("SELECT role, content, created_at FROM pipeline_conversations "
|
||
"WHERE created_by=${u}$")
|
||
_kw = {"u": uid, "lim": 100}
|
||
if _sid:
|
||
_sql += " AND session_id=${sid}$"
|
||
_kw["sid"] = _sid
|
||
if _pl:
|
||
_sql += " AND pipeline_id=${pl}$"
|
||
_kw["pl"] = _pl
|
||
# DESC+LIMIT 取最新再倒序（与 _load_history 隔离策略一致；ASC+LIMIT 会取最旧）
|
||
_sql += " ORDER BY created_at DESC LIMIT ${lim}$"
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
ms = await sor.sqlExe(_sql, _kw)
|
||
result = [{"role": getattr(m, 'role', ''), "content": getattr(m, 'content', '')}
|
||
for m in reversed(ms or [])]
|
||
return json.dumps({"success": True, "messages": result}, ensure_ascii=False, default=str)
|
||
|
||
else:
|
||
return json.dumps({"error": "Unknown: " + str(action)}, ensure_ascii=False)
|