feat(rag-access): 产线接入知识库——全角色按项目owner权限检索(rag_search/rag_kb_list)+PM建议入库待办闭环+统一API模式

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"]}
+    ]
+}

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:

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()
 

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 "-") + "，后台解析中）"

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)