From 21c8c16daa41cab0db475e7a8f89739cfac936c7 Mon Sep 17 00:00:00 2001 From: ymq Date: Fri, 4 Sep 2026 15:57:41 +0800 Subject: [PATCH] =?UTF-8?q?feat(rag-access):=20=E4=BA=A7=E7=BA=BF=E6=8E=A5?= =?UTF-8?q?=E5=85=A5=E7=9F=A5=E8=AF=86=E5=BA=93=E2=80=94=E2=80=94=E5=85=A8?= =?UTF-8?q?=E8=A7=92=E8=89=B2=E6=8C=89=E9=A1=B9=E7=9B=AEowner=E6=9D=83?= =?UTF-8?q?=E9=99=90=E6=A3=80=E7=B4=A2(rag=5Fsearch/rag=5Fkb=5Flist)+PM?= =?UTF-8?q?=E5=BB=BA=E8=AE=AE=E5=85=A5=E5=BA=93=E5=BE=85=E5=8A=9E=E9=97=AD?= =?UTF-8?q?=E7=8E=AF+=E7=BB=9F=E4=B8=80API=E6=A8=A1=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- models/pipeline_kb_suggestions.json | 30 +++ pipeline_service/agent_loop.py | 32 ++- pipeline_service/init.py | 5 + pipeline_service/kb_ingest_capability.py | 207 +++++++++++++++++ pipeline_service/rag_client.py | 276 +++++++++++++++++++++++ 5 files changed, 549 insertions(+), 1 deletion(-) create mode 100644 models/pipeline_kb_suggestions.json create mode 100644 pipeline_service/kb_ingest_capability.py create mode 100644 pipeline_service/rag_client.py diff --git a/models/pipeline_kb_suggestions.json b/models/pipeline_kb_suggestions.json new file mode 100644 index 0000000..285030d --- /dev/null +++ b/models/pipeline_kb_suggestions.json @@ -0,0 +1,30 @@ +{ + "summary": [ + { + "name": "pipeline_kb_suggestions", + "title": "PM 知识库入库建议", + "primary": ["id"], + "catelog": "entity" + } + ], + "fields": [ + {"name": "id", "title": "建议ID", "type": "str", "length": 32, "nullable": "no"}, + {"name": "project_id", "title": "项目ID", "type": "str", "length": 32, "nullable": "no"}, + {"name": "kb_id", "title": "目标知识库ID", "type": "str", "length": 32, "nullable": "no"}, + {"name": "file_path", "title": "文件绝对路径", "type": "str", "length": 500, "nullable": "no"}, + {"name": "file_name", "title": "文件名", "type": "str", "length": 255, "nullable": "no"}, + {"name": "reason", "title": "入库理由", "type": "text"}, + {"name": "status", "title": "状态", "type": "str", "length": 16, "nullable": "no", "default": "pending"}, + {"name": "suggested_by", "title": "建议人", "type": "str", "length": 64, "nullable": "no", "default": ""}, + {"name": "task_id", "title": "关联任务ID", "type": "str", "length": 32, "nullable": "no", "default": ""}, + {"name": "result_comment", "title": "处理结果备注", "type": "str", "length": 1000, "nullable": "no", "default": ""}, + {"name": "decided_by", "title": "决策人", "type": "str", "length": 32, "nullable": "no", "default": ""}, + {"name": "decided_at", "title": "决策时间", "type": "timestamp", "nullable": "yes"}, + {"name": "created_at", "title": "创建时间", "type": "timestamp", "nullable": "no"}, + {"name": "updated_at", "title": "更新时间", "type": "timestamp", "nullable": "no"} + ], + "indexes": [ + {"name": "idx_kbs_proj_status", "idxtype": "index", "idxfields": ["project_id", "status"]}, + {"name": "idx_kbs_kb", "idxtype": "index", "idxfields": ["kb_id"]} + ] +} diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index 28ebe26..5e4c91d 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -544,6 +544,8 @@ async def _git_clone(repo_url, target_dir, branch='main'): AGENT_TOOLS = [ {"name":"read_file","description":"读取工作空间中的文件","params":{"path":"相对路径"}}, + {"name":"rag_search","description":"检索知识库（按项目owner权限，自动限定可检范围）。查资料/找依据/了解背景时使用","params":{"query":"检索内容","kb_id":"知识库ID(可选,缺省检索全部可见知识库)","top_k":"返回条数(可选,默认10)"}}, + {"name":"rag_kb_list","description":"列出项目可见的知识库（名称+ID），不确定检索哪个库时先调这个","params":{}}, {"name":"load_skill","description":"按需加载技能全文或子文件——需要具体规范/目录结构/路径/格式/流程时先加载对应技能（如 project-directory-spec 项目目录规范），不要凭记忆瞎写。只给 name 加载 SKILL.md 全文，给 file_path 加载 references/scripts/templates 下的子文件","params":{"name":"技能名","file_path":"子文件相对路径(可选,如 references/api.md)"}}, {"name":"write_file","description":"写入文件（自动创建父目录）","params":{"path":"相对路径","content":"文件内容"}}, {"name":"list_files","description":"列出目录内容","params":{"path":"相对路径(可选,默认工作空间根)"}}, @@ -651,6 +653,9 @@ __ROLE_SKILLS__ - list_tasks(role, state) — 列出项目现有任务（派发前先查，避免重复） - cancel_task(task_id) — 取消任务（重做/作废前必须先取消旧任务，避免两个相同任务并存） - update_task_deps(task_id, add_depends_on, dep_policy?) — 给已存在任务补前置依赖（系统发现编排缺口时通知你，核实后用此工具修正；add_depends_on 是要追加的任务ID数组，dep_policy 可选、同上） +- rag_kb_list() — 列出项目可见的知识库（按项目owner权限） +- rag_search(query, kb_id?, top_k?) — 检索知识库（按项目owner权限，自动限定可检范围） +- rag_suggest_ingest(kb_id, file_path, reason) — 建议将项目产出文件加入知识库：必须给出理由；提交后生成 owner 待办，owner 批准才真正入库。用于沉淀有复用价值的项目产出（设计文档/规范/案例等）。同一文件不要重复建议 ## 审核流程 1. 先用工具检查代码/交付件是否实际产出（git_status / list_files / read_file） @@ -2011,6 +2016,29 @@ async def _exec_agent_tool(tool, params, workspace_dir, ctx=None): target = _resolve_repo_target(workspace_dir, p.get('repo_dir', '')) r = await _git_commit_push(target, msg) return f"rc={r['rc']} {r['message']}" + # 知识库工具（rag 对外 API 模式；检索权限=项目 owner；建议入库仅 PM） + if tool in ('rag_search', 'rag_kb_list', 'rag_suggest_ingest'): + from . import rag_client as _rc + _pid = str((ctx or {}).get('project_id', '') or '') + if tool == 'rag_search': + return await _rc.tool_rag_search( + _pid, p.get('query', ''), kb_id=p.get('kb_id', ''), + top_k=p.get('top_k', 10)) + if tool == 'rag_kb_list': + return await _rc.tool_rag_kb_list(_pid) + # rag_suggest_ingest：仅 PM 可用（防角色 agent 越权提建议） + if str((ctx or {}).get('who', '') or '') != 'agent.pm': + return 'FAIL: 仅 PM 可提交知识库入库建议' + from .kb_ingest_capability import suggest_kb_ingest + ok, msg = await suggest_kb_ingest( + _pid, p.get('kb_id', ''), p.get('file_path', ''), + p.get('reason', ''), who=str((ctx or {}).get('who', '') or ''), + agent_id=str((ctx or {}).get('agent_id', '') or ''), + task_id=str((ctx or {}).get('task_id', '') or '')) + if ok: + return ("OK: 入库建议已提交（编号 " + msg + + "），已生成项目 owner 待办，等待批准后自动入库") + return 'FAIL: ' + msg # 能力工具（propose_feature/create_case/report_bug 等，按角色 capability 注入） from .capability_tools import exec_capability_tool, TOOL_SCHEMAS if tool in TOOL_SCHEMAS: @@ -2912,7 +2940,9 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): result = (f"已到最后收尾阶段（第 {turn + 1}/5 轮），拒绝执行探索类工具 {tool}。" f"请立即输出 review_approve / review_reject / review_complete / review_rollback 之一，不要再调用工具。") else: - result = await _exec_agent_tool(tool, params, space_dir) + result = await _exec_agent_tool(tool, params, space_dir, { + "project_id": project_id, "who": "agent.pm", + "agent_id": agent_id or "", "task_id": task_id}) msgs.append({"role": "assistant", "content": raw}) msgs.append({"role": "user", "content": f"工具 {tool} 结果：\n{result}"}) else: diff --git a/pipeline_service/init.py b/pipeline_service/init.py index a9a0409..031a70d 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -526,6 +526,11 @@ def load_pipeline_service(): env.check_task_owner = check_task_owner env.check_tenant_owner = check_tenant_owner + # 知识库入库建议（PM 建议 → owner 待办批准 → rag API 上传） + from .kb_ingest_capability import suggest_kb_ingest, decide_kb_ingest + env.suggest_kb_ingest = suggest_kb_ingest + env.decide_kb_ingest = decide_kb_ingest + # Register default handler register_default_handler() diff --git a/pipeline_service/kb_ingest_capability.py b/pipeline_service/kb_ingest_capability.py new file mode 100644 index 0000000..15d4cd2 --- /dev/null +++ b/pipeline_service/kb_ingest_capability.py @@ -0,0 +1,207 @@ +# -*- coding: utf-8 -*- +"""pipeline_service/kb_ingest_capability.py — PM 建议文件入知识库能力。 + +用户需求（2026-09-04）：PM agent 有权建议「将什么文件加入什么知识库」并说明理由。 +设计（遵守平台铁律：建议不直接执行，人工确认后才动数据；待办内完成不跳转；只发一次）： + +1. PM 调 rag_suggest_ingest(kb_id, file_path, reason)： + - 校验项目工作空间内文件真实存在（不允许编造文件）； + - 落 pipeline_kb_suggestions（status=pending）+ 创建 owner 人工待办（一次性）； +2. owner 在待办详情里看到文件（可预览/下载）+ 理由，批准/驳回： + - 批准 → 后端用「项目 owner 身份的 rag API key」调 /rag/api/doc_upload.dspy 真实入库 + （与检索同一权限模型），建议置 approved； + - 驳回 → 建议置 rejected，记录意见。 + +状态机：pending → approved/rejected（终态，幂等防重复上传）。 +""" + +import json +import logging +import os + +from appPublic.uniqueID import getID +from sqlor.dbpools import DBPools + +logger = logging.getLogger("pipeline.kb_ingest") + +DBNAME = "pipeline" + +S_PENDING = "pending" +S_APPROVED = "approved" +S_REJECTED = "rejected" + +T_KB_INGEST = "kb_ingest_confirm" + + +def _get_db(): + db = DBPools() + if not db.databases: + from appPublic.jsonConfig import getConfig + config = getConfig() + if config.databases: + db.databases = config.databases + return db + + +def _rec_to_dict(rec): + if rec is None: + return {} + try: + return dict(rec) + except (TypeError, ValueError): + return {} + + +async def suggest_kb_ingest(project_id, kb_id, file_path, reason, + who="", agent_id="", task_id=""): + """PM 建议将文件加入知识库。返回 (True, suggestion_id) 或 (False, 错误)。""" + if not project_id or not kb_id or not file_path: + return False, "project_id/kb_id/file_path 均为必填" + if not (reason or '').strip(): + return False, "必须说明入库理由" + db = _get_db() + async with db.sqlorContext(DBNAME) as sor: + proj = await sor.sqlExe( + "SELECT id, workspace_dir, org_id FROM sd_projects WHERE id=${pid}$", + {"pid": project_id}) + await sor.sqlExe("COMMIT", {}) + if not proj: + return False, "项目不存在" + workspace_dir = str(getattr(proj[0], "workspace_dir", "") or "") + # 相对路径解析：兼容「项目目录相对路径」「机构工作空间相对路径 + # （projects/{项目}/…）」「项目内子目录相对路径」三种写法 + path = file_path.strip() + if not os.path.isabs(path): + candidates = [] + if workspace_dir: + candidates.append(os.path.join(workspace_dir, path)) + projects_dir = os.path.dirname(workspace_dir) # projects/ 层 + candidates.append(os.path.join(projects_dir, path)) + space_dir = os.path.dirname(projects_dir) # 机构工作空间层 + candidates.append(os.path.join(space_dir, path)) + full = "" + for c in candidates: + if os.path.isfile(c): + full = c + break + if not full: + return False, ("文件不存在（已查项目工作空间）: " + path + + "。请先确认文件已产出再建议入库") + else: + full = path + if not os.path.isfile(full): + return False, "文件不存在: " + full + # 幂等：同项目同文件同库未决建议不重复发待办（铁律：同事待办只发一次） + dup = await sor.sqlExe( + "SELECT id FROM pipeline_kb_suggestions WHERE project_id=${p}$ AND kb_id=${k}$ " + "AND file_path=${f}$ AND status='pending' LIMIT 1", + {"p": project_id, "k": kb_id, "f": full}) + await sor.sqlExe("COMMIT", {}) + if dup: + return False, "该文件已有待确认的入库建议（" + str(getattr(dup[0], 'id', ''))[:8] + "…），勿重复提交" + # 知识库可见性校验（用 owner 权限查，防止建议一个 owner 看不到的库） + from .rag_client import rag_kb_list + kbdata, kberr = await rag_kb_list(project_id) + if kberr: + return False, "校验知识库失败: " + kberr + kbs = (kbdata or {}).get("kbs") if isinstance(kbdata, dict) else [] + known = {str(k.get("id") or "") for k in kbs if isinstance(k, dict)} + if known and str(kb_id) not in known: + return False, "知识库不存在或项目 owner 无权检索该库: " + str(kb_id) + + sid = getID() + await sor.C("pipeline_kb_suggestions", { + "id": sid, "project_id": project_id, "kb_id": str(kb_id), + "file_path": full, "file_name": os.path.basename(full), + "reason": reason.strip(), "status": S_PENDING, + "suggested_by": who or agent_id or "agent.pm", + "task_id": task_id or "", "result_comment": "", + }) + await sor.sqlExe("COMMIT", {}) + + # 解析真人 owner 作为待办处理人 + from .rag_client import resolve_project_owner + owner, oerr = await resolve_project_owner(project_id) + if oerr or not owner: + owner = "" + # 待办（一次一条；suggestion_id 塞进 form_schema 供待办详情读取， + # fields 为空=无表单，只有批准/驳回按钮） + from .human_task_capability import create_human_task + desc = ("PM 建议将文件加入知识库，请审阅理由后决定。\n" + "文件：" + os.path.basename(full) + "\n" + "理由：" + reason.strip()) + ok, ht = await create_human_task( + project_id, + "知识库入库建议：" + os.path.basename(full), + description=desc, task_type=T_KB_INGEST, + assignee_id=owner or None, + assignee_role=None if owner else "owner", + created_by="agent.pm", task_id=task_id or "", + form_schema={"fields": [], "suggestion_id": sid}) + if not ok: + logger.warning("kb_ingest todo create failed: %s", ht) + return True, sid + + +async def decide_kb_ingest(suggestion_id, approve, operator_id, comment=""): + """owner 决策入库建议。批准 → 真实上传（rag API，owner 身份）。 + + 返回 (True, 消息) 或 (False, 错误)。 + """ + if not suggestion_id: + return False, "缺少建议 id" + if not operator_id: + return False, "未登录" + db = _get_db() + async with db.sqlorContext(DBNAME) as sor: + recs = await sor.sqlExe( + "SELECT * FROM pipeline_kb_suggestions WHERE id=${i}$", {"i": suggestion_id}) + await sor.sqlExe("COMMIT", {}) + if not recs: + return False, "建议不存在" + sg = _rec_to_dict(recs[0]) + if sg.get("status") != S_PENDING: + return False, "建议已处理（当前状态: " + str(sg.get("status")) + "）" + project_id = sg.get("project_id", "") + # 操作者须为项目真人 owner（与待办处理人一致） + from .rag_client import resolve_project_owner + owner, oerr = await resolve_project_owner(project_id) + if owner and str(operator_id) != str(owner): + return False, "仅项目 owner 可批准/驳回入库建议" + + # 关联人工待办置为已处理（避免待办重复出现） + await sor.sqlExe( + "UPDATE pipeline_human_tasks SET status='done', submitted_by=${o}$, submitted_at=NOW(), " + "result_data=${rd}$ WHERE task_type='kb_ingest_confirm' AND status='pending' " + "AND description LIKE ${pat}$", + {"o": operator_id, "rd": json.dumps({"suggestion_id": suggestion_id, + "approve": bool(approve)}, ensure_ascii=False), + "pat": "%" + str(sg.get("file_name", "")) + "%"}) + await sor.sqlExe("COMMIT", {}) + + if not approve: + await sor.sqlExe( + "UPDATE pipeline_kb_suggestions SET status='rejected', result_comment=${c}$, " + "decided_by=${o}$, decided_at=NOW() WHERE id=${i}$", + {"c": comment or "owner 驳回", "o": operator_id, "i": suggestion_id}) + await sor.sqlExe("COMMIT", {}) + return True, "已驳回，不会入库" + + # 批准 → 真实上传（API 模式，owner 身份） + from .rag_client import rag_doc_upload + data, err = await rag_doc_upload(project_id, sg.get("kb_id", ""), + sg.get("file_path", ""), + sg.get("file_name", "")) + if err: + await sor.sqlExe( + "UPDATE pipeline_kb_suggestions SET result_comment=${c}$ WHERE id=${i}$", + {"c": "上传失败: " + err, "i": suggestion_id}) + await sor.sqlExe("COMMIT", {}) + return False, "上传失败: " + err + doc_id = str((data or {}).get("doc_id", "") or "") + await sor.sqlExe( + "UPDATE pipeline_kb_suggestions SET status='approved', result_comment=${c}$, " + "decided_by=${o}$, decided_at=NOW() WHERE id=${i}$", + {"c": "已入库，doc_id=" + doc_id, "o": operator_id, "i": suggestion_id}) + await sor.sqlExe("COMMIT", {}) + return True, "已批准并入库（文档号 " + (doc_id or "-") + "，后台解析中）" diff --git a/pipeline_service/rag_client.py b/pipeline_service/rag_client.py new file mode 100644 index 0000000..b50262c --- /dev/null +++ b/pipeline_service/rag_client.py @@ -0,0 +1,276 @@ +# -*- coding: utf-8 -*- +"""pipeline_service/rag_client.py — 产线引擎访问 rag 知识库的唯一通道（API 模式）。 + +架构（2026-09-04 用户需求：所有产线按权限检索知识库 + 接口统一 API 模式）： +- 检索/上传一律走 rag 对外 HTTP API（/rag/api/*.dspy），不做同进程直调； +- 认证用 dapi 平台 Bearer key：按「项目 owner」身份发放并缓存 + （create_user_apikey(dappid='pipeline-rag', user_id=owner)，幂等）； +- 检索范围 = owner 的机构隔离 + 知识库 search_roles（rag 侧 _resolve_search_kbs 实现， + 本模块不重复造轮子）——即「能检索哪个知识库，用项目 owner 的权限」； +- key 明文只存 params 表（RC4 加密），进程内缓存；不写日志。 + +owner 解析链（与 check_project_owner 的降级语义一致）： +1. sd_projects.created_by 是真人（不以 'agent.' 开头）→ 直接用； +2. agent 创建的项目 → 反查 pipeline_conversations（iteration_id=项目）的真人用户 + （会话 agent 的 created_by 是发起会话的人类用户）； +3. 仍无 → 返回错误，调用方冒泡待办问用户（不硬编、不假成功）。 +""" + +import json +import logging + +import aiohttp + +from appPublic.rc4 import password as _rc4_encode +from appPublic.rc4 import unpassword as _rc4_decode +from appPublic.uniqueID import getID +from sqlor.dbpools import DBPools + +logger = logging.getLogger("pipeline.rag_client") + +DBNAME = "pipeline" +RAG_DAPP_ID = "pipeline-rag" # dapi downapp 记录（m0009 迁移种子） +_KEY_PARAM_PREFIX = "pipeline_rag_key_" # params 表按用户缓存 key 的前缀 +_RAG_TIMEOUT = aiohttp.ClientTimeout(total=90, connect=10) + +_key_cache = {} # user_id -> apikey（进程内，重启后从 params 重建） + + +def _get_db(): + db = DBPools() + if not db.databases: + from appPublic.jsonConfig import getConfig + config = getConfig() + if config.databases: + db.databases = config.databases + return db + + +async def resolve_project_owner(project_id): + """解析项目的真人 owner。返回 (user_id, '') 或 ('', 错误原因)。""" + if not project_id: + return "", "缺少 project_id" + db = _get_db() + async with db.sqlorContext(DBNAME) as sor: + recs = await sor.sqlExe( + "SELECT created_by FROM sd_projects WHERE id=${pid}$", {"pid": project_id}) + await sor.sqlExe("COMMIT", {}) + if not recs: + return "", "项目不存在" + owner = str(getattr(recs[0], "created_by", "") or "") + if owner and not owner.startswith("agent."): + return owner, "" + # agent 创建的项目：反查创建它的会话的真人用户 + crecs = await sor.sqlExe( + "SELECT created_by FROM pipeline_conversations " + "WHERE iteration_id=${pid}$ AND created_by != '' " + "AND created_by NOT LIKE 'agent.%' ORDER BY created_at ASC LIMIT 1", + {"pid": project_id}) + await sor.sqlExe("COMMIT", {}) + if crecs: + return str(getattr(crecs[0], "created_by", "")), "" + return "", ("项目由 agent 创建且反查不到会话用户，" + "无法确定知识库权限身份（请在项目里发起一次会话后重试）") + + +async def _rag_base(): + """rag API 基址：本机应用端口自调用（与 llm_bridge 内部代理同范式，不出内网）。""" + from appPublic.jsonConfig import getConfig + config = getConfig() + port = 9090 + try: + port = int(getattr(config.website, "port", 9090) or 9090) + except (TypeError, ValueError): + port = 9090 + return "http://127.0.0.1:%d/rag/api" % port + + +async def _ensure_downapp(sor): + """确保 dapi 的 downapp 记录存在（迁移种子兜底：个别环境漏跑迁移时自愈）。""" + recs = await sor.sqlExe("SELECT id FROM downapp WHERE id=${d}$", {"d": RAG_DAPP_ID}) + await sor.sqlExe("COMMIT", {}) + if not recs: + await sor.C("downapp", { + "id": RAG_DAPP_ID, "name": "pipeline-rag", + "description": "产线平台访问 rag 知识库（检索/入库）", + "secretkey": "", "allowedips": "", "orgid": "0", + }) + await sor.sqlExe("COMMIT", {}) + + +async def _load_cached_key(sor, user_id): + pname = _KEY_PARAM_PREFIX + str(user_id) + recs = await sor.sqlExe( + "SELECT params_value FROM params WHERE id=${n}$", {"n": pname}) + await sor.sqlExe("COMMIT", {}) + if not recs: + return "" + enc = str(getattr(recs[0], "params_value", "") or "") + if not enc: + return "" + try: + return _rc4_decode(enc) or "" + except Exception as e: + logger.warning("rag key decode failed for %s: %s", user_id, e) + return "" + + +async def get_owner_apikey(owner_id): + """获取 owner 的 rag Bearer key（缓存→params→dapi 发放）。返回 (key, '') 或 ('', err)。""" + if not owner_id: + return "", "缺少 owner" + key = _key_cache.get(owner_id, "") + if key: + return key, "" + db = _get_db() + async with db.sqlorContext(DBNAME) as sor: + key = await _load_cached_key(sor, owner_id) + if key: + _key_cache[owner_id] = key + return key, "" + await _ensure_downapp(sor) + try: + from dapi.dapi import create_user_apikey + res = await create_user_apikey(sor, RAG_DAPP_ID, owner_id, "") + except Exception as e: + return "", "dapi 发 key 失败: " + str(e)[:200] + await sor.sqlExe("COMMIT", {}) + if not res or res.get("status") != "success" or not res.get("apikey"): + return "", "dapi 发 key 失败: " + str(res or {}).get("message", "未知错误") + key = res["apikey"] + # RC4 加密存 params（与平台密码字段约定一致） + pname = _KEY_PARAM_PREFIX + str(owner_id) + enc = _rc4_encode(key) + await sor.sqlExe( + "INSERT INTO params (id, params_name, params_value) VALUES (${i}$, ${n}$, ${v}$) " + "ON DUPLICATE KEY UPDATE params_value=${v}$", + {"i": getID(), "n": pname, "v": enc}) + await sor.sqlExe("COMMIT", {}) + _key_cache[owner_id] = key + return key, "" + + +async def _rag_call(project_id, endpoint, payload=None, raw_body=None, query=None): + """以项目 owner 身份调 rag API。返回 (dict|bytes, '') 或 (None, 错误)。""" + owner, err = await resolve_project_owner(project_id) + if err: + return None, err + key, err = await get_owner_apikey(owner) + if err: + return None, err + base = await _rag_base() + url = base + "/" + endpoint + if query: + from urllib.parse import urlencode + url += "?" + urlencode(query) + headers = {"Authorization": "***" + key} + try: + async with aiohttp.ClientSession(timeout=_RAG_TIMEOUT) as session: + if raw_body is not None: + headers["Content-Type"] = "application/octet-stream" + async with session.post(url, headers=headers, data=raw_body) as resp: + body = await resp.read() + else: + headers["Content-Type"] = "application/json" + async with session.post(url, headers=headers, + json=payload or {}) as resp: + body = await resp.read() + except Exception as e: + return None, "rag API 调用失败: " + str(e)[:200] + try: + data = json.loads(body.decode("utf-8")) + except Exception: + return None, "rag API 返回非 JSON: " + body[:200].decode("utf-8", "replace") + # 统一返回 {"status":"ok","data":...}；少数 UI 兼容端点（kb_options）返回裸数组 + if isinstance(data, list): + return data, "" + if data.get("status") != "ok": + return None, str(data.get("message") or data.get("error") or "未知错误") + return data.get("data"), "" + + +async def rag_search(project_id, query, kb_id="", top_k=10): + """按项目 owner 权限检索知识库。返回 (results_data, '') 或 (None, err)。""" + payload = {"query": query, "top_k": top_k} + if kb_id: + payload["kb_id"] = kb_id + return await _rag_call(project_id, "search.dspy", payload=payload) + + +async def rag_kb_list(project_id): + """列出项目 owner 可见的知识库（rag 对外 API：/rag/api/kb_list.dspy）。""" + return await _rag_call(project_id, "kb_list.dspy", payload={}) + + +async def rag_doc_upload(project_id, kb_id, file_path, file_name=""): + """以 owner 身份上传文件入库（API 模式：原始字节 + query 参数）。""" + import os + if not os.path.isfile(file_path): + return None, "文件不存在: " + file_path + with open(file_path, "rb") as f: + raw = f.read() + if not raw: + return None, "文件为空: " + file_path + name = file_name or os.path.basename(file_path) + return await _rag_call(project_id, "doc_upload.dspy", raw_body=raw, + query={"kb_id": kb_id, "file_name": name}) + + +# ────────────────────────── agent 工具包装（返回 LLM 友好文本） ────────────────────────── + +def _fmt_search_data(data): + if not isinstance(data, dict): + return "(无检索结果)" + results = data.get("results") or [] + if not results: + msg = data.get("message") or "" + return ("没有检索到内容。" + (msg or "提示：可先用 rag_kb_list 查看可见知识库，" + "确认知识库已有文档入库（status=done）。")) + lines = [] + for i, h in enumerate(results[:20], 1): + if not isinstance(h, dict): + continue + text = str(h.get("text") or h.get("content") or "").replace("\n", " ") + score = h.get("score", "") + try: + score = "%.1f%%" % (float(score) * 100) + except (TypeError, ValueError): + score = str(score) + doc = h.get("document") if isinstance(h.get("document"), dict) else {} + fname = doc.get("file_name") or h.get("file_name") or "-" + kb = h.get("kb_id") or doc.get("kb_id") or "-" + lines.append("%d. [%s | %s | 库:%s] %s" % (i, score, fname, kb, text[:400])) + head = ("检索到 %s 条（共搜 %s 个知识库，召回 %s）：" + % (data.get("total", len(results)), data.get("kbs_searched", "-"), + data.get("recall", "-"))) + return head + "\n" + "\n".join(lines) + + +async def tool_rag_search(project_id, query, kb_id="", top_k=10): + """角色 agent 检索工具：按项目 owner 权限检索知识库。""" + if not (query or "").strip(): + return "FAIL: 需要检索内容 query" + try: + top_k = int(top_k or 10) + except (TypeError, ValueError): + top_k = 10 + data, err = await rag_search(project_id, query.strip(), kb_id=kb_id or "", top_k=top_k) + if err: + return "FAIL: " + err + return _fmt_search_data(data) + + +async def tool_rag_kb_list(project_id): + """角色 agent 工具：列出项目 owner 可见的知识库。""" + data, err = await rag_kb_list(project_id) + if err: + return "FAIL: " + err + kbs = (data or {}).get("kbs") if isinstance(data, dict) else [] + if not kbs: + return "当前项目 owner 没有可见的知识库。" + lines = [] + for k in kbs: + if isinstance(k, dict): + lines.append("- %s（ID: %s，文档数: %s）" % ( + k.get("name") or "-", k.get("id") or "-", k.get("doc_count", 0))) + return "可用知识库：\n" + "\n".join(lines)