""" 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 result = await self._db.execute( select(LLMModel) .where( LLMModel.is_deleted == False, LLMModel.is_active == True, LLMModel.model_type == "chat", ) .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) 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 _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)