Documentation Index
Fetch the complete documentation index at: https://ai-gateway.juniortree.com/llms.txt
Use this file to discover all available pages before exploring further.
开发概述
AI Gateway 采用插件化架构设计，支持轻松扩展新的AI平台。通过实现标准的BaseAgent接口，您可以快速集成任何AI服务。
流程展示
开发步骤
1. 继承BaseAgent基类
所有Agent都必须继承自BaseAgent抽象基类：
from abc import ABC, abstractmethod
from typing import Dict, Any
from fastapi import UploadFile
class BaseAgent(ABC):
@abstractmethod
async def chat(self, request: ChatRequest) -> UnifiedChatResponse:
pass
@abstractmethod
async def upload_file(self, file: UploadFile, created_by: str) -> FileResponse:
pass
2. 实现核心方法
创建新的Agent类并实现必要的方法：
class NewPlatformAgent(BaseAgent):
def __init__(self, config: Dict[str, Any]):
self.config = config
self._validate_config()
async def chat(self, request: ChatRequest) -> UnifiedChatResponse:
# 1. 转换请求格式
platform_request = self._convert_request(request)
# 2. 调用平台API
response = await self._call_platform_api(platform_request)
# 3. 转换响应格式
return self.format_response(response)
async def upload_file(self, file: UploadFile, created_by: str) -> FileResponse:
# 实现文件上传逻辑
pass
3. 注册Agent
将新Agent注册到注册表中：
from .registry import registry
def register_new_platform_agent():
config = {
"base_url": os.getenv("NEW_PLATFORM_BASE_URL"),
"api_key": os.getenv("NEW_PLATFORM_API_KEY")
}
if config["base_url"] and config["api_key"]:
agent = NewPlatformAgent(config)
registry.register("new_platform", agent)
# 自动注册
register_new_platform_agent()
4. 配置环境变量
在 .env 文件中添加配置：
NEW_PLATFORM_BASE_URL=https://api.newplatform.com
NEW_PLATFORM_API_KEY=your-api-key
关键实现要点
请求格式转换
将统一的请求格式转换为平台特定格式：
def _convert_request(self, request: ChatRequest) -> Dict[str, Any]:
return {
"message": request.query,
"user_id": request.user,
"stream": request.stream
}
响应格式转换
将平台响应转换为统一格式：
def format_response(self, response: Any) -> UnifiedChatResponse:
return UnifiedChatResponse(
id=response.get("id"),
model=response.get("model"),
usage=response.get("usage"),
choices=response.get("choices", [])
)
错误处理
实现完善的错误处理：
async def chat(self, request: ChatRequest) -> UnifiedChatResponse:
try:
# 实现聊天逻辑
pass
except Exception as e:
logger.error(f"Chat error: {e}")
raise HTTPException(status_code=500, detail="Internal error")
测试验证
单元测试
def test_config_validation(self):
"""测试配置验证"""
with pytest.raises(ValueError):
NewPlatformAgent({})
async def test_chat(self):
"""测试聊天功能"""
request = ChatRequest(query="Hello", user="test_user")
response = await self.agent.chat(request)
assert response.id is not None
集成测试
async def test_api_integration():
"""测试API集成"""
response = await client.post(
"/api/v1/chat",
headers={"agent": "new_platform"},
json={"query": "Hello", "user": "test_user"}
)
assert response.status_code == 200
部署检查清单
