
LLM Gateway MCP Server

2025.04.09
0
PythonAI任务委派成本优化多模型集成开发效率
LLM Gateway MCP Server 是一个基于模型上下文协议(MCP)的服务器,旨在实现从高能力AI代理(如Claude 3.7 Sonnet)到成本效益更高的LLM(如Gemini Flash 2.0 Lite)的智能任务委派。它提供了一个统一的接口,支持多个大型语言模型(LLM)提供商,同时优化成本、性能和质量。
View on GitHub
Overview
产品定位
LLM Gateway MCP Server 是一个专为AI代理设计的任务委派服务器,通过智能路由和优化,实现高效、经济的AI任务处理。
核心功能
- MCP协议集成:
- 原生MCP服务器,支持AI代理集成
- 所有功能通过标准化的MCP工具暴露
-
支持工具组合和发现
-
智能任务委派:
- 任务路由和提供商选择
- 成本-性能平衡
-
委派跟踪
-
高级缓存:
- 多级缓存策略(精确匹配、语义相似、任务感知)
- 持久化缓存
-
缓存分析
-
文档工具:
- 智能分块(基于令牌、语义边界、结构分析)
-
文档操作(摘要、实体提取、问题生成、批量处理)
-
结构化数据提取:
- JSON提取(带模式验证)
- 表格提取
- 键值对提取
-
语义模式推断
-
锦标赛模式:
- 代码和文本竞赛
- 多模型比较
- 性能指标评估
-
结果存储
-
高级向量操作:
- 语义搜索
- 向量存储
- 混合搜索(关键字+语义)
- 批量处理
适用场景
- AI代理编排:高级AI代理(如Claude 3.7)委派常规任务给成本更低的模型
- 企业文档处理:大规模文档的高效处理(分块、摘要、实体提取)
- 研究与分析:研究论文处理、结构化信息提取、模型比较
- 模型基准测试与选择:运行模型竞赛,生成性能指标
工具列表
- 文档分块工具:将大文档分割为语义上有意义的小块
- 文档摘要工具:生成文档摘要,支持多种格式
- 实体提取工具:从文本中提取特定类型的实体
- JSON提取工具:从文本中提取结构化JSON数据
- 多完成工具:从多个提供商获取相同提示的完成结果
- 优化工作流工具:执行多步骤工作流,优化模型选择
- 锦标赛工具:运行模型竞赛,比较输出
- RAG查询工具:基于检索的生成,提高事实准确性
- 本地文本处理工具:离线文本操作(清理、格式化)
- 工具列表工具:动态发现可用工具
常见问题解答
- 如何配置API密钥?
-
在
.env
文件中设置各提供商的API密钥 -
如何启用缓存?
-
设置
CACHE_ENABLED=true
,并可配置缓存类型和TTL -
如何优化成本?
- 使用智能路由将适当任务委派给更便宜的模型
-
启用高级缓存避免冗余API调用
-
如何扩展新的LLM提供商?
- 服务器采用可扩展架构,可添加新的提供商集成
使用教程
使用依赖
# 安装uv(如果尚未安装)
curl -LsSf https://astral.sh/uv/install.sh | sh
安装教程
# 克隆仓库
git clone https://github.com/yourusername/llm_gateway_mcp_server.git
cd llm_gateway_mcp_server
# 使用uv创建venv并安装
uv venv --python 3.13
source .venv/bin/activate
uv pip install -e ".[all]"
环境设置
创建.env
文件并配置API密钥:
# API密钥(至少需要一个提供商)
OPENAI_API_KEY=your_openai_key
ANTHROPIC_API_KEY=your_anthropic_key
GEMINI_API_KEY=your_gemini_key
DEEPSEEK_API_KEY=your_deepseek_key
# 服务器配置
SERVER_PORT=8013
SERVER_HOST=127.0.0.1
# 日志配置
LOG_LEVEL=INFO
USE_RICH_LOGGING=true
# 缓存配置
CACHE_ENABLED=true
CACHE_TTL=86400
运行服务器
# 启动MCP服务器
python -m llm_gateway.cli.main run
# 或使用Docker
docker compose up
调试方式
服务器运行后,可通过以下方式测试:
- 使用MCP客户端连接服务器
- 调用工具接口进行测试
- 检查日志输出(日志级别可设置为DEBUG以获取更详细信息)
示例调试命令(Python):
import asyncio
from mcp.client import Client
async def test_connection():
client = Client("http://localhost:8013")
response = await client.tools.list_tools()
print(response)
await client.close()
asyncio.run(test_connection())