241 lines
8.5 KiB
Python
241 lines
8.5 KiB
Python
# llm_adapters.py
|
||
# -*- coding: utf-8 -*-
|
||
import logging
|
||
from typing import Optional
|
||
from langchain_openai import ChatOpenAI, AzureChatOpenAI
|
||
from google import genai
|
||
from google.genai import types
|
||
|
||
def ensure_openai_base_url_has_v1(url: str) -> str:
|
||
"""
|
||
处理base_url的规则：
|
||
1. 如果url以#结尾，则移除#并直接使用用户提供的url
|
||
2. 否则检查是否需要添加/v1后缀
|
||
"""
|
||
import re
|
||
url = url.strip()
|
||
if not url:
|
||
return url
|
||
|
||
if url.endswith('#'):
|
||
return url.rstrip('#')
|
||
|
||
if not re.search(r'/v\d+$', url):
|
||
if '/v1' not in url:
|
||
url = url.rstrip('/') + '/v1'
|
||
return url
|
||
|
||
class BaseLLMAdapter:
|
||
"""
|
||
统一的 LLM 接口基类，为不同后端（OpenAI、Ollama、ML Studio、Gemini等）提供一致的方法签名。
|
||
"""
|
||
def invoke(self, prompt: str) -> str:
|
||
raise NotImplementedError("Subclasses must implement .invoke(prompt) method.")
|
||
|
||
class DeepSeekAdapter(BaseLLMAdapter):
|
||
"""
|
||
适配官方/OpenAI兼容接口（使用 langchain.ChatOpenAI）
|
||
"""
|
||
def __init__(self, api_key: str, base_url: str, model_name: str, max_tokens: int, temperature: float = 0.7, timeout: Optional[int] = 600):
|
||
self.base_url = ensure_openai_base_url_has_v1(base_url)
|
||
self.api_key = api_key
|
||
self.model_name = model_name
|
||
self.max_tokens = max_tokens
|
||
self.temperature = temperature
|
||
self.timeout = timeout
|
||
|
||
self._client = ChatOpenAI(
|
||
model=self.model_name,
|
||
api_key=self.api_key,
|
||
base_url=self.base_url,
|
||
max_tokens=self.max_tokens,
|
||
temperature=self.temperature,
|
||
timeout=self.timeout
|
||
)
|
||
|
||
def invoke(self, prompt: str) -> str:
|
||
response = self._client.invoke(prompt)
|
||
if not response:
|
||
logging.warning("No response from DeepSeekAdapter.")
|
||
return ""
|
||
return response.content
|
||
|
||
class OpenAIAdapter(BaseLLMAdapter):
|
||
"""
|
||
适配官方/OpenAI兼容接口（使用 langchain.ChatOpenAI）
|
||
"""
|
||
def __init__(self, api_key: str, base_url: str, model_name: str, max_tokens: int, temperature: float = 0.7, timeout: Optional[int] = 600):
|
||
self.base_url = ensure_openai_base_url_has_v1(base_url)
|
||
self.api_key = api_key
|
||
self.model_name = model_name
|
||
self.max_tokens = max_tokens
|
||
self.temperature = temperature
|
||
self.timeout = timeout
|
||
|
||
self._client = ChatOpenAI(
|
||
model=self.model_name,
|
||
api_key=self.api_key,
|
||
base_url=self.base_url,
|
||
max_tokens=self.max_tokens,
|
||
temperature=self.temperature,
|
||
timeout=self.timeout
|
||
)
|
||
|
||
def invoke(self, prompt: str) -> str:
|
||
response = self._client.invoke(prompt)
|
||
if not response:
|
||
logging.warning("No response from OpenAIAdapter.")
|
||
return ""
|
||
return response.content
|
||
|
||
class GeminiAdapter(BaseLLMAdapter):
|
||
"""
|
||
适配 Google Gemini (Google Generative AI) 接口
|
||
"""
|
||
def __init__(self, api_key: str, model_name: str, max_tokens: int, temperature: float = 0.7, timeout: Optional[int] = 600):
|
||
self.api_key = api_key
|
||
self.model_name = model_name
|
||
self.max_tokens = max_tokens
|
||
self.temperature = temperature
|
||
self.timeout = timeout
|
||
|
||
self._client = genai.Client(api_key=self.api_key)
|
||
|
||
def invoke(self, prompt: str) -> str:
|
||
try:
|
||
response = self._client.models.generate_content(
|
||
model = self.model_name,
|
||
contents = prompt,
|
||
config = types.GenerateContentConfig(
|
||
max_output_tokens=self.max_tokens,
|
||
temperature=self.temperature,
|
||
)
|
||
)
|
||
if response and response.text:
|
||
return response.text
|
||
else:
|
||
logging.warning("No text response from Gemini API.")
|
||
return ""
|
||
except Exception as e:
|
||
logging.error(f"Gemini API 调用失败: {e}")
|
||
return ""
|
||
|
||
class AzureOpenAIAdapter(BaseLLMAdapter):
|
||
"""
|
||
适配 Azure OpenAI 接口（使用 langchain.ChatOpenAI）
|
||
"""
|
||
def __init__(self, api_key: str, base_url: str, model_name: str, max_tokens: int, temperature: float = 0.7, timeout: Optional[int] = 600):
|
||
import re
|
||
match = re.match(r'https://(.+?)/openai/deployments/(.+?)/chat/completions\?api-version=(.+)', base_url)
|
||
if match:
|
||
self.azure_endpoint = f"https://{match.group(1)}"
|
||
self.azure_deployment = match.group(2)
|
||
self.api_version = match.group(3)
|
||
else:
|
||
raise ValueError("Invalid Azure OpenAI base_url format")
|
||
|
||
self.api_key = api_key
|
||
self.model_name = self.azure_deployment
|
||
self.max_tokens = max_tokens
|
||
self.temperature = temperature
|
||
self.timeout = timeout
|
||
|
||
self._client = AzureChatOpenAI(
|
||
azure_endpoint=self.azure_endpoint,
|
||
azure_deployment=self.azure_deployment,
|
||
api_version=self.api_version,
|
||
api_key=self.api_key,
|
||
max_tokens=self.max_tokens,
|
||
temperature=self.temperature,
|
||
timeout=self.timeout
|
||
)
|
||
|
||
def invoke(self, prompt: str) -> str:
|
||
response = self._client.invoke(prompt)
|
||
if not response:
|
||
logging.warning("No response from AzureOpenAIAdapter.")
|
||
return ""
|
||
return response.content
|
||
|
||
class OllamaAdapter(BaseLLMAdapter):
|
||
"""
|
||
Ollama 同样有一个 OpenAI-like /v1/chat 接口，可直接使用 ChatOpenAI。
|
||
"""
|
||
def __init__(self, api_key: str, base_url: str, model_name: str, max_tokens: int, temperature: float = 0.7, timeout: Optional[int] = 600):
|
||
self.base_url = ensure_openai_base_url_has_v1(base_url)
|
||
self.api_key = api_key
|
||
self.model_name = model_name
|
||
self.max_tokens = max_tokens
|
||
self.temperature = temperature
|
||
self.timeout = timeout
|
||
|
||
self._client = ChatOpenAI(
|
||
model=self.model_name,
|
||
api_key=self.api_key,
|
||
base_url=self.base_url,
|
||
max_tokens=self.max_tokens,
|
||
temperature=self.temperature,
|
||
timeout=self.timeout
|
||
)
|
||
|
||
def invoke(self, prompt: str) -> str:
|
||
response = self._client.invoke(prompt)
|
||
if not response:
|
||
logging.warning("No response from OllamaAdapter.")
|
||
return ""
|
||
return response.content
|
||
|
||
class MLStudioAdapter(BaseLLMAdapter):
|
||
def __init__(self, api_key: str, base_url: str, model_name: str, max_tokens: int, temperature: float = 0.7, timeout: Optional[int] = 600):
|
||
self.base_url = ensure_openai_base_url_has_v1(base_url)
|
||
self.api_key = api_key
|
||
self.model_name = model_name
|
||
self.max_tokens = max_tokens
|
||
self.temperature = temperature
|
||
self.timeout = timeout
|
||
|
||
self._client = ChatOpenAI(
|
||
model=self.model_name,
|
||
api_key=self.api_key,
|
||
base_url=self.base_url,
|
||
max_tokens=self.max_tokens,
|
||
temperature=self.temperature,
|
||
timeout=self.timeout
|
||
)
|
||
|
||
def invoke(self, prompt: str) -> str:
|
||
response = self._client.invoke(prompt)
|
||
if not response:
|
||
logging.warning("No response from MLStudioAdapter.")
|
||
return ""
|
||
return response.content
|
||
|
||
def create_llm_adapter(
|
||
interface_format: str,
|
||
base_url: str,
|
||
model_name: str,
|
||
api_key: str,
|
||
temperature: float,
|
||
max_tokens: int,
|
||
timeout: int
|
||
) -> BaseLLMAdapter:
|
||
"""
|
||
工厂函数：根据 interface_format 返回不同的适配器实例。
|
||
"""
|
||
fmt = interface_format.strip().lower()
|
||
if fmt == "deepseek":
|
||
return DeepSeekAdapter(api_key, base_url, model_name, max_tokens, temperature, timeout)
|
||
elif fmt == "openai":
|
||
return OpenAIAdapter(api_key, base_url, model_name, max_tokens, temperature, timeout)
|
||
elif fmt == "azure openai":
|
||
return AzureOpenAIAdapter(api_key, base_url, model_name, max_tokens, temperature, timeout)
|
||
elif fmt == "ollama":
|
||
return OllamaAdapter(api_key, base_url, model_name, max_tokens, temperature, timeout)
|
||
elif fmt == "ml studio":
|
||
return MLStudioAdapter(api_key, base_url, model_name, max_tokens, temperature, timeout)
|
||
elif fmt == "gemini":
|
||
# base_url 对 Gemini 暂无用处，可忽略
|
||
return GeminiAdapter(api_key, model_name, max_tokens, temperature, timeout)
|
||
else:
|
||
raise ValueError(f"Unknown interface_format: {interface_format}")
|