Implement open knowledge access scope

ai_knowledge_base_v2/apps/backend/app/services/feishu_service.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import time
 from typing import TYPE_CHECKING, Any
 
 import httpx
@@ -11,49 +12,277 @@ from app.services.knowledge_service import KnowledgeScope
 if TYPE_CHECKING:
     from app.services.rag_service import RetrievedChunk
 
+# 飞书 API 基地址
+FEISHU_OPEN_API = "https://open.feishu.cn"
+
+# tenant_access_token 缓存
+_token_cache: dict[str, Any] = {"token": "", "expires_at": 0.0}
+
+
+def _get_tenant_token() -> str:
+    """获取飞书 tenant_access_token，自动缓存"""
+    settings = get_settings()
+    now = time.time()
+
+    if _token_cache["token"] and now < _token_cache["expires_at"]:
+        return _token_cache["token"]
+
+    url = f"{FEISHU_OPEN_API}/open-apis/auth/v3/tenant_access_token/internal"
+    payload = {
+        "app_id": settings.feishu_app_id,
+        "app_secret": settings.feishu_app_secret,
+    }
+
+    try:
+        resp = httpx.post(url, json=payload, timeout=settings.feishu_timeout_seconds)
+        resp.raise_for_status()
+        data = resp.json()
+
+        if data.get("code") != 0:
+            raise ExternalServiceError(
+                f"获取飞书Token失败：{data.get('msg', '未知错误')}",
+                provider="feishu",
+            )
+
+        token = data["tenant_access_token"]
+        expire = data.get("expire", 7200)
+
+        _token_cache["token"] = token
+        _token_cache["expires_at"] = now + expire - 300  # 提前5分钟过期
+
+        return token
+    except httpx.HTTPError as exc:
+        raise ExternalServiceError(f"获取飞书Token网络错误：{exc}", provider="feishu") from exc
+
+
+def _feishu_get(path: str, params: dict | None = None) -> dict:
+    """封装飞书 GET 请求"""
+    token = _get_tenant_token()
+    url = f"{FEISHU_OPEN_API}{path}"
+    headers = {"Authorization": f"Bearer {token}"}
+    settings = get_settings()
+
+    resp = httpx.get(url, headers=headers, params=params, timeout=settings.feishu_timeout_seconds)
+    resp.raise_for_status()
+    data = resp.json()
+
+    if data.get("code") != 0:
+        raise ExternalServiceError(
+            f"飞书API调用失败 [{path}]：{data.get('msg', '未知错误')}",
+            provider="feishu",
+        )
+    return data
+
+
+def _feishu_post(path: str, payload: dict | None = None) -> dict:
+    """封装飞书 POST 请求"""
+    token = _get_tenant_token()
+    url = f"{FEISHU_OPEN_API}{path}"
+    headers = {"Authorization": f"Bearer {token}"}
+    settings = get_settings()
+
+    resp = httpx.post(url, headers=headers, json=payload, timeout=settings.feishu_timeout_seconds)
+    resp.raise_for_status()
+    data = resp.json()
+
+    if data.get("code") != 0:
+        raise ExternalServiceError(
+            f"飞书API调用失败 [{path}]：{data.get('msg', '未知错误')}",
+            provider="feishu",
+        )
+    return data
+
+
+def _extract_text(elements: list[dict]) -> str:
+    """从 elements 列表中提取纯文本"""
+    texts = []
+    for elem in elements:
+        text_run = elem.get("text_run", {})
+        content = text_run.get("content", "")
+        if content:
+            texts.append(content)
+    return "".join(texts)
+
+
+def _parse_docx_content(blocks: list[dict]) -> str:
+    """将飞书文档 block 列表解析为纯文本"""
+    texts = []
+    for block in blocks:
+        block_type = block.get("block_type", 0)
+
+        # Page block (block_type=1): 标题可能在 page.elements
+        if block_type == 1:
+            elements = block.get("page", {}).get("elements", [])
+            content = _extract_text(elements)
+            if content:
+                texts.append(content)
+                texts.append("\n")
+            # 子内容在 children 里，但子 block 也会遍历到
+
+        # 文本块 (block_type=2 is text paragraph in new API)
+        elif block_type == 2:
+            elements = block.get("text", {}).get("elements", [])
+            content = _extract_text(elements)
+            if content:
+                texts.append(content)
+                texts.append("\n")
+
+        # 标题块 (3=heading1, 4=heading2, 5=heading3, 6=heading4, 7=heading5, 8=heading6)
+        elif 3 <= block_type <= 8:
+            heading_level = block_type - 2  # 3->h1, 4->h2, etc.
+            key = f"heading{heading_level}"
+            elements = block.get(key, {}).get("elements", [])
+            content = _extract_text(elements)
+            if content:
+                texts.append(f"{'#' * heading_level} {content}")
+                texts.append("\n")
+
+        # 引用块 (15=quote)
+        elif block_type == 15:
+            elements = block.get("quote", {}).get("elements", [])
+            content = _extract_text(elements)
+            if content:
+                texts.append(f"  > {content}")
+                texts.append("\n")
+
+        # 列表块 - bullet (9), ordered (10)
+        elif block_type in (9, 10):
+            key = "bullet" if block_type == 9 else "ordered"
+            elements = block.get(key, {}).get("elements", [])
+            content = _extract_text(elements)
+            if content:
+                prefix = "-" if block_type == 9 else "1."
+                texts.append(f"  {prefix} {content}")
+                texts.append("\n")
+
+        # 其他 block 类型也尝试提取文本
+        else:
+            for sub_key in ("text", "quote", "bullet", "ordered"):
+                if sub_key in block:
+                    elements = block.get(sub_key, {}).get("elements", [])
+                    content = _extract_text(elements)
+                    if content:
+                        texts.append(content)
+                        texts.append("\n")
+                        break
+
+    return "".join(texts).strip()
+
+
+def _get_docx_content(doc_token: str) -> str:
+    """读取单个飞书云文档内容"""
+    data = _feishu_get(f"/open-apis/docx/v1/documents/{doc_token}/blocks")
+    items = data.get("data", {}).get("items", [])
+    return _parse_docx_content(items)
+
+
+def _get_wiki_node_info(node_token: str) -> dict:
+    """获取 Wiki 节点信息，返回 space_id 和 obj_token"""
+    data = _feishu_get("/open-apis/wiki/v2/spaces/get_node", params={"token": node_token})
+    node = data.get("data", {}).get("node", {})
+    return {
+        "space_id": node.get("space_id", ""),
+        "obj_token": node.get("obj_token", ""),  # 文档的实际 token
+        "obj_type": node.get("obj_type", ""),      # docx / doc / sheet
+        "title": node.get("title", ""),
+        "node_token": node.get("node_token", ""),
+    }
+
+
+def _get_wiki_children(node_token: str, page_size: int = 50) -> list[dict]:
+    """获取 Wiki 节点的子节点列表"""
+    all_children = []
+    page_token = ""
+    while True:
+        params: dict[str, Any] = {
+            "page_size": page_size,
+            "parent_node_token": node_token,
+        }
+        if page_token:
+            params["page_token"] = page_token
+
+        data = _feishu_get("/open-apis/wiki/v2/spaces/nodes/get_child_nodes", params=params)
+        children = data.get("data", {}).get("items", [])
+        all_children.extend(children)
+
+        page_token = data.get("data", {}).get("page_token", "")
+        if not page_token:
+            break
+
+    return all_children
+
+
+def _get_space_node_tree(space_id: str) -> list[dict]:
+    """获取知识空间下所有根节点"""
+    all_nodes = []
+    page_token = ""
+    while True:
+        params: dict[str, Any] = {
+            "page_size": 50,
+            "space_id": space_id,
+        }
+        if page_token:
+            params["page_token"] = page_token
+
+        data = _feishu_get("/open-apis/wiki/v2/spaces/nodes", params=params)
+        items = data.get("data", {}).get("items", [])
+        all_nodes.extend(items)
+
+        page_token = data.get("data", {}).get("page_token", "")
+        if not page_token:
+            break
+
+    return all_nodes
+
 
 class FeishuKnowledgeService:
-    _mock_documents = [
-        {
-            "title": "一期产品目标",
-            "keywords": {"一期", "目标", "问答", "登录", "知识库", "飞书", "用户端", "后台"},
-            "content": "一期要交付企业飞书知识库 AI 问答系统，核心包括用户登录、AI 问答、权限内知识库回答和后台管理能力。",
-        },
-        {
-            "title": "权限过滤规则",
-            "keywords": {"权限", "授权", "越权", "用户", "知识库", "过期", "禁用"},
-            "content": "RAG 检索前必须先过滤用户授权知识库，只允许未禁用、未过期、当前用户有权限的知识库参与回答。",
-        },
-        {
-            "title": "无命中兜底规则",
-            "keywords": {"无命中", "兜底", "编造", "检索", "答案", "命中"},
-            "content": "当知识库没有命中相关内容时，系统必须固定返回无命中兜底文案。",
-        },
-        {
-            "title": "技术实现边界",
-            "keywords": {"模型", "prompt", "大模型", "日志", "sse", "流式", "追溯"},
-            "content": "后端需要组装系统 Prompt、检索片段、历史上下文和当前问题，通过 SSE 流式输出，并记录 AI 请求日志。",
-        },
-    ]
+    """飞书知识库检索服务 - 直接调用飞书官方 API"""
+
+    # 文档内容缓存: node_token -> (content, timestamp)
+    _content_cache: dict[str, tuple[str, float]] = {}
+    _CACHE_TTL = 300  # 缓存 5 分钟
 
     @classmethod
-    def retrieve(cls, question: str, scopes: list[KnowledgeScope]) -> list["RetrievedChunk"]:
+    def retrieve(cls, question: str, scopes: list[KnowledgeScope]) -> list[RetrievedChunk]:
         if not scopes:
             return []
 
         settings = get_settings()
         if settings.feishu_mock_enabled:
             return cls._retrieve_mock(question, scopes)
-        return cls._retrieve_remote(question, scopes)
+        return cls._retrieve_from_feishu(question, scopes)
 
     @classmethod
-    def _retrieve_mock(cls, question: str, scopes: list[KnowledgeScope]) -> list["RetrievedChunk"]:
+    def _retrieve_mock(cls, question: str, scopes: list[KnowledgeScope]) -> list[RetrievedChunk]:
         from app.services.rag_service import RetrievedChunk
 
+        mock_documents = [
+            {
+                "title": "一期产品目标",
+                "keywords": {"一期", "目标", "问答", "登录", "知识库", "飞书", "用户端", "后台"},
+                "content": "一期要交付企业飞书知识库 AI 问答系统，核心包括用户登录、AI 问答、基于已开放知识库回答和后台管理能力。",
+            },
+            {
+                "title": "知识库开放过滤规则",
+                "keywords": {"权限", "开放", "用户", "知识库", "禁用", "可见"},
+                "content": "RAG 检索前必须先过滤知识库开放状态，只允许已开放、未禁用的知识库参与回答，已开放知识库默认所有用户可见。",
+            },
+            {
+                "title": "无命中兜底规则",
+                "keywords": {"无命中", "兜底", "编造", "检索", "答案", "命中"},
+                "content": "当知识库没有命中相关内容时，系统必须固定返回无命中兜底文案。",
+            },
+            {
+                "title": "技术实现边界",
+                "keywords": {"模型", "prompt", "大模型", "日志", "sse", "流式", "追溯"},
+                "content": "后端需要组装系统 Prompt、检索片段、历史上下文和当前问题，通过 SSE 流式输出，并记录 AI 请求日志。",
+            },
+        ]
+
         normalized_question = question.lower()
         matched_documents = []
-        for document in cls._mock_documents:
-            if any(keyword.lower() in normalized_question for keyword in document["keywords"]):
+        for document in mock_documents:
+            if any(keyword in normalized_question for keyword in document["keywords"]):
                 matched_documents.append(document)
 
         if not matched_documents:
@@ -72,51 +301,195 @@ class FeishuKnowledgeService:
         ]
 
     @classmethod
-    def _retrieve_remote(cls, question: str, scopes: list[KnowledgeScope]) -> list["RetrievedChunk"]:
+    def _retrieve_from_feishu(cls, question: str, scopes: list[KnowledgeScope]) -> list[RetrievedChunk]:
+        """从飞书知识库检索文档内容"""
         from app.services.rag_service import RetrievedChunk
 
-        settings = get_settings()
-        if not settings.feishu_search_url:
-            raise ExternalServiceError("飞书检索地址未配置", provider="feishu")
+        # 收集所有文档片段
+        all_chunks: list[tuple[KnowledgeScope, str, str, str | None]] = []
 
-        payload = {
-            "query": question,
-            "knowledgeScopes": [
-                {
-                    "id": scope.id,
-                    "spaceId": scope.feishu_space_id,
-                    "nodeId": scope.feishu_node_id,
-                    "name": scope.name,
-                }
-                for scope in scopes
-            ],
-        }
-        data = cls._post_with_retry(settings.feishu_search_url, payload)
-        chunks = data.get("chunks", [])
+        for scope in scopes:
+            try:
+                documents = cls._load_documents(scope)
+            except ExternalServiceError:
+                continue
+
+            for title, content, source_url in documents:
+                all_chunks.append((scope, title, content, source_url))
+
+        if not all_chunks:
+            return []
+
+        # 关键词匹配评分（简单但够用的检索方式）
+        scored_chunks = cls._score_and_rank(question, all_chunks)
+
+        # 返回 top 5 相关片段
         return [
             RetrievedChunk(
-                knowledge_id=int(item.get("knowledgeId") or 0),
-                knowledge_name=str(item.get("knowledgeName") or "飞书知识库"),
-                title=str(item.get("title") or "未命名片段"),
-                content=str(item.get("content") or ""),
-                source_url=item.get("sourceUrl"),
+                knowledge_id=scope.id,
+                knowledge_name=scope.name,
+                title=title,
+                content=content,
+                source_url=source_url,
             )
-            for item in chunks
-            if item.get("content")
+            for scope, title, content, source_url in scored_chunks[:5]
         ]
 
-    @staticmethod
-    def _post_with_retry(url: str, payload: dict[str, Any]) -> dict[str, Any]:
-        settings = get_settings()
-        last_error: Exception | None = None
-        for _ in range(settings.feishu_retry_count + 1):
+    @classmethod
+    def _load_documents(cls, scope: KnowledgeScope) -> list[tuple[str, str, str | None]]:
+        """加载指定知识库 scope 下的所有文档"""
+        node_id = scope.feishu_node_id
+        base_url = f"https://zcn7hk6n047c.feishu.cn"
+
+        if not node_id:
+            return []
+
+        # 1. 判断 node 类型：先查 wiki 节点信息
+        try:
+            node_info = cls._get_node_info_with_cache(node_id)
+        except ExternalServiceError:
+            # 可能是普通 docx token，直接尝试读取
             try:
-                response = httpx.post(url, json=payload, timeout=settings.feishu_timeout_seconds)
-                response.raise_for_status()
-                data = response.json()
-                if not isinstance(data, dict):
-                    raise ExternalServiceError("飞书检索响应格式不正确", provider="feishu")
-                return data
-            except (httpx.HTTPError, ValueError, ExternalServiceError) as exc:
-                last_error = exc
-        raise ExternalServiceError(f"飞书检索失败：{last_error}", provider="feishu")
+                content = cls._get_document_content_with_cache(node_id)
+                title = node_id
+                return [(title, content, f"{base_url}/docx/{node_id}")]
+            except ExternalServiceError:
+                return []
+
+        obj_type = node_info.get("obj_type", "")
+        obj_token = node_info.get("obj_token", "")
+        title = node_info.get("title", node_id)
+        space_id = node_info.get("space_id", scope.feishu_space_id)
+
+        # 如果节点本身就是一个文档
+        if obj_type == "docx":
+            content = cls._get_document_content_with_cache(obj_token)
+            source_url = f"{base_url}/wiki/{node_id}"
+            return [(title, content, source_url)]
+
+        # 如果节点是文件夹，获取子节点
+        children = cls._get_wiki_children_with_cache(node_id)
+
+        documents = []
+        for child in children:
+            child_node_token = child.get("node_token", "")
+            child_obj_type = child.get("obj_type", "")
+            child_obj_token = child.get("obj_token", "")
+            child_title = child.get("title", "")
+
+            if child_obj_type == "docx" and child_obj_token:
+                try:
+                    content = cls._get_document_content_with_cache(child_obj_token)
+                    source_url = f"{base_url}/wiki/{child_node_token}"
+                    documents.append((child_title, content, source_url))
+                except ExternalServiceError:
+                    continue
+
+        # 也读取节点本身的文档（如果有）
+        if obj_type == "docx" and obj_token:
+            try:
+                content = cls._get_document_content_with_cache(obj_token)
+                source_url = f"{base_url}/wiki/{node_id}"
+                documents.insert(0, (title, content, source_url))
+            except ExternalServiceError:
+                pass
+
+        return documents
+
+    @classmethod
+    def _get_node_info_with_cache(cls, node_token: str) -> dict:
+        """带缓存获取节点信息"""
+        cache_key = f"node_info:{node_token}"
+        now = time.time()
+
+        if cache_key in cls._content_cache:
+            cached_data, ts = cls._content_cache[cache_key]
+            if now - ts < cls._CACHE_TTL:
+                return eval(cached_data)  # noqa: S307
+
+        info = _get_wiki_node_info(node_token)
+        cls._content_cache[cache_key] = (str(info), now)
+        return info
+
+    @classmethod
+    def _get_document_content_with_cache(cls, doc_token: str) -> str:
+        """带缓存读取文档内容"""
+        now = time.time()
+
+        if doc_token in cls._content_cache:
+            content, ts = cls._content_cache[doc_token]
+            if now - ts < cls._CACHE_TTL:
+                return content
+
+        content = _get_docx_content(doc_token)
+        cls._content_cache[doc_token] = (content, now)
+        return content
+
+    @classmethod
+    def _get_wiki_children_with_cache(cls, node_token: str) -> list[dict]:
+        """带缓存获取子节点"""
+        cache_key = f"children:{node_token}"
+        now = time.time()
+
+        if cache_key in cls._content_cache:
+            cached_data, ts = cls._content_cache[cache_key]
+            if now - ts < cls._CACHE_TTL:
+                return eval(cached_data)  # noqa: S307
+
+        children = _get_wiki_children(node_token)
+        cls._content_cache[cache_key] = (str(children), now)
+        return children
+
+    @classmethod
+    def _score_and_rank(
+        cls,
+        question: str,
+        chunks: list[tuple[KnowledgeScope, str, str, str | None]],
+    ) -> list[tuple[KnowledgeScope, str, str, str | None]]:
+        """基于关键词匹配对文档片段打分排序"""
+        question_lower = question.lower()
+
+        # 提取问题中的关键词（按字/词拆分）
+        question_chars = set(question_lower)
+        # 2-gram
+        question_bigrams = set(question_lower[i : i + 2] for i in range(len(question_lower) - 1))
+        # 3-gram
+        question_trigrams = set(question_lower[i : i + 3] for i in range(len(question_lower) - 2))
+
+        scored = []
+        for scope, title, content, source_url in chunks:
+            content_lower = content.lower()
+            title_lower = title.lower()
+
+            # 计算匹配分
+            score = 0.0
+
+            # 标题匹配（权重高）
+            for char in question_chars:
+                if char in title_lower:
+                    score += 3.0
+            for bigram in question_bigrams:
+                if bigram in title_lower:
+                    score += 5.0
+            for trigram in question_trigrams:
+                if trigram in title_lower:
+                    score += 8.0
+
+            # 内容匹配
+            for char in question_chars:
+                if char in content_lower:
+                    score += 1.0
+            for bigram in question_bigrams:
+                if bigram in content_lower:
+                    score += 2.0
+            for trigram in question_trigrams:
+                if trigram in content_lower:
+                    score += 3.0
+
+            if score > 0:
+                scored.append((scope, title, content, source_url, score))
+
+        # 按分数降序排列
+        scored.sort(key=lambda x: x[4], reverse=True)
+
+        return [(s, t, c, u) for s, t, c, u, _ in scored]

ai_knowledge_base_v2/apps/backend/app/services/knowledge_service.py

@@ -1,13 +1,12 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from datetime import UTC, datetime
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from app.core.config import get_settings
-from app.models.knowledge import Knowledge, UserKnowledgePermission
+from app.models.knowledge import Knowledge
 from app.models.user import User
 
 
@@ -23,20 +22,7 @@ class KnowledgeScope:
 class KnowledgeAccessService:
     @staticmethod
     def get_allowed_knowledge(db: Session, user: User) -> list[KnowledgeScope]:
-        now = datetime.now(UTC).replace(tzinfo=None)
-        rows = db.execute(
-            select(Knowledge)
-            .join(UserKnowledgePermission, UserKnowledgePermission.knowledge_id == Knowledge.id)
-            .where(
-                UserKnowledgePermission.user_id == user.id,
-                Knowledge.status == 1,
-                (UserKnowledgePermission.effective_at.is_(None))
-                | (UserKnowledgePermission.effective_at <= now),
-                (UserKnowledgePermission.expired_at.is_(None))
-                | (UserKnowledgePermission.expired_at >= now),
-            )
-            .order_by(Knowledge.id.asc())
-        ).scalars()
+        rows = db.scalars(select(Knowledge).where(Knowledge.status == 1).order_by(Knowledge.id.asc()))
         scopes = [
             KnowledgeScope(
                 id=knowledge.id,

ai_knowledge_base_v2/apps/backend/app/services/model_service.py

@@ -85,7 +85,7 @@ def _mock_answer(rag_result: RagResult) -> str:
         f"- {chunk.content}\n  来源：{chunk.knowledge_name} / {chunk.title}" for chunk in rag_result.chunks
     )
     return (
-        "根据当前已授权知识库，整理到的信息如下：\n\n"
+        "根据当前已开放知识库，整理到的信息如下：\n\n"
         f"{summaries}\n\n"
         "当前仍是 mock RAG + mock 模型阶段，后续会把检索服务替换为飞书实时检索，"
         "把模型服务替换为后台启用的大模型配置。"
@@ -136,6 +136,8 @@ def _call_configured_model(model: ModelConfig, rag_result: RagResult, *, allow_n
         return _call_anthropic_messages(model, rag_result)
     if api_type == "gemini_generate_content":
         return _call_gemini_generate_content(model, rag_result)
+    if api_type == "minimax":
+        return _call_minimax(model, rag_result)
     return _call_openai_compatible_model(model, rag_result)
 
 
@@ -166,7 +168,8 @@ def _call_openai_compatible_model(model: ModelConfig, rag_result: RagResult) ->
             timeout=model.timeout_second,
         )
         response.raise_for_status()
-        return _extract_openai_answer(response.json())
+        data = response.json()
+        return _extract_openai_answer(data)
     except (httpx.HTTPError, ValueError, KeyError, TypeError) as exc:
         raise ExternalServiceError(f"模型调用失败：{exc}", provider="model") from exc
 
@@ -225,7 +228,15 @@ def _call_gemini_generate_content(model: ModelConfig, rag_result: RagResult) ->
             timeout=model.timeout_second,
         )
         response.raise_for_status()
-        return _extract_gemini_answer(response.json())
+        data = response.json()
+        # Gemini 有时返回 HTTP 200 但 body 里有 error
+        if "error" in data:
+            err = data["error"]
+            msg = err.get("message", str(err))
+            raise ExternalServiceError(f"Gemini 调用失败：{msg}", provider="model")
+        return _extract_gemini_answer(data)
+    except ExternalServiceError:
+        raise
     except (httpx.HTTPError, ValueError, KeyError, TypeError) as exc:
         raise ExternalServiceError(f"模型调用失败：{exc}", provider="model") from exc
 
@@ -260,6 +271,64 @@ def _extract_anthropic_answer(data: dict[str, Any]) -> str:
     return answer
 
 
+def _call_minimax(model: ModelConfig, rag_result: RagResult) -> str:
+    """调用 MiniMax 官方 API (非 OpenAI 兼容协议)"""
+    import urllib.parse
+
+    extra = _load_extra_params(model.extra_params)
+    group_id = extra.pop("group_id", "")
+    if not group_id:
+        raise ExternalServiceError("MiniMax 调用需要 group_id，请在模型高级参数中配置：{\"group_id\": \"xxx\"}", provider="model")
+
+    base_url = (model.base_url or "https://api.minimaxi.com/v1/text/chatcompletion_pro").rstrip("/")
+    # MiniMax 的 baseUrl 在快速添加模板里已经包含完整路径，但如果用户自定义 baseUrl 没有 query，需要拼接
+    if "?" not in base_url:
+        base_url = f"{base_url}?GroupId={urllib.parse.quote(group_id)}"
+    elif "GroupId" not in base_url:
+        base_url = f"{base_url}&GroupId={urllib.parse.quote(group_id)}"
+
+    payload: dict[str, Any] = {
+        "model": model.model_name,
+        "tokens_to_generate": model.max_token or 1024,
+        "reply_constraints": {"sender_type": "BOT", "sender_name": "AI知识库助手"},
+        "messages": [
+            {"sender_type": "USER", "sender_name": "用户", "text": rag_result.prompt},
+        ],
+        "bot_setting": [
+            {
+                "bot_name": "AI知识库助手",
+                "content": "你是企业知识库问答助手，只能基于已提供的知识片段回答。",
+            }
+        ],
+    }
+    _put_if_not_none(payload, "temperature", _decimal_to_float(model.temperature))
+    _put_if_not_none(payload, "top_p", _decimal_to_float(model.top_p))
+
+    headers = {"Authorization": f"Bearer {model.api_key}", "Content-Type": "application/json"}
+
+    try:
+        response = httpx.post(base_url, json=payload, headers=headers, timeout=model.timeout_second)
+        response.raise_for_status()
+        return _extract_minimax_answer(response.json())
+    except (httpx.HTTPError, ValueError, KeyError, TypeError) as exc:
+        raise ExternalServiceError(f"MiniMax 模型调用失败：{exc}", provider="model") from exc
+
+
+def _extract_minimax_answer(data: dict[str, Any]) -> str:
+    choices = data.get("choices")
+    if not isinstance(choices, list) or not choices:
+        raise ValueError("MiniMax 响应缺少 choices")
+    first = choices[0]
+    messages = first.get("messages", [])
+    if not isinstance(messages, list):
+        raise ValueError("MiniMax 响应缺少 messages")
+    texts = [m.get("text", "") for m in messages if isinstance(m, dict)]
+    answer = "".join(texts).strip()
+    if not answer:
+        raise ValueError("MiniMax 响应缺少回答内容")
+    return answer
+
+
 def _extract_gemini_answer(data: dict[str, Any]) -> str:
     candidates = data.get("candidates")
     if not isinstance(candidates, list) or not candidates: