from fastapi import FastAPI, Request, Security, HTTPException
from fastapi.responses import JSONResponse, StreamingResponse
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.middleware.gzip import GZipMiddleware
from collections import OrderedDict
from dataclasses import dataclass, field
from pydantic_settings import BaseSettings
import httpx, json, time, logging, hashlib, asyncio, uuid, pickle, pathlib
from datetime import datetime, timedelta
_start_time = time.time() # 记录启动时间
# ── 配置外部化 ─────────────────────────────────────────────
class Settings(BaseSettings):
target: str = "https://llm.test.com/v1"
api_key: str = "xxx"
proxy_api_key: str = "proxy-secret-key"
cache_ttl: int = 10
max_cache_size: int = 100
max_request_size: int = 10 * 1024 * 1024 # 10MB
max_retries: int = 2
retry_delay: float = 1.0
tool_desc_max_len: int = 100
connect_timeout: float = 5.0
read_timeout: float = 60.0
write_timeout: float = 10.0
log_file: str = "proxy.log"
# 模型映射：Claude Desktop 模型名 → 实际私有模型名
model_sonnet: str = "pri-kimi-26"
model_haiku: str = "pri-glm-51"
model_opus: str = "pri-deepseek-v4"
# 工具过滤：白名单优先，两者都空则不过滤
# 格式：逗号分隔的工具名，如 read_file,write_file
tool_whitelist: str = "" # 只保留列出的工具
tool_blacklist: str = "" # 过滤掉列出的工具
class Config:
env_file = ".env"
settings = Settings()
MODEL_MAP = {
"claude-sonnet-4-5": settings.model_sonnet,
"claude-haiku-4-5": settings.model_haiku,
"claude-opus-4-5": settings.model_opus,
}
# ── 结构化日志（终端 + 文件）─────────────────────────────
class JSONFormatter(logging.Formatter):
def format(self, record):
return json.dumps({
"time": datetime.now().isoformat(),
"level": record.levelname,
"message": record.getMessage(),
}, ensure_ascii=False)
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
_console = logging.StreamHandler()
_console.setFormatter(JSONFormatter())
logger.addHandler(_console)
_file = logging.FileHandler(settings.log_file, encoding="utf-8")
_file.setFormatter(JSONFormatter())
logger.addHandler(_file)
app = FastAPI()
# ── 响应压缩 ───────────────────────────────────────────────
app.add_middleware(GZipMiddleware, minimum_size=1000)
# ── API Key 鉴权 ───────────────────────────────────────────
security = HTTPBearer(auto_error=False)
def verify_key(credentials: HTTPAuthorizationCredentials = Security(security)):
if not credentials or credentials.credentials != settings.proxy_api_key:
raise HTTPException(status_code=401, detail="Invalid API Key")
return credentials
# ── LRU 缓存 ──────────────────────────────────────────────
CACHE_FILE = pathlib.Path("cache.pkl")
STATS_FILE = pathlib.Path("stats.json")
class LRUCache:
def __init__(self, max_size: int, ttl: int):
self.cache = OrderedDict()
self.max_size = max_size
self.ttl = ttl
def get(self, key: str):
if key not in self.cache:
return None
cached_time, value = self.cache[key]
if datetime.now() - cached_time > timedelta(seconds=self.ttl):
del self.cache[key]
return None
self.cache.move_to_end(key)
return value
def set(self, key: str, value):
if key in self.cache:
self.cache.move_to_end(key)
self.cache[key] = (datetime.now(), value)
if len(self.cache) > self.max_size:
self.cache.popitem(last=False)
cache = LRUCache(max_size=settings.max_cache_size, ttl=settings.cache_ttl)
# ── 并发去重 ───────────────────────────────────────────────
pending_requests: dict[str, asyncio.Future] = {}
# ── 统计（含并发锁）──────────────────────────────────────
@dataclass
class Stats:
total_requests: int = 0
cache_hits: int = 0
gateway_errors: int = 0
retries: int = 0
total_input_tokens: int = 0
total_output_tokens: int = 0
@property
def hit_rate(self) -> str:
if self.total_requests == 0:
return "0%"
return f"{self.cache_hits / self.total_requests * 100:.1f}%"
@property
def total_tokens(self) -> int:
return self.total_input_tokens + self.total_output_tokens
stats = Stats()
stats_lock = asyncio.Lock()
# 最近一次请求携带的工具列表（用于 stats 页面展示）
_last_tools: list[str] = []
_tools_lock = asyncio.Lock()
# 运行时工具过滤状态（从 .env 初始化，可通过页面动态修改）
_tool_filter_lock = asyncio.Lock()
_tool_whitelist: set[str] = set(t.strip() for t in settings.tool_whitelist.split(",") if t.strip())
_tool_blacklist: set[str] = set(t.strip() for t in settings.tool_blacklist.split(",") if t.strip())
ENV_FILE_PATH = pathlib.Path(".env")
def _save_tool_filter_to_env():
"""将当前工具过滤状态写回 .env 文件"""
try:
if ENV_FILE_PATH.exists():
lines = ENV_FILE_PATH.read_text(encoding="utf-8").splitlines()
new_lines = []
whitelist_written = blacklist_written = False
for line in lines:
if line.startswith("TOOL_WHITELIST="):
new_lines.append("TOOL_WHITELIST=" + ",".join(sorted(_tool_whitelist)))
whitelist_written = True
elif line.startswith("TOOL_BLACKLIST="):
new_lines.append("TOOL_BLACKLIST=" + ",".join(sorted(_tool_blacklist)))
blacklist_written = True
else:
new_lines.append(line)
if not whitelist_written:
new_lines.append("TOOL_WHITELIST=" + ",".join(sorted(_tool_whitelist)))
if not blacklist_written:
new_lines.append("TOOL_BLACKLIST=" + ",".join(sorted(_tool_blacklist)))
ENV_FILE_PATH.write_text("\n".join(new_lines) + "\n", encoding="utf-8")
except Exception as e:
logger.warning(json.dumps({"message": f"工具过滤配置写入失败: {str(e)}"}))
def _apply_tool_filter(tools: list) -> list:
"""根据黑白名单过滤工具列表"""
if _tool_whitelist:
return [t for t in tools if t["function"]["name"] in _tool_whitelist]
if _tool_blacklist:
return [t for t in tools if t["function"]["name"] not in _tool_blacklist]
return tools
async def inc_stat(field: str, delta: int = 1):
"""线程安全的统计自增"""
async with stats_lock:
setattr(stats, field, getattr(stats, field) + delta)
# ── 工具调用格式转换 ───────────────────────────────────────
def slim_schema(schema) -> dict:
"""精简参数 schema，去掉非必要字段节省 token"""
remove_keys = {"title", "examples", "default", "$schema"}
if isinstance(schema, dict):
return {k: slim_schema(v) for k, v in schema.items() if k not in remove_keys}
if isinstance(schema, list):
return [slim_schema(i) for i in schema]
return schema
def convert_tools_to_openai(tools: list, max_desc_len: int = 100) -> list:
"""Anthropic 工具定义 → OpenAI 格式（含描述压缩和 schema 精简）"""
result = []
for tool in tools:
desc = tool.get("description", "")
if len(desc) > max_desc_len:
desc = desc[:max_desc_len] + "..."
result.append({
"type": "function",
"function": {
"name": tool.get("name", ""),
"description": desc,
"parameters": slim_schema(tool.get("input_schema", {}))
}
})
return result
def convert_messages_to_openai(messages: list) -> list:
"""Anthropic messages → OpenAI 格式（含多轮工具调用）"""
result = []
for msg in messages:
role = msg.get("role", "user")
content = msg.get("content", "")
if isinstance(content, str):
result.append({"role": role, "content": content})
continue
if isinstance(content, list):
tool_calls, tool_results, text_parts = [], [], []
for block in content:
btype = block.get("type", "")
if btype == "tool_use":
tool_calls.append({
"id": block.get("id", ""),
"type": "function",
"function": {
"name": block.get("name", ""),
"arguments": json.dumps(block.get("input", {}), ensure_ascii=False)
}
})
elif btype == "tool_result":
tool_content = block.get("content", "")
if isinstance(tool_content, list):
tool_content = " ".join(
b.get("text", "") for b in tool_content if b.get("type") == "text"
)
tool_results.append({
"role": "tool",
"tool_call_id": block.get("tool_use_id", ""),
"content": tool_content
})
elif btype == "text":
text_parts.append(block.get("text", ""))
if tool_calls:
msg_out = {"role": "assistant", "tool_calls": tool_calls}
if text_parts:
msg_out["content"] = " ".join(text_parts)
result.append(msg_out)
if tool_results:
result.extend(tool_results)
if text_parts and not tool_calls and not tool_results:
result.append({"role": role, "content": " ".join(text_parts)})
return result
def has_pending_tool_calls(messages: list) -> bool:
"""检测是否有未完成的多轮工具调用"""
for msg in reversed(messages):
if msg.get("role") == "assistant":
content = msg.get("content", [])
if isinstance(content, list):
return any(b.get("type") == "tool_use" for b in content)
return False
def normalize_usage(usage: dict) -> dict:
"""标准化 usage 字段，兼容不同模型返回格式"""
return {
"input_tokens": usage.get("prompt_tokens") or usage.get("input_tokens") or 0,
"output_tokens": usage.get("completion_tokens") or usage.get("output_tokens") or 0,
}
# 需要过滤的异常 token 列表
_ABNORMAL_TOKENS = [
"<|begin_of_sentence|>", "<|end_of_sentence|>",
"<|im_start|>", "<|im_end|>",
"<|endoftext|>", "", "",
]
def clean_content(text: str) -> str:
"""过滤模型输出中的异常特殊 token"""
if not text:
return ""
for token in _ABNORMAL_TOKENS:
text = text.replace(token, "")
return text.strip()
def convert_response_to_anthropic(data: dict, requested_model: str) -> dict:
"""OpenAI 响应 → Anthropic 格式（含工具调用）"""
choice = data["choices"][0]
message = choice.get("message", {})
finish_reason = choice.get("finish_reason", "stop")
usage = normalize_usage(data.get("usage", {}))
if finish_reason == "tool_calls" and message.get("tool_calls"):
content_blocks = []
if message.get("content"):
content_blocks.append({"type": "text", "text": clean_content(message["content"])})
for tool_call in message["tool_calls"]:
fn = tool_call.get("function", {})
try:
input_data = json.loads(fn.get("arguments", "{}"))
except Exception:
input_data = {}
content_blocks.append({
"type": "tool_use",
"id": tool_call.get("id", ""),
"name": fn.get("name", ""),
"input": input_data
})
return {
"id": data.get("id", "msg_001"),
"type": "message", "role": "assistant",
"content": content_blocks,
"model": requested_model,
"stop_reason": "tool_use",
"usage": usage
}
return {
"id": data.get("id", "msg_001"),
"type": "message", "role": "assistant",
"content": [{"type": "text", "text": clean_content(message.get("content", ""))}],
"model": requested_model,
"stop_reason": "end_turn",
"usage": usage
}
# ── 重试请求 ───────────────────────────────────────────────
async def request_with_retry(client, url, payload, headers, request_id, actual_model):
last_error = None
for attempt in range(settings.max_retries + 1):
try:
if attempt > 0:
await inc_stat("retries")
logger.warning(json.dumps({
"request_id": request_id,
"message": f"第 {attempt} 次重试",
"model": actual_model
}))
await asyncio.sleep(settings.retry_delay * attempt)
resp = await client.post(url, json=payload, headers=headers)
resp.raise_for_status()
return resp.json()
except (httpx.ReadTimeout, httpx.ConnectTimeout) as e:
last_error = e
continue
except httpx.HTTPStatusError as e:
if e.response.status_code >= 500:
last_error = e
continue
raise
raise last_error
def get_cache_key(model: str, messages: list, tools: list) -> str:
content = json.dumps({"messages": messages, "tools": tools}, ensure_ascii=False, sort_keys=True)
return hashlib.md5(f"{model}:{content}".encode()).hexdigest()
# ── 启动：恢复缓存和统计 ──────────────────────────────────
@app.on_event("startup")
async def startup_check():
# 恢复缓存
if CACHE_FILE.exists():
try:
with open(CACHE_FILE, "rb") as f:
cache.cache = pickle.load(f)
logger.info(json.dumps({"message": f"已恢复缓存 {len(cache.cache)} 条"}))
except Exception as e:
logger.warning(json.dumps({"message": f"缓存恢复失败: {str(e)}"}))
# 恢复统计
if STATS_FILE.exists():
try:
data = json.loads(STATS_FILE.read_text())
stats.total_requests = data.get("total_requests", 0)
stats.cache_hits = data.get("cache_hits", 0)
stats.gateway_errors = data.get("gateway_errors", 0)
stats.retries = data.get("retries", 0)
stats.total_input_tokens = data.get("total_input_tokens", 0)
stats.total_output_tokens = data.get("total_output_tokens", 0)
logger.info(json.dumps({"message": "已恢复历史统计数据"}))
except Exception as e:
logger.warning(json.dumps({"message": f"统计恢复失败: {str(e)}"}))
# 打印当前配置摘要
logger.info(json.dumps({
"message": "代理启动",
"target": settings.target,
"models": MODEL_MAP,
"cache_ttl": settings.cache_ttl,
"tool_desc_max_len": settings.tool_desc_max_len,
"read_timeout": settings.read_timeout,
}))
# 网关连通性检查
try:
async with httpx.AsyncClient(timeout=5.0) as client:
resp = await client.get(
f"{settings.target}/models",
headers={"Authorization": f"Bearer {settings.api_key}"}
)
resp.raise_for_status()
logger.info(json.dumps({"message": f"网关连通性检查通过: {settings.target}"}))
except Exception as e:
logger.warning(json.dumps({"message": f"网关连通性检查失败: {str(e)}"}))
# ── 关闭：持久化缓存和统计 ───────────────────────────────
@app.on_event("shutdown")
async def shutdown():
logger.info(json.dumps({"message": "代理正在关闭，等待进行中的请求完成..."}))
if pending_requests:
await asyncio.gather(*pending_requests.values(), return_exceptions=True)
# 持久化缓存
try:
with open(CACHE_FILE, "wb") as f:
pickle.dump(cache.cache, f)
logger.info(json.dumps({"message": f"缓存已持久化 {len(cache.cache)} 条"}))
except Exception as e:
logger.warning(json.dumps({"message": f"缓存持久化失败: {str(e)}"}))
# 持久化统计
try:
STATS_FILE.write_text(json.dumps({
"total_requests": stats.total_requests,
"cache_hits": stats.cache_hits,
"gateway_errors": stats.gateway_errors,
"retries": stats.retries,
"total_input_tokens": stats.total_input_tokens,
"total_output_tokens": stats.total_output_tokens,
}))
logger.info(json.dumps({"message": "统计数据已持久化"}))
except Exception as e:
logger.warning(json.dumps({"message": f"统计持久化失败: {str(e)}"}))
logger.info(json.dumps({"message": "代理已安全关闭"}))
# ── 请求日志中间件 ─────────────────────────────────────────
@app.middleware("http")
async def log_requests(request: Request, call_next):
start = time.time()
response = await call_next(request)
duration = time.time() - start
logger.info(json.dumps({
"method": request.method,
"path": request.url.path,
"status": response.status_code,
"duration": f"{duration:.2f}s"
}))
return response
# ── 健康检查 ───────────────────────────────────────────────
@app.get("/")
@app.head("/")
async def health():
return JSONResponse({"status": "ok"})
# ── 统计接口（HTML 页面）──────────────────────────────────
@app.get("/stats")
async def get_stats():
from fastapi.responses import HTMLResponse
uptime = int(time.time() - _start_time)
h, m, s = uptime // 3600, (uptime % 3600) // 60, uptime % 60
model_rows = "".join(
f"
{k}
{v}
" for k, v in MODEL_MAP.items()
)
# 获取工具列表和当前过滤状态
async with _tools_lock:
last_tools_snapshot = list(_last_tools)
async with _tool_filter_lock:
cur_whitelist = set(_tool_whitelist)
cur_blacklist = set(_tool_blacklist)
if cur_whitelist:
cur_mode = "whitelist"
cur_filtered = cur_whitelist
elif cur_blacklist:
cur_mode = "blacklist"
cur_filtered = cur_blacklist
else:
cur_mode = "none"
cur_filtered = set()
# 构建工具勾选列表
if last_tools_snapshot:
tool_items = ""
for name in last_tools_snapshot:
if cur_mode == "whitelist":
checked = "checked" if name in cur_filtered else ""
status_color = "#080" if name in cur_filtered else "#aaa"
status_text = "发送" if name in cur_filtered else "过滤"
elif cur_mode == "blacklist":
checked = "checked" if name not in cur_filtered else ""
status_color = "#080" if name not in cur_filtered else "#c00"
status_text = "发送" if name not in cur_filtered else "过滤"
else:
checked = "checked"
status_color = "#080"
status_text = "发送"
tool_items += (
f"