Pamudu13/gemma-3-chat
0
1import httpx2from typing import Optional, List, Dict, Any, AsyncIterator, Union3import functools4 5class AsyncClaudeAsOpenAI:6 """7 完全模拟 AsyncOpenAI 客户端，底层用 litellm.acompletion（懒加载）8 """9 10 def __init__(11 self, 12 api_key: str, 13 base_url: Optional[str] = None,14 default_model: Optional[str] = "claude-3-5-sonnet-20241022",15 http_client: Optional[httpx.AsyncClient] = None,16 timeout: Optional[float] = None,17 max_retries: Optional[int] = None,18 **kwargs19 ):20 self.api_key = api_key21 self.base_url = base_url 22 self.default_model = default_model23 self.http_client = http_client24 self.timeout = timeout25 self.max_retries = max_retries26 self._extra_kwargs = kwargs27 self._litellm_module = None # 缓存 litellm 模块28 29 @property30 def _litellm(self):31 """懒加载 litellm，第一次调用时才 import"""32 if self._litellm_module is None:33 import litellm34 self._litellm_module = litellm35 # 可选：配置 litellm（如关闭日志）36 # litellm.set_verbose = False37 # litellm.suppress_debug_info = True38 return self._litellm_module39 40 @property41 def models(self):42 return self._ModelsResource(self)43 44 class _ModelsResource:45 def __init__(self, parent: "AsyncClaudeAsOpenAI"):46 self._parent = parent47 48 async def list(self):49 # 构造兼容 OpenAI 返回形式的对象50 class ModelItem:51 def __init__(self, model_id: str):52 self.id = model_id53 54 class ModelList:55 def __init__(self, data: list):56 self.data = data57 58 # 处理请求 URL，Anthropic 的获取模型接口通常是 /v1/models59 base_url = self._parent.base_url or "https://api.anthropic.com"60 if base_url.endswith("/v1") or base_url.endswith("/v1/"):61 url = f"{base_url.rstrip('/')}/models"62 else:63 url = f"{base_url.rstrip('/')}/v1/models"64 65 headers = {66 "x-api-key": self._parent.api_key,67 "anthropic-version": "2023-06-01"68 }69 70 try:71 # 优先复用全局 http_client 以走系统代理配置72 client = self._parent.http_client73 need_close = False74 if not client:75 client = httpx.AsyncClient()76 need_close = True77 78 response = await client.get(url, headers=headers)79 80 if need_close:81 await client.aclose()82 83 # 如果 API 成功响应84 if response.status_code == 200:85 data = response.json()86 # 解析官方格式: {"type": "list", "data":[{"id": "claude-3-opus-...", ...}]}87 models = [ModelItem(m["id"]) for m in data.get("data",[])]88 if models:89 return ModelList(models)90 except Exception as e:91 print(f"动态获取 Anthropic 模型列表失败 (可能代理/代理商不支持): {e}")92 93 # [静态兜底方案]：如果请求报错或代理商 API 未实现 /models 端点，返回常见的 Claude 模型94 fallback_models =[]95 return ModelList([ModelItem(m) for m in fallback_models])96 97 def _convert_tools(self, tools: Optional[List[Dict]]) -> Optional[List[Dict]]:98 """OpenAI Tools -> Claude Tools"""99 if not tools:100 return None101 102 claude_tools = []103 for tool in tools:104 tool_type = tool.get("type")105 106 if tool_type == "custom":107 continue # Claude 不支持108 elif tool_type == "function":109 func = tool.get("function", {})110 claude_tools.append({111 "name": func.get("name"),112 "description": func.get("description", ""),113 "input_schema": func.get("parameters", {"type": "object", "properties": {}})114 })115 elif tool_type in ["web_search_20250305", "web_search_20260209"]:116 claude_tools.append(tool)117 118 return claude_tools if claude_tools else None119 120 def _convert_tool_choice(self, tool_choice: Any) -> Any:121 """OpenAI tool_choice -> Claude tool_choice"""122 if tool_choice is None:123 return None124 125 if isinstance(tool_choice, str):126 return tool_choice127 128 if isinstance(tool_choice, dict) and tool_choice.get("type") == "function":129 func_name = tool_choice.get("function", {}).get("name")130 if func_name:131 return {"type": "tool", "name": func_name}132 133 return tool_choice134 135 @property136 def chat(self):137 return self._ChatResource(self)138 139 class _ChatResource:140 def __init__(self, parent: "AsyncClaudeAsOpenAI"):141 self.completions = self._CompletionsResource(parent)142 143 class _CompletionsResource:144 def __init__(self, parent: "AsyncClaudeAsOpenAI"):145 self._parent = parent146 147 async def create(148 self,149 model: Optional[str] = None,150 messages: Optional[List[Dict[str, Any]]] = None,151 temperature: Optional[float] = None,152 max_tokens: Optional[int] = None,153 stream: bool = False,154 top_p: Optional[float] = None,155 stop: Optional[Union[str, List[str]]] = None,156 tools: Optional[List[Dict]] = None,157 tool_choice: Optional[Any] = None,158 **kwargs159 ):160 model = model or self._parent.default_model161 if not model:162 raise ValueError("model is required")163 164 if not model.startswith("anthropic/"):165 model = f"anthropic/{model}"166 167 # ===== 懒加载 litellm =====168 litellm = self._parent._litellm169 170 completion_kwargs = {171 "model": model,172 "messages": messages,173 "api_key": self._parent.api_key,174 "stream": stream,175 }176 177 # Tools 转换178 if tools:179 converted_tools = self._parent._convert_tools(tools)180 if converted_tools:181 completion_kwargs["tools"] = converted_tools182 183 if tool_choice:184 completion_kwargs["tool_choice"] = self._parent._convert_tool_choice(tool_choice)185 186 # 其他参数187 if self._parent.base_url:188 completion_kwargs["api_base"] = self._parent.base_url189 if temperature is not None:190 completion_kwargs["temperature"] = temperature191 if max_tokens is not None:192 completion_kwargs["max_tokens"] = max_tokens193 if top_p is not None:194 completion_kwargs["top_p"] = top_p195 if stop is not None:196 completion_kwargs["stop"] = stop197 if self._parent.timeout is not None:198 completion_kwargs["timeout"] = self._parent.timeout199 if self._parent.http_client is not None:200 completion_kwargs["client"] = self._parent.http_client201 202 # 过滤 OpenAI 特有参数203 safe_kwargs = {k: v for k, v in kwargs.items() 204 if k not in ['logprobs', 'top_logprobs', 'response_format', 'n']}205 206 return await litellm.acompletion(**completion_kwargs, **safe_kwargs)