574 lines
19 KiB
Python
574 lines
19 KiB
Python
"""
|
|
LLM 服务
|
|
"""
|
|
import logging
|
|
from typing import AsyncGenerator, Dict, List, Optional
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from ai_platform.providers import (
|
|
BaseLLMProvider,
|
|
LLMConfig,
|
|
LLMMessage,
|
|
LLMResponse,
|
|
ProviderRegistry,
|
|
)
|
|
from ai_platform.providers.base import LLMStreamChunk, ToolDefinition
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class LLMService:
|
|
"""
|
|
LLM 服务
|
|
|
|
统一管理 LLM 调用，支持多种提供商
|
|
"""
|
|
|
|
def __init__(self, db: Optional[AsyncSession] = None):
|
|
self._provider_cache: Dict[str, BaseLLMProvider] = {}
|
|
self._db = db
|
|
|
|
@staticmethod
|
|
def _is_missing_model_id(model_id: Optional[str]) -> bool:
|
|
if model_id is None:
|
|
return True
|
|
return str(model_id).strip().lower() in {"", "none", "null", "undefined"}
|
|
|
|
async def resolve_chat_model_id(self, model_id: Optional[str]) -> str:
|
|
"""解析 chat 模型。未指定时使用当前启用的默认 chat 模型。"""
|
|
if not self._is_missing_model_id(model_id):
|
|
return str(model_id)
|
|
|
|
if not self._db:
|
|
raise ValueError("未找到可用的 chat 模型，请先在模型配置中启用一个模型")
|
|
|
|
from ai_platform.models import LLMModel, LLMProvider
|
|
|
|
result = await self._db.execute(
|
|
select(LLMModel)
|
|
.join(LLMProvider, LLMProvider.id == LLMModel.provider_id)
|
|
.where(
|
|
LLMModel.is_deleted == False,
|
|
LLMModel.is_active == True,
|
|
LLMModel.model_type == "chat",
|
|
LLMProvider.is_deleted == False,
|
|
LLMProvider.is_active == True,
|
|
)
|
|
.order_by(LLMModel.sort.desc(), LLMModel.sys_create_datetime.desc())
|
|
)
|
|
model = result.scalars().first()
|
|
if not model:
|
|
raise ValueError("未找到可用的 chat 模型，请先在模型配置中启用一个模型")
|
|
|
|
logger.info("No model_id supplied, fallback to default chat model %s", model.id)
|
|
return str(model.id)
|
|
|
|
@staticmethod
|
|
def _run_async_sync(coro):
|
|
import asyncio
|
|
import concurrent.futures
|
|
|
|
try:
|
|
loop = asyncio.get_event_loop()
|
|
except RuntimeError:
|
|
return asyncio.run(coro)
|
|
|
|
if loop.is_running():
|
|
with concurrent.futures.ThreadPoolExecutor() as executor:
|
|
return executor.submit(asyncio.run, coro).result()
|
|
|
|
return loop.run_until_complete(coro)
|
|
|
|
def resolve_chat_model_id_sync(self, model_id: Optional[str]) -> str:
|
|
"""同步解析 chat 模型，供同步节点和流式生成器复用。"""
|
|
return self._run_async_sync(self.resolve_chat_model_id(model_id))
|
|
|
|
async def _get_provider_async(self, model_id: str) -> tuple:
|
|
"""
|
|
异步获取模型对应的提供商实例
|
|
|
|
Args:
|
|
model_id: 模型 ID
|
|
|
|
Returns:
|
|
(provider, model_name)
|
|
"""
|
|
from ai_platform.models import LLMModel, LLMProvider
|
|
|
|
if not self._db:
|
|
raise ValueError("数据库会话未初始化")
|
|
|
|
model_id = await self.resolve_chat_model_id(model_id)
|
|
|
|
# 查询模型
|
|
result = await self._db.execute(
|
|
select(LLMModel).where(
|
|
LLMModel.id == model_id,
|
|
LLMModel.is_active == True,
|
|
LLMModel.is_deleted == False
|
|
)
|
|
)
|
|
model = result.scalar_one_or_none()
|
|
if not model:
|
|
raise ValueError(f'模型不存在或已禁用: {model_id}')
|
|
|
|
# 查询提供商
|
|
provider_result = await self._db.execute(
|
|
select(LLMProvider).where(LLMProvider.id == model.provider_id)
|
|
)
|
|
provider = provider_result.scalar_one_or_none()
|
|
if not provider or not provider.is_active:
|
|
raise ValueError(f'提供商不存在或已禁用')
|
|
|
|
# 缓存提供商实例
|
|
cache_key = str(provider.id)
|
|
if cache_key not in self._provider_cache:
|
|
provider_instance = ProviderRegistry.create_instance(
|
|
provider_type=provider.provider_type,
|
|
api_key=provider.api_key,
|
|
api_base=provider.api_base or "",
|
|
ollama_host=provider.ollama_host or "",
|
|
)
|
|
if not provider_instance:
|
|
raise ValueError(f'不支持的提供商类型: {provider.provider_type}')
|
|
self._provider_cache[cache_key] = provider_instance
|
|
|
|
provider_instance = self._provider_cache[cache_key]
|
|
setattr(provider_instance, 'config_name', provider.name or provider.provider_type)
|
|
setattr(provider_instance, 'config_id', str(provider.id))
|
|
setattr(provider_instance, 'config_api_base', provider.api_base or '')
|
|
|
|
return provider_instance, model.model_name
|
|
|
|
def _get_provider_sync(self, model_id: str, model_data: dict) -> tuple:
|
|
"""
|
|
同步获取提供商实例（使用预加载的数据）
|
|
|
|
Args:
|
|
model_id: 模型 ID
|
|
model_data: 预加载的模型和提供商数据
|
|
|
|
Returns:
|
|
(provider, model_name)
|
|
"""
|
|
provider_type = model_data.get('provider_type')
|
|
api_key = model_data.get('api_key', '')
|
|
api_base = model_data.get('api_base', '')
|
|
ollama_host = model_data.get('ollama_host', '')
|
|
model_name = model_data.get('model_name', '')
|
|
provider_id = model_data.get('provider_id', '')
|
|
|
|
# 缓存提供商实例
|
|
cache_key = str(provider_id)
|
|
if cache_key not in self._provider_cache:
|
|
provider_instance = ProviderRegistry.create_instance(
|
|
provider_type=provider_type,
|
|
api_key=api_key,
|
|
api_base=api_base,
|
|
ollama_host=ollama_host,
|
|
)
|
|
if not provider_instance:
|
|
raise ValueError(f'不支持的提供商类型: {provider_type}')
|
|
self._provider_cache[cache_key] = provider_instance
|
|
|
|
provider_instance = self._provider_cache[cache_key]
|
|
setattr(provider_instance, 'config_name', model_data.get('provider_name') or provider_type)
|
|
setattr(provider_instance, 'config_id', str(provider_id))
|
|
setattr(provider_instance, 'config_api_base', api_base or '')
|
|
|
|
return provider_instance, model_name
|
|
|
|
@staticmethod
|
|
def _extract_upstream_message(error_text: str) -> str:
|
|
import re
|
|
|
|
for pattern in (
|
|
r"'message'\s*:\s*'([^']+)'",
|
|
r'"message"\s*:\s*"([^"]+)"',
|
|
):
|
|
match = re.search(pattern, error_text)
|
|
if match:
|
|
return match.group(1)
|
|
return error_text[:240]
|
|
|
|
def _format_provider_error(
|
|
self,
|
|
exc: Exception,
|
|
provider: Optional[BaseLLMProvider] = None,
|
|
model_name: str = '',
|
|
) -> str:
|
|
error_text = str(exc)
|
|
lower_text = error_text.lower()
|
|
provider_name = (
|
|
getattr(provider, 'config_name', '')
|
|
or getattr(provider, 'provider_name', '')
|
|
or '未知提供商'
|
|
)
|
|
model_text = model_name or '未知模型'
|
|
|
|
if (
|
|
'401' in lower_text
|
|
or 'unauthorized' in lower_text
|
|
or 'invalid api key' in lower_text
|
|
or 'incorrect api key' in lower_text
|
|
or '无效的令牌' in error_text
|
|
or '鉴权' in error_text
|
|
):
|
|
reason = '上游模型鉴权失败，API Key 无效或已过期'
|
|
elif (
|
|
'404' in lower_text
|
|
or 'model_not_found' in lower_text
|
|
or 'model not found' in lower_text
|
|
or 'does not exist' in lower_text
|
|
):
|
|
reason = '上游模型不存在或当前账号无权访问该模型'
|
|
elif 'timeout' in lower_text or 'timed out' in lower_text or '超时' in error_text:
|
|
reason = '上游模型请求超时'
|
|
elif 'rate limit' in lower_text or '429' in lower_text or 'quota' in lower_text:
|
|
reason = '上游模型限流或额度不足'
|
|
else:
|
|
reason = '上游模型调用失败'
|
|
|
|
upstream_message = self._extract_upstream_message(error_text)
|
|
return (
|
|
f'{reason}：提供商 {provider_name}，模型 {model_text}。'
|
|
f'请在 AI 平台的模型提供商配置中检查 Base URL、API Key 和模型名称。'
|
|
f'上游返回：{upstream_message}'
|
|
)
|
|
|
|
def chat_with_provider(
|
|
self,
|
|
provider: BaseLLMProvider,
|
|
model_name: str,
|
|
messages: List[Dict[str, str]],
|
|
temperature: float = 0.7,
|
|
max_tokens: int = 2048,
|
|
tools: List[Dict] = None,
|
|
tool_choice: str = 'auto',
|
|
**kwargs
|
|
) -> LLMResponse:
|
|
"""
|
|
使用指定提供商进行同步对话
|
|
|
|
Args:
|
|
provider: 提供商实例
|
|
model_name: 模型名称
|
|
messages: 消息列表 [{"role": "user", "content": "..."}]
|
|
temperature: 温度参数
|
|
max_tokens: 最大 Token
|
|
tools: 工具定义列表 [{"name": "...", "description": "...", "parameters": {...}}]
|
|
tool_choice: 工具选择策略 (auto, none, required, 或具体工具名)
|
|
**kwargs: 其他参数
|
|
|
|
Returns:
|
|
LLMResponse
|
|
"""
|
|
llm_messages = self._convert_messages(messages)
|
|
|
|
# 转换工具定义
|
|
tool_definitions = None
|
|
if tools:
|
|
tool_definitions = [
|
|
ToolDefinition(
|
|
name=t['name'],
|
|
description=t.get('description', ''),
|
|
parameters=t.get('parameters', {}),
|
|
)
|
|
for t in tools
|
|
]
|
|
|
|
config = LLMConfig(
|
|
model=model_name,
|
|
temperature=temperature,
|
|
max_tokens=max_tokens,
|
|
tools=tool_definitions,
|
|
tool_choice=tool_choice,
|
|
**kwargs
|
|
)
|
|
|
|
try:
|
|
return provider.chat(llm_messages, config)
|
|
except Exception as exc:
|
|
raise RuntimeError(self._format_provider_error(exc, provider, model_name)) from exc
|
|
|
|
def chat(
|
|
self,
|
|
model_id: Optional[str],
|
|
messages: List[Dict[str, str]],
|
|
temperature: float = 0.7,
|
|
max_tokens: int = 2048,
|
|
tools: List[Dict] = None,
|
|
tool_choice: str = 'auto',
|
|
**kwargs
|
|
) -> LLMResponse:
|
|
"""
|
|
同步对话入口。
|
|
|
|
旧节点仍会调用该方法；这里统一走 provider 解析和默认 chat 模型兜底，
|
|
避免不同节点各自处理 model_id 为空的情况。
|
|
"""
|
|
provider, model_name = self._run_async_sync(self._get_provider_async(model_id))
|
|
return self.chat_with_provider(
|
|
provider=provider,
|
|
model_name=model_name,
|
|
messages=messages,
|
|
temperature=temperature,
|
|
max_tokens=max_tokens,
|
|
tools=tools,
|
|
tool_choice=tool_choice,
|
|
**kwargs,
|
|
)
|
|
|
|
def _convert_messages(self, messages: List[Dict]) -> List[LLMMessage]:
|
|
"""转换消息格式，支持 tool 消息"""
|
|
llm_messages = []
|
|
for m in messages:
|
|
msg = LLMMessage(
|
|
role=m['role'],
|
|
content=m.get('content', ''),
|
|
name=m.get('name'),
|
|
tool_calls=m.get('tool_calls'),
|
|
tool_call_id=m.get('tool_call_id'),
|
|
)
|
|
llm_messages.append(msg)
|
|
return llm_messages
|
|
|
|
async def chat_async(
|
|
self,
|
|
model_id: str,
|
|
messages: List[Dict[str, str]],
|
|
temperature: float = 0.7,
|
|
max_tokens: int = 2048,
|
|
tools: List[Dict] = None,
|
|
tool_choice: str = 'auto',
|
|
**kwargs
|
|
) -> LLMResponse:
|
|
"""
|
|
异步对话
|
|
|
|
Args:
|
|
model_id: 模型 ID
|
|
messages: 消息列表
|
|
temperature: 温度参数
|
|
max_tokens: 最大 Token
|
|
tools: 工具定义列表
|
|
tool_choice: 工具选择策略
|
|
**kwargs: 其他参数
|
|
|
|
Returns:
|
|
LLMResponse
|
|
"""
|
|
provider, model_name = await self._get_provider_async(model_id)
|
|
|
|
llm_messages = self._convert_messages(messages)
|
|
|
|
# 转换工具定义
|
|
tool_definitions = None
|
|
if tools:
|
|
tool_definitions = [
|
|
ToolDefinition(
|
|
name=t['name'],
|
|
description=t.get('description', ''),
|
|
parameters=t.get('parameters', {}),
|
|
)
|
|
for t in tools
|
|
]
|
|
|
|
config = LLMConfig(
|
|
model=model_name,
|
|
temperature=temperature,
|
|
max_tokens=max_tokens,
|
|
tools=tool_definitions,
|
|
tool_choice=tool_choice,
|
|
**kwargs
|
|
)
|
|
|
|
try:
|
|
return await provider.chat_async(llm_messages, config)
|
|
except Exception as exc:
|
|
raise RuntimeError(self._format_provider_error(exc, provider, model_name)) from exc
|
|
|
|
async def chat_stream(
|
|
self,
|
|
model_id: str,
|
|
messages: List[Dict[str, str]],
|
|
temperature: float = 0.7,
|
|
max_tokens: int = 2048,
|
|
tools: List[Dict] = None,
|
|
tool_choice: str = 'auto',
|
|
**kwargs
|
|
) -> AsyncGenerator[LLMStreamChunk, None]:
|
|
"""
|
|
异步流式对话
|
|
|
|
Args:
|
|
model_id: 模型 ID
|
|
messages: 消息列表
|
|
temperature: 温度参数
|
|
max_tokens: 最大 Token
|
|
tools: 工具定义列表
|
|
tool_choice: 工具选择策略
|
|
**kwargs: 其他参数
|
|
|
|
Yields:
|
|
LLMStreamChunk
|
|
"""
|
|
provider, model_name = await self._get_provider_async(model_id)
|
|
|
|
llm_messages = self._convert_messages(messages)
|
|
|
|
# 转换工具定义
|
|
tool_definitions = None
|
|
if tools:
|
|
tool_definitions = [
|
|
ToolDefinition(
|
|
name=t['name'],
|
|
description=t.get('description', ''),
|
|
parameters=t.get('parameters', {}),
|
|
)
|
|
for t in tools
|
|
]
|
|
|
|
config = LLMConfig(
|
|
model=model_name,
|
|
temperature=temperature,
|
|
max_tokens=max_tokens,
|
|
tools=tool_definitions,
|
|
tool_choice=tool_choice,
|
|
**kwargs
|
|
)
|
|
|
|
try:
|
|
async for chunk in provider.chat_stream(llm_messages, config):
|
|
yield chunk
|
|
except Exception as exc:
|
|
raise RuntimeError(self._format_provider_error(exc, provider, model_name)) from exc
|
|
|
|
def chat_stream_sync(
|
|
self,
|
|
model_id: str,
|
|
messages: List[Dict[str, str]],
|
|
temperature: float = 0.7,
|
|
max_tokens: int = 2048,
|
|
tools: List[Dict] = None,
|
|
tool_choice: str = 'auto',
|
|
**kwargs
|
|
):
|
|
"""
|
|
同步流式对话（生成器）
|
|
|
|
注意：此方法会在内部运行异步代码来获取 provider，
|
|
需要确保 LLMService 初始化时传入了 db_session
|
|
|
|
Args:
|
|
model_id: 模型 ID
|
|
messages: 消息列表
|
|
temperature: 温度参数
|
|
max_tokens: 最大 Token
|
|
tools: 工具定义列表
|
|
tool_choice: 工具选择策略
|
|
**kwargs: 其他参数
|
|
|
|
Yields:
|
|
LLMStreamChunk
|
|
"""
|
|
import asyncio
|
|
import concurrent.futures
|
|
|
|
# 在同步上下文中运行异步代码获取 provider
|
|
async def get_provider():
|
|
return await self._get_provider_async(model_id)
|
|
|
|
try:
|
|
loop = asyncio.get_event_loop()
|
|
except RuntimeError:
|
|
provider, model_name = asyncio.run(get_provider())
|
|
else:
|
|
if loop.is_running():
|
|
# 如果已有事件循环在运行，使用线程池
|
|
with concurrent.futures.ThreadPoolExecutor() as executor:
|
|
future = executor.submit(asyncio.run, get_provider())
|
|
provider, model_name = future.result()
|
|
else:
|
|
provider, model_name = loop.run_until_complete(get_provider())
|
|
|
|
# 使用获取到的 provider 进行流式调用
|
|
yield from self.chat_stream_sync_with_provider(
|
|
provider=provider,
|
|
model_name=model_name,
|
|
messages=messages,
|
|
temperature=temperature,
|
|
max_tokens=max_tokens,
|
|
tools=tools,
|
|
tool_choice=tool_choice,
|
|
**kwargs
|
|
)
|
|
|
|
def chat_stream_sync_with_provider(
|
|
self,
|
|
provider: BaseLLMProvider,
|
|
model_name: str,
|
|
messages: List[Dict[str, str]],
|
|
temperature: float = 0.7,
|
|
max_tokens: int = 2048,
|
|
tools: List[Dict] = None,
|
|
tool_choice: str = 'auto',
|
|
**kwargs
|
|
):
|
|
"""
|
|
使用指定提供商进行同步流式对话（生成器）
|
|
|
|
Args:
|
|
provider: 提供商实例
|
|
model_name: 模型名称
|
|
messages: 消息列表
|
|
temperature: 温度参数
|
|
max_tokens: 最大 Token
|
|
tools: 工具定义列表
|
|
tool_choice: 工具选择策略
|
|
**kwargs: 其他参数
|
|
|
|
Yields:
|
|
LLMStreamChunk
|
|
"""
|
|
llm_messages = self._convert_messages(messages)
|
|
|
|
# 转换工具定义
|
|
tool_definitions = None
|
|
if tools:
|
|
tool_definitions = [
|
|
ToolDefinition(
|
|
name=t['name'],
|
|
description=t.get('description', ''),
|
|
parameters=t.get('parameters', {}),
|
|
)
|
|
for t in tools
|
|
]
|
|
|
|
config = LLMConfig(
|
|
model=model_name,
|
|
temperature=temperature,
|
|
max_tokens=max_tokens,
|
|
tools=tool_definitions,
|
|
tool_choice=tool_choice,
|
|
**kwargs
|
|
)
|
|
|
|
try:
|
|
for chunk in provider.chat_stream_sync(llm_messages, config):
|
|
yield chunk
|
|
except Exception as exc:
|
|
raise RuntimeError(self._format_provider_error(exc, provider, model_name)) from exc
|
|
|
|
@staticmethod
|
|
def get_available_providers() -> List[Dict]:
|
|
"""获取可用的提供商列表"""
|
|
return ProviderRegistry.get_all_types()
|
|
|
|
@staticmethod
|
|
def get_default_models(provider_type: str) -> List[Dict]:
|
|
"""获取提供商的默认模型列表"""
|
|
return ProviderRegistry.get_default_models(provider_type)
|