Add V2 external service providers

This commit is contained in:
2026-07-06 18:03:29 +08:00
parent 86cbed2437
commit 301661e7b4
13 changed files with 318 additions and 55 deletions

View File

@@ -33,6 +33,11 @@ class Settings(BaseSettings):
mock_sms_code: str = "123456"
sms_code_expire_minutes: int = 5
mock_rag_enabled: bool = True
mock_model_enabled: bool = True
feishu_mock_enabled: bool = True
feishu_search_url: str = ""
feishu_timeout_seconds: int = 20
feishu_retry_count: int = 2
default_daily_chat_limit: int = 100
default_user_name_prefix: str = "用户"

View File

@@ -37,3 +37,32 @@ class AiRequestLogService:
status="SUCCESS",
)
)
@staticmethod
def write_failed(
db: Session,
*,
session_id: int,
message_id: int | None,
user_id: int,
model_name: str | None,
prompt: str | None,
knowledge_ids: str | None,
retrieve_count: int,
cost_ms: int,
error_message: str,
) -> None:
db.add(
AiRequestLog(
session_id=session_id,
message_id=message_id,
user_id=user_id,
model_name=model_name,
prompt=prompt,
knowledge_ids=knowledge_ids,
retrieve_count=retrieve_count,
cost_ms=cost_ms,
status="FAILED",
error_message=error_message,
)
)

View File

@@ -10,6 +10,7 @@ from sqlalchemy.orm import Session
from app.models.chat import ChatMessage, ChatSession
from app.models.user import User
from app.services.ai_request_log_service import AiRequestLogService
from app.services.external_errors import ExternalServiceError
from app.services.model_service import ModelClientService
from app.services.rag_service import RagService
@@ -86,8 +87,28 @@ class ChatService:
db.flush()
started_at = perf_counter()
rag_result = RagService.build_result(db, user, normalized_question)
completion = ModelClientService.complete(db, rag_result)
try:
rag_result = RagService.build_result(db, user, normalized_question)
completion = ModelClientService.complete(db, rag_result)
except ExternalServiceError as exc:
cost_ms = int((perf_counter() - started_at) * 1000)
AiRequestLogService.write_failed(
db,
session_id=session.id,
message_id=user_message.id,
user_id=user.id,
model_name=None,
prompt=normalized_question,
knowledge_ids=None,
retrieve_count=0,
cost_ms=cost_ms,
error_message=str(exc),
)
db.commit()
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail="外部服务暂时不可用,请稍后再试",
) from exc
cost_ms = int((perf_counter() - started_at) * 1000)
assistant_message = ChatMessage(

View File

@@ -0,0 +1,7 @@
from __future__ import annotations
class ExternalServiceError(RuntimeError):
def __init__(self, message: str, *, provider: str) -> None:
super().__init__(message)
self.provider = provider

View File

@@ -0,0 +1,122 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any
import httpx
from app.core.config import get_settings
from app.services.external_errors import ExternalServiceError
from app.services.knowledge_service import KnowledgeScope
if TYPE_CHECKING:
from app.services.rag_service import RetrievedChunk
class FeishuKnowledgeService:
_mock_documents = [
{
"title": "一期产品目标",
"keywords": {"一期", "目标", "问答", "登录", "知识库", "飞书", "用户端", "后台"},
"content": "一期要交付企业飞书知识库 AI 问答系统核心包括用户登录、AI 问答、权限内知识库回答和后台管理能力。",
},
{
"title": "权限过滤规则",
"keywords": {"权限", "授权", "越权", "用户", "知识库", "过期", "禁用"},
"content": "RAG 检索前必须先过滤用户授权知识库,只允许未禁用、未过期、当前用户有权限的知识库参与回答。",
},
{
"title": "无命中兜底规则",
"keywords": {"无命中", "兜底", "编造", "检索", "答案", "命中"},
"content": "当知识库没有命中相关内容时,系统必须固定返回无命中兜底文案。",
},
{
"title": "技术实现边界",
"keywords": {"模型", "prompt", "大模型", "日志", "sse", "流式", "追溯"},
"content": "后端需要组装系统 Prompt、检索片段、历史上下文和当前问题通过 SSE 流式输出,并记录 AI 请求日志。",
},
]
@classmethod
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)
@classmethod
def _retrieve_mock(cls, question: str, scopes: list[KnowledgeScope]) -> list["RetrievedChunk"]:
from app.services.rag_service import RetrievedChunk
normalized_question = question.lower()
matched_documents = []
for document in cls._mock_documents:
if any(keyword.lower() in normalized_question for keyword in document["keywords"]):
matched_documents.append(document)
if not matched_documents:
return []
primary_scope = scopes[0]
return [
RetrievedChunk(
knowledge_id=primary_scope.id,
knowledge_name=primary_scope.name,
title=document["title"],
content=document["content"],
source_url=None,
)
for document in matched_documents[:3]
]
@classmethod
def _retrieve_remote(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")
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", [])
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"),
)
for item in chunks
if item.get("content")
]
@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):
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")

View File

@@ -1,11 +1,15 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import httpx
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.config import get_settings
from app.models.ai_config import ModelConfig
from app.services.external_errors import ExternalServiceError
from app.services.rag_service import NO_HIT_ANSWER, RagResult
@@ -27,8 +31,14 @@ class ModelClientService:
.order_by(ModelConfig.id.desc())
.limit(1)
)
model_name = model.model_name if model is not None else "mock-model"
answer = _mock_answer(rag_result)
settings = get_settings()
if settings.mock_model_enabled or model is None:
model_name = model.model_name if model is not None else "mock-model"
answer = _mock_answer(rag_result)
else:
model_name = model.model_name
answer = _call_openai_compatible_model(model, rag_result)
return ModelCompletion(
answer=answer,
model_id=model.id if model is not None else None,
@@ -55,3 +65,55 @@ def _mock_answer(rag_result: RagResult) -> str:
def _rough_token_count(text: str) -> int:
return max(1, len(text.strip()) // 2)
def _call_openai_compatible_model(model: ModelConfig, rag_result: RagResult) -> str:
if not rag_result.is_hit:
return NO_HIT_ANSWER
if not model.api_url or not model.api_key:
raise ExternalServiceError("模型 API URL 或 API Key 未配置", provider="model")
payload = {
"model": model.model_name,
"messages": [
{"role": "system", "content": "你是企业知识库问答助手,只能基于已提供的知识片段回答。"},
{"role": "user", "content": rag_result.prompt},
],
"temperature": float(model.temperature) if model.temperature is not None else 0.2,
"max_tokens": model.max_token or 1024,
"stream": False,
}
headers = {
"Authorization": f"Bearer {model.api_key}",
"Content-Type": "application/json",
}
try:
response = httpx.post(
model.api_url,
json=payload,
headers=headers,
timeout=model.timeout_second,
)
response.raise_for_status()
return _extract_answer(response.json())
except (httpx.HTTPError, ValueError, KeyError, TypeError) as exc:
raise ExternalServiceError(f"模型调用失败:{exc}", provider="model") from exc
def _extract_answer(data: dict[str, Any]) -> str:
choices = data.get("choices")
if not isinstance(choices, list) or not choices:
raise ValueError("模型响应缺少 choices")
first_choice = choices[0]
if not isinstance(first_choice, dict):
raise ValueError("模型响应 choices 格式不正确")
message = first_choice.get("message")
if isinstance(message, dict) and message.get("content"):
return str(message["content"])
if first_choice.get("text"):
return str(first_choice["text"])
raise ValueError("模型响应缺少回答内容")

View File

@@ -7,6 +7,7 @@ from sqlalchemy.orm import Session
from app.models.ai_config import Prompt
from app.models.user import User
from app.services.feishu_service import FeishuKnowledgeService
from app.services.knowledge_service import KnowledgeAccessService, KnowledgeScope
NO_HIT_ANSWER = "当前知识库中未检索到相关内容,请联系管理员补充相关知识。"
@@ -46,57 +47,6 @@ class RagService:
return RagResult(question=question, knowledge_scopes=scopes, chunks=chunks, prompt=prompt)
class FeishuKnowledgeService:
_mock_documents = [
{
"title": "一期产品目标",
"keywords": {"一期", "目标", "问答", "登录", "知识库", "飞书", "用户端", "后台"},
"content": "一期要交付企业飞书知识库 AI 问答系统核心包括用户登录、AI 问答、权限内知识库回答和后台管理能力。",
},
{
"title": "权限过滤规则",
"keywords": {"权限", "授权", "越权", "用户", "知识库", "过期", "禁用"},
"content": "RAG 检索前必须先过滤用户授权知识库,只允许未禁用、未过期、当前用户有权限的知识库参与回答。",
},
{
"title": "无命中兜底规则",
"keywords": {"无命中", "兜底", "编造", "检索", "答案", "命中"},
"content": f"当知识库没有命中相关内容时,系统必须固定返回:{NO_HIT_ANSWER}",
},
{
"title": "技术实现边界",
"keywords": {"模型", "prompt", "大模型", "日志", "sse", "流式", "追溯"},
"content": "后端需要组装系统 Prompt、检索片段、历史上下文和当前问题通过 SSE 流式输出,并记录 AI 请求日志。",
},
]
@classmethod
def retrieve(cls, question: str, scopes: list[KnowledgeScope]) -> list[RetrievedChunk]:
if not scopes:
return []
normalized_question = question.lower()
matched_documents = []
for document in cls._mock_documents:
if any(keyword.lower() in normalized_question for keyword in document["keywords"]):
matched_documents.append(document)
if not matched_documents:
return []
primary_scope = scopes[0]
return [
RetrievedChunk(
knowledge_id=primary_scope.id,
knowledge_name=primary_scope.name,
title=document["title"],
content=document["content"],
source_url=None,
)
for document in matched_documents[:3]
]
class PromptService:
_default_prompt = (
"你是企业飞书知识库 AI 助手。你只能基于提供的知识片段回答,不能编造。"