332 lines
12 KiB
Python
332 lines
12 KiB
Python
"""
|
||
LLM 服务模块
|
||
提供统一的 LLM 调用接口，用于查询扩展、标签提取等任务
|
||
支持 MiniMax / DeepSeek / OpenAI（均为 OpenAI 兼容格式）
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import httpx
|
||
import json
|
||
import logging
|
||
import re
|
||
from typing import List, Optional
|
||
|
||
from app.config import settings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class LLMService:
|
||
"""
|
||
LLM 服务类
|
||
根据配置的 LLM_PROVIDER 自动选择对应的 API 端点。
|
||
所有支持的提供商均兼容 OpenAI Chat Completions 格式。
|
||
"""
|
||
|
||
_client = None
|
||
|
||
@classmethod
|
||
def _get_client(cls):
|
||
"""延迟初始化 OpenAI 兼容客户端"""
|
||
if cls._client is not None:
|
||
return cls._client
|
||
|
||
from openai import AsyncOpenAI
|
||
|
||
provider = settings.LLM_PROVIDER.lower()
|
||
|
||
if provider == "minimax":
|
||
cls._client = AsyncOpenAI(
|
||
api_key=settings.MINIMAX_API_KEY or "EMPTY",
|
||
base_url=settings.MINIMAX_BASE_URL,
|
||
)
|
||
cls._model = settings.MINIMAX_CHAT_MODEL
|
||
elif provider == "deepseek":
|
||
cls._client = AsyncOpenAI(
|
||
api_key=settings.DEEPSEEK_API_KEY or "EMPTY",
|
||
base_url=settings.DEEPSEEK_BASE_URL,
|
||
)
|
||
cls._model = settings.DEEPSEEK_OCR_MODEL
|
||
elif provider == "openai":
|
||
cls._client = AsyncOpenAI(
|
||
api_key=settings.OPENAI_API_KEY or "EMPTY",
|
||
base_url=settings.OPENAI_BASE_URL,
|
||
)
|
||
cls._model = settings.EMBEDDING_MODEL
|
||
else:
|
||
raise ValueError(f"不支持的 LLM 提供商: {provider}")
|
||
|
||
logger.info("LLM 服务初始化完成: provider=%s, model=%s", provider, cls._model)
|
||
return cls._client
|
||
|
||
@staticmethod
|
||
def _clean_reasoning_content(content: str) -> str:
|
||
"""清理推理模型的思考过程，只保留最终回答"""
|
||
import re
|
||
|
||
# 方式1: 清理 🧠...💠 标签（MiniMax M2.7 reasoning 模式）
|
||
content = re.sub(r'🧠[\s\S]*?💠', '', content)
|
||
|
||
# 方式2: 清理 💭(U+1F4AD) 包裹的思考块
|
||
if '💭' in content:
|
||
parts = content.split('💭')
|
||
answer_parts = [parts[i] for i in range(len(parts)) if i % 2 == 1]
|
||
if answer_parts:
|
||
content = ''.join(answer_parts)
|
||
|
||
# 方式3: 清理 </think 标签（M2.7 有时使用）
|
||
think_end = content.find('</think')
|
||
if think_end >= 0:
|
||
after = content[think_end + len('</think'):]
|
||
after = re.sub(r'^[>\n\s]*', '', after)
|
||
content = after
|
||
|
||
# 方式4: 清理 \x0f 分隔的思考块
|
||
if '\x0f' in content:
|
||
parts = content.split('\x0f')
|
||
content = ''.join(parts[1::2]) if len(parts) > 1 else parts[-1] if parts else content
|
||
|
||
# 方式5: 清理 0x00 等特殊字符
|
||
content = content.replace('\x00', '').strip()
|
||
|
||
# 方式6: 如果内容仍然很长（说明还有思考残留），取最后一个短段落作为答案
|
||
if len(content) > 100 and '\n\n' in content:
|
||
segments = [s.strip() for s in content.split('\n\n')]
|
||
for seg in reversed(segments):
|
||
if seg and len(seg) <= 100:
|
||
content = seg
|
||
break
|
||
|
||
return content.strip()
|
||
|
||
@classmethod
|
||
async def chat(cls, messages: List[dict], temperature: float = 0.3, max_tokens: int = 1024) -> str:
|
||
"""
|
||
发送聊天请求。
|
||
|
||
Args:
|
||
messages: 消息列表，格式 [{"role": "system/user/assistant", "content": "..."}]
|
||
temperature: 采样温度
|
||
max_tokens: 最大生成 token 数
|
||
|
||
Returns:
|
||
模型回复文本（已清理思考过程）
|
||
"""
|
||
client = cls._get_client()
|
||
|
||
# MiniMax 模型默认开启思考模式，使用 reasoning_split 将思考内容
|
||
# 分离到 reasoning_details 字段，content 只保留最终回答
|
||
extra_body = {}
|
||
if settings.LLM_PROVIDER.lower() == "minimax":
|
||
extra_body["reasoning_split"] = True
|
||
|
||
response = await client.chat.completions.create(
|
||
model=cls._model,
|
||
messages=messages,
|
||
temperature=temperature,
|
||
max_tokens=max_tokens,
|
||
**({"extra_body": extra_body} if extra_body else {}),
|
||
)
|
||
content = response.choices[0].message.content or ""
|
||
|
||
# 兜底清理：以防 reasoning_split 未生效或使用了其他模型
|
||
content = cls._clean_reasoning_content(content)
|
||
|
||
return content
|
||
|
||
@classmethod
|
||
async def expand_query(cls, query: str) -> List[str]:
|
||
"""
|
||
查询扩展：将原始查询改写为多个语义相近的表述。
|
||
用于混合搜索时提高召回率。
|
||
|
||
Args:
|
||
query: 原始查询文本
|
||
|
||
Returns:
|
||
扩展后的查询列表（包含原始查询）
|
||
"""
|
||
messages = [
|
||
{
|
||
"role": "system",
|
||
"content": (
|
||
"你是一个搜索查询优化助手。用户会给你一个搜索查询，"
|
||
"你需要生成 2 个语义相近但表述不同的替代查询。"
|
||
"只输出 JSON 数组格式，不要添加任何其他文字。"
|
||
'例如：["原始查询", "替代查询1", "替代查询2"]'
|
||
),
|
||
},
|
||
{
|
||
"role": "user",
|
||
"content": f"请为以下查询生成 2 个替代表述：{query}",
|
||
},
|
||
]
|
||
|
||
try:
|
||
result = await cls.chat(messages, temperature=0.3, max_tokens=256)
|
||
# 尝试解析 JSON
|
||
result = result.strip()
|
||
if result.startswith("```"):
|
||
result = result.split("\n", 1)[-1]
|
||
if result.endswith("```"):
|
||
result = result[:-3]
|
||
result = result.strip()
|
||
queries = json.loads(result)
|
||
if isinstance(queries, list):
|
||
# 去重并限制数量
|
||
seen = set()
|
||
unique = []
|
||
for q in queries:
|
||
q = str(q).strip()
|
||
if q and q not in seen:
|
||
seen.add(q)
|
||
unique.append(q)
|
||
return unique[:3]
|
||
except Exception as exc:
|
||
logger.warning("查询扩展失败: %s，使用原始查询", exc)
|
||
|
||
return [query]
|
||
|
||
@classmethod
|
||
async def detect_entities(cls, text: str) -> List[dict]:
|
||
"""
|
||
实体检测：从文本中提取知识点、人物、概念等实体。
|
||
|
||
Args:
|
||
text: 待分析的文本
|
||
|
||
Returns:
|
||
实体列表，格式 [{"name": "...", "type": "knowledge_point|person|concept|technique", "description": "..."}]
|
||
"""
|
||
messages = [
|
||
{
|
||
"role": "system",
|
||
"content": (
|
||
"你是一个教育内容分析助手。用户会给你一段直播课的文字稿，"
|
||
"你需要从中提取关键实体（知识点、技巧、人物、概念等）。"
|
||
"只输出 JSON 数组格式，不要添加任何其他文字。\n"
|
||
'格式：[{"name": "实体名称", "type": "knowledge_point|technique|person|concept", "description": "简要描述"}]\n'
|
||
"type 说明：\n"
|
||
"- knowledge_point: 知识点（如公式、定理、定义）\n"
|
||
"- technique: 技巧/方法（如解题技巧、记忆方法）\n"
|
||
"- person: 人物\n"
|
||
"- concept: 概念/术语"
|
||
),
|
||
},
|
||
{
|
||
"role": "user",
|
||
"content": f"请从以下文字中提取关键实体：\n\n{text[:3000]}",
|
||
},
|
||
]
|
||
|
||
try:
|
||
result = await cls.chat(messages, temperature=0.1, max_tokens=2048)
|
||
result = result.strip()
|
||
if result.startswith("```"):
|
||
result = result.split("\n", 1)[-1]
|
||
if result.endswith("```"):
|
||
result = result[:-3]
|
||
result = result.strip()
|
||
entities = json.loads(result)
|
||
if isinstance(entities, list):
|
||
return entities
|
||
except Exception as exc:
|
||
logger.warning("实体检测失败: %s", exc)
|
||
|
||
return []
|
||
|
||
@classmethod
|
||
async def judge_batch(cls, query: str, articles: List[dict]) -> List[int]:
|
||
"""
|
||
用 LLM 批量判断哪些文章符合查询。
|
||
|
||
Args:
|
||
query: 用户搜索查询
|
||
articles: [{"id": int, "text": str}, ...]，最多10个
|
||
|
||
Returns:
|
||
匹配的文章 id 列表（空列表表示都不匹配）
|
||
"""
|
||
if not articles:
|
||
return []
|
||
|
||
# 构建文章列表
|
||
articles_text = ""
|
||
for i, article in enumerate(articles, 1):
|
||
text = (article.get("text") or "")[:800]
|
||
articles_text += f"{i}. {text}\n"
|
||
|
||
messages = [
|
||
{
|
||
"role": "system",
|
||
"content": (
|
||
"你是一个严格的语义搜索匹配引擎。用户会给你一个搜索查询和多篇文章片段，"
|
||
"你需要判断哪些文章与查询在**核心语义上高度相关**。\n\n"
|
||
"判断原则：\n"
|
||
"- 文章的**核心主题**必须与查询直接相关，而不是仅仅提及了查询中的某个词\n"
|
||
"- 例如：查询\"爸爸出轨\"，文章只提到\"爸爸带孩子玩\"是不相关的，"
|
||
"只有文章讨论出轨、婚姻危机、夫妻矛盾等才算相关\n"
|
||
"- 例如：查询\"孩子不想上学\"，文章提到\"孩子打游戏不去学校\"是相关的，"
|
||
"但只提到\"孩子\"或\"学校\"其他方面则不相关\n\n"
|
||
"不相关的典型情况（必须排除）：\n"
|
||
"- 仅包含查询中的某个词，但讨论的是完全不同的话题\n"
|
||
"- 主题相近但具体内容无关（如查询出轨，文章讲的是正常的夫妻相处）\n"
|
||
"- 只有一个宽泛的关联词，没有实质性的语义联系\n\n"
|
||
"只输出匹配的文章编号，用逗号分隔。\n"
|
||
"如果没有文章与查询相关，只输出 0。\n"
|
||
"不要输出任何解释。宁可漏判，不可误判。"
|
||
),
|
||
},
|
||
{
|
||
"role": "user",
|
||
"content": f"搜索查询: {query}\n\n{articles_text}\n请输出匹配的文章编号:",
|
||
},
|
||
]
|
||
|
||
try:
|
||
from openai import AsyncOpenAI
|
||
http_client = httpx.AsyncClient(timeout=60.0)
|
||
try:
|
||
client = AsyncOpenAI(
|
||
api_key=settings.MINIMAX_API_KEY,
|
||
base_url=settings.MINIMAX_BASE_URL,
|
||
http_client=http_client,
|
||
)
|
||
response = await client.chat.completions.create(
|
||
model=settings.MINIMAX_CHAT_MODEL,
|
||
messages=messages,
|
||
temperature=0.0,
|
||
max_tokens=512,
|
||
stream=False,
|
||
extra_body={"reasoning_split": True},
|
||
)
|
||
content = response.choices[0].message.content or ""
|
||
content = cls._clean_reasoning_content(content)
|
||
|
||
# 解析返回的编号
|
||
numbers = re.findall(r'\d+', content)
|
||
|
||
if not numbers:
|
||
return []
|
||
|
||
matched_ids = []
|
||
for num_str in numbers:
|
||
num = int(num_str)
|
||
if num == 0:
|
||
continue # 0 表示都不匹配
|
||
if 1 <= num <= len(articles):
|
||
matched_ids.append(articles[num - 1]["id"])
|
||
|
||
return matched_ids
|
||
finally:
|
||
await http_client.aclose()
|
||
except Exception as exc:
|
||
logger.warning("LLM 批量判断失败: %s", exc)
|
||
return []
|
||
|
||
@classmethod
|
||
def reset(cls) -> None:
|
||
"""重置客户端（用于测试或切换配置）"""
|
||
cls._client = None
|