taijibaga/taiji-AI-PAD
9
0
Fork 0
forked from xiaohei/taiji-AI-PAD
Code Pull Requests Activity

更新api文档

Browse Source
This commit is contained in:
Ubuntu
2025-12-31 10:56:00 +00:00
parent fb1f5a7b28
commit 56c390077e
18 changed files with 4470 additions and 5309 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plans/aks-agent-execution-plan.md
+740
View File
@@ -0,0 +1,740 @@
# AKS Agent 执行方案设计
## 1. 问题分析
### 当前状态
```
用户请求 → MCP Server → 本地 MCP 协议处理 → 工具执行
↓
(未使用 AKS Pod 的 access_url)
```
**问题**：虽然 AKS 部署后返回了 `access_url` 和 `endpoints`，但当前代码并没有使用这些 URL 来调用 AKS 中运行的 Pod。
### 目标状态
```
用户请求 → MCP Server → 判断 Agent 类型 → 转发到 AKS Pod → 返回结果
↓
本地执行（无 K8s 部署）
```
---
## 2. 架构设计
### 2.1 整体架构图
```mermaid
flowchart TB
subgraph Client[客户端]
User[用户]
end
subgraph MCPServer[MCP Server]
API[API Gateway]
Router[Agent Router]
LocalHandler[本地 MCP Handler]
K8sProxy[K8s Agent Proxy]
end
subgraph AKS[Azure Kubernetes Service]
Pod1[Agent Pod 1]
Pod2[Agent Pod 2]
Pod3[Agent Pod N]
end
subgraph AgentManager[Agent Manager Service]
AM[Agent Manager API]
end
User --> API
API --> Router
Router -->|无 K8s 部署| LocalHandler
Router -->|有 K8s 部署| K8sProxy
K8sProxy --> Pod1
K8sProxy --> Pod2
K8sProxy --> Pod3
AM -.->|管理| Pod1
AM -.->|管理| Pod2
AM -.->|管理| Pod3
```
### 2.2 执行流程图
```mermaid
sequenceDiagram
participant User as 用户
participant API as MCP Server API
participant Router as Agent Router
participant DB as 数据库
participant Proxy as K8s Proxy
participant Pod as AKS Agent Pod
participant Local as 本地 Handler
User->>API: POST /agents/{id}/execute
API->>DB: 获取 Agent 信息
DB-->>API: Agent 数据
API->>Router: 路由决策
alt Agent 有 access_url
Router->>Proxy: 转发请求
Proxy->>Pod: HTTP POST /execute
Pod-->>Proxy: 执行结果
Proxy-->>API: 返回结果
else Agent 无 K8s 部署
Router->>Local: 本地执行
Local-->>API: 执行结果
end
API->>DB: 记录执行和计费
API-->>User: 返回结果
```
---
## 3. 详细设计
### 3.1 新增组件：K8s Agent Proxy
**文件位置**: `services/mcp-server/app/k8s_agent_proxy.py`
```python
"""
K8s Agent Proxy - 负责转发请求到 AKS 部署的 Agent Pod
"""
import httpx
import structlog
from typing import Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime
logger = structlog.get_logger(__name__)
@dataclass
class ProxyConfig:
"""代理配置"""
timeout: float = 30.0
max_retries: int = 3
retry_delay: float = 1.0
health_check_interval: int = 30
@dataclass
class ProxyResult:
"""代理执行结果"""
success: bool
result: Optional[Dict[str, Any]] = None
error: Optional[str] = None
execution_time_ms: float = 0.0
pod_name: Optional[str] = None
status_code: Optional[int] = None
class K8sAgentProxy:
"""K8s Agent 代理类"""
def __init__(self, config: Optional[ProxyConfig] = None):
self.config = config or ProxyConfig()
self._client: Optional[httpx.AsyncClient] = None
async def _get_client(self) -> httpx.AsyncClient:
"""获取或创建 HTTP 客户端"""
if self._client is None or self._client.is_closed:
self._client = httpx.AsyncClient(
timeout=self.config.timeout,
headers={"Content-Type": "application/json"}
)
return self._client
async def execute(
self,
access_url: str,
request_data: Dict[str, Any],
pod_name: Optional[str] = None,
headers: Optional[Dict[str, str]] = None
) -> ProxyResult:
"""
转发执行请求到 AKS Agent Pod
Args:
access_url: Agent Pod 的访问 URL
request_data: MCP 请求数据
pod_name: Pod 名称（用于日志）
headers: 额外的请求头
Returns:
ProxyResult: 执行结果
"""
start_time = datetime.utcnow()
try:
client = await self._get_client()
# 构建完整的执行 URL
execute_url = f"{access_url.rstrip('/')}/execute"
logger.info(
"转发请求到 AKS Agent",
url=execute_url,
pod_name=pod_name,
method=request_data.get("method")
)
# 合并请求头
request_headers = {"Content-Type": "application/json"}
if headers:
request_headers.update(headers)
# 发送请求（带重试）
response = await self._request_with_retry(
client, execute_url, request_data, request_headers
)
execution_time = (datetime.utcnow() - start_time).total_seconds() * 1000
if response.status_code == 200:
result_data = response.json()
return ProxyResult(
success=True,
result=result_data,
execution_time_ms=execution_time,
pod_name=pod_name,
status_code=response.status_code
)
else:
error_detail = response.text
try:
error_detail = response.json()
except Exception:
pass
return ProxyResult(
success=False,
error=f"Pod 返回错误: {response.status_code} - {error_detail}",
execution_time_ms=execution_time,
pod_name=pod_name,
status_code=response.status_code
)
except httpx.TimeoutException as e:
execution_time = (datetime.utcnow() - start_time).total_seconds() * 1000
logger.error("请求 AKS Agent 超时", pod_name=pod_name, error=str(e))
return ProxyResult(
success=False,
error=f"请求超时: {str(e)}",
execution_time_ms=execution_time,
pod_name=pod_name
)
except httpx.ConnectError as e:
execution_time = (datetime.utcnow() - start_time).total_seconds() * 1000
logger.error("无法连接到 AKS Agent", pod_name=pod_name, error=str(e))
return ProxyResult(
success=False,
error=f"连接失败: {str(e)}",
execution_time_ms=execution_time,
pod_name=pod_name
)
except Exception as e:
execution_time = (datetime.utcnow() - start_time).total_seconds() * 1000
logger.error("转发请求失败", pod_name=pod_name, error=str(e))
return ProxyResult(
success=False,
error=f"执行失败: {str(e)}",
execution_time_ms=execution_time,
pod_name=pod_name
)
async def _request_with_retry(
self,
client: httpx.AsyncClient,
url: str,
data: Dict[str, Any],
headers: Dict[str, str]
) -> httpx.Response:
"""带重试的请求"""
import asyncio
last_exception = None
for attempt in range(self.config.max_retries):
try:
response = await client.post(url, json=data, headers=headers)
return response
except (httpx.TimeoutException, httpx.ConnectError) as e:
last_exception = e
if attempt < self.config.max_retries - 1:
await asyncio.sleep(self.config.retry_delay * (attempt + 1))
logger.warning(
f"重试请求 {attempt + 1}/{self.config.max_retries}",
url=url,
error=str(e)
)
raise last_exception
async def health_check(self, access_url: str) -> bool:
"""检查 Agent Pod 健康状态"""
try:
client = await self._get_client()
health_url = f"{access_url.rstrip('/')}/health"
response = await client.get(health_url, timeout=5.0)
return response.status_code == 200
except Exception:
return False
async def close(self):
"""关闭客户端"""
if self._client and not self._client.is_closed:
await self._client.aclose()
self._client = None
# 全局代理实例
_k8s_proxy: Optional[K8sAgentProxy] = None
def get_k8s_agent_proxy() -> K8sAgentProxy:
"""获取全局 K8s Agent 代理实例"""
global _k8s_proxy
if _k8s_proxy is None:
_k8s_proxy = K8sAgentProxy()
return _k8s_proxy
async def close_k8s_agent_proxy():
"""关闭全局代理"""
global _k8s_proxy
if _k8s_proxy:
await _k8s_proxy.close()
_k8s_proxy = None
```
### 3.2 修改 Agent 执行端点
**文件位置**: `services/mcp-server/app/routes/agents.py`
修改 `execute_agent` 函数：
```python
@router.post("/{agent_id}/execute", response_model=ExecutionResult)
async def execute_agent(
agent_id: str,
request: MCPRequest,
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(get_current_user),
session_id: Optional[str] = None,
) -> ExecutionResult:
"""
执行 Agent 请求。
如果 Agent 部署在 AKS 中，请求将被转发到对应的 Pod。
否则，请求将在本地执行。
"""
state = get_state()
try:
agent_uuid = uuid.UUID(agent_id)
user_id = uuid.UUID(current_user["user_id"])
except ValueError as exc:
raise HTTPException(status_code=400, detail="Invalid agent ID or user ID") from exc
agent = await db.get(Agent, agent_uuid)
if not agent:
raise HTTPException(status_code=404, detail="Agent not found")
# 权限检查
if agent.owner_id != user_id and current_user.get("role") != "super_admin":
raise HTTPException(status_code=403, detail="Access denied")
# 资源管控检查
await enforce_resource_control(
user_id=str(user_id),
resource_type="agent",
resource_id=agent_id,
estimated_cost=Decimal("0.01"),
db=db
)
start_time = time.time()
execution_id = str(uuid.uuid4())
try:
# ========== 路由决策：K8s Pod 或本地执行 ==========
if agent.access_url and agent.k8s_status == "Running":
# 转发到 AKS Agent Pod
result = await _execute_on_k8s_pod(
agent=agent,
request=request,
execution_id=execution_id,
user_id=str(user_id)
)
else:
# 本地执行
result = await _execute_locally(
agent=agent,
request=request,
execution_id=execution_id,
user_id=str(user_id),
db=db
)
# ================================================
duration = time.time() - start_time
# 记录执行和计费
await _record_execution_and_billing(
db=db,
agent=agent,
request=request,
result=result,
execution_id=execution_id,
start_time=start_time,
duration=duration,
session_id=session_id
)
return result
except Exception as exc:
duration = time.time() - start_time
logger.error("执行Agent任务失败", error=str(exc))
raise HTTPException(status_code=500, detail=str(exc)) from exc
async def _execute_on_k8s_pod(
agent: Agent,
request: MCPRequest,
execution_id: str,
user_id: str
) -> ExecutionResult:
"""转发请求到 AKS Agent Pod"""
from ..k8s_agent_proxy import get_k8s_agent_proxy
proxy = get_k8s_agent_proxy()
# 构建请求数据
request_data = {
"jsonrpc": request.jsonrpc,
"id": str(request.id),
"method": request.method,
"params": request.params or {},
"metadata": {
"execution_id": execution_id,
"user_id": user_id,
"agent_id": str(agent.id)
}
}
# 转发请求
proxy_result = await proxy.execute(
access_url=agent.access_url,
request_data=request_data,
pod_name=agent.pod_name
)
if proxy_result.success:
return ExecutionResult(
execution_id=execution_id,
success=True,
result=proxy_result.result,
execution_time=proxy_result.execution_time_ms,
started_at=datetime.utcnow(),
completed_at=datetime.utcnow()
)
else:
return ExecutionResult(
execution_id=execution_id,
success=False,
error=proxy_result.error,
execution_time=proxy_result.execution_time_ms,
started_at=datetime.utcnow(),
completed_at=datetime.utcnow()
)
async def _execute_locally(
agent: Agent,
request: MCPRequest,
execution_id: str,
user_id: str,
db: AsyncSession
) -> ExecutionResult:
"""本地执行 MCP 请求"""
state = get_state()
handler = state.mcp_handler
if not handler:
raise HTTPException(status_code=500, detail="MCP handler not initialized")
return await handler.execute_request(
str(agent.id),
request,
user_id=user_id,
db_session=db
)
```
### 3.3 AKS Agent Pod 端点规范
每个部署在 AKS 中的 Agent Pod 需要实现以下端点：
| 端点 | 方法 | 描述 |
|-----|------|------|
| `/health` | GET | 健康检查 |
| `/execute` | POST | 执行 MCP 请求 |
| `/status` | GET | 获取 Agent 状态 |
| `/metrics` | GET | 获取资源使用指标 |
#### 3.3.1 `/execute` 端点请求格式
```json
{
"jsonrpc": "2.0",
"id": "request-uuid",
"method": "tools/call",
"params": {
"name": "tool_name",
"arguments": {}
},
"metadata": {
"execution_id": "exec-uuid",
"user_id": "user-uuid",
"agent_id": "agent-uuid"
}
}
```
#### 3.3.2 `/execute` 端点响应格式
```json
{
"success": true,
"result": {
"content": [
{
"type": "text",
"text": "执行结果"
}
]
},
"execution_time_ms": 150.5,
"resource_usage": {
"cpu_ms": 50,
"memory_mb": 128
}
}
```
---
## 4. 数据库变更
### 4.1 Agent 表新增字段（已存在）
当前 `Agent` 模型已包含必要字段：
| 字段 | 类型 | 描述 |
|-----|------|------|
| `access_url` | String(500) | Pod 访问 URL |
| `pod_name` | String(100) | Pod 名称 |
| `pod_ip` | String(45) | Pod IP |
| `k8s_status` | String(20) | Pod 状态 |
| `service_port` | Integer | Service 端口 |
| `endpoints` | JSON | 端点字典 |
### 4.2 新增执行记录字段
在 `Execution` 表中添加：
```python
# 执行位置
execution_location = Column(String(20), default="local") # local, k8s
pod_name = Column(String(100)) # 执行的 Pod 名称
```
---
## 5. 配置变更
### 5.1 环境变量
```bash
# K8s Agent Proxy 配置
K8S_PROXY_TIMEOUT=30.0
K8S_PROXY_MAX_RETRIES=3
K8S_PROXY_RETRY_DELAY=1.0
K8S_PROXY_HEALTH_CHECK_INTERVAL=30
```
### 5.2 应用配置
在 `config.py` 中添加：
```python
class K8sProxyConfig:
timeout: float = float(os.getenv("K8S_PROXY_TIMEOUT", "30.0"))
max_retries: int = int(os.getenv("K8S_PROXY_MAX_RETRIES", "3"))
retry_delay: float = float(os.getenv("K8S_PROXY_RETRY_DELAY", "1.0"))
health_check_interval: int = int(os.getenv("K8S_PROXY_HEALTH_CHECK_INTERVAL", "30"))
```
---
## 6. 实施计划
### 6.1 任务清单
- [ ] **Phase 1: 基础设施**
- [ ] 创建 `k8s_agent_proxy.py` 模块
- [ ] 添加配置项
- [ ] 编写单元测试
- [ ] **Phase 2: 路由逻辑**
- [ ] 修改 `execute_agent` 端点
- [ ] 实现路由决策逻辑
- [ ] 添加本地执行回退
- [ ] **Phase 3: 监控和日志**
- [ ] 添加执行位置记录
- [ ] 添加 Prometheus 指标
- [ ] 完善日志记录
- [ ] **Phase 4: 健康检查**
- [ ] 实现 Pod 健康检查
- [ ] 添加自动故障转移
- [ ] 实现连接池管理
- [ ] **Phase 5: 测试和文档**
- [ ] 集成测试
- [ ] 性能测试
- [ ] 更新 API 文档
### 6.2 文件变更清单
| 文件 | 操作 | 描述 |
|-----|------|------|
| `services/mcp-server/app/k8s_agent_proxy.py` | 新增 | K8s Agent 代理模块 |
| `services/mcp-server/app/routes/agents.py` | 修改 | 添加路由逻辑 |
| `services/mcp-server/config.py` | 修改 | 添加代理配置 |
| `services/mcp-server/app/lifecycle.py` | 修改 | 添加代理生命周期管理 |
| `services/mcp-server/models.py` | 修改 | 添加执行位置字段 |
---
## 7. 错误处理
### 7.1 错误场景和处理策略
| 场景 | 处理策略 |
|-----|---------|
| Pod 不可达 | 重试 3 次后返回错误 |
| Pod 返回 5xx | 记录错误，返回给用户 |
| 请求超时 | 返回超时错误，建议重试 |
| Pod 状态非 Running | 回退到本地执行或返回错误 |
| access_url 为空 | 使用本地执行 |
### 7.2 故障转移策略
```python
async def execute_with_fallback(agent, request, ...):
"""带故障转移的执行"""
# 1. 尝试 K8s Pod 执行
if agent.access_url and agent.k8s_status == "Running":
result = await _execute_on_k8s_pod(...)
if result.success:
return result
# 2. K8s 执行失败，检查是否可以本地执行
if agent.tools and not agent.template:
logger.warning("K8s 执行失败，回退到本地执行")
return await _execute_locally(...)
# 3. 本地执行
return await _execute_locally(...)
```
---
## 8. 监控指标
### 8.1 新增 Prometheus 指标
```python
# K8s Agent 执行指标
k8s_agent_requests_total = Counter(
"k8s_agent_requests_total",
"Total K8s agent requests",
["pod_name", "status"]
)
k8s_agent_request_duration = Histogram(
"k8s_agent_request_duration_seconds",
"K8s agent request duration",
["pod_name"]
)
k8s_agent_health_status = Gauge(
"k8s_agent_health_status",
"K8s agent health status",
["pod_name"]
)
```
### 8.2 日志格式
```json
{
"timestamp": "2024-01-01T00:00:00Z",
"level": "INFO",
"message": "转发请求到 AKS Agent",
"execution_id": "exec-uuid",
"agent_id": "agent-uuid",
"pod_name": "my-agent-abc123",
"access_url": "http://my-agent.ai-agents.svc.cluster.local:8080",
"method": "tools/call",
"execution_location": "k8s"
}
```
---
## 9. 安全考虑
### 9.1 网络安全
- Pod 间通信使用 K8s 内部网络
- 不暴露 Pod 到公网
- 使用 NetworkPolicy 限制访问
### 9.2 认证授权
- 请求中携带 `user_id` 和 `execution_id`
- Pod 可验证请求来源
- 支持 mTLS（可选）
### 9.3 数据安全
- 敏感数据不在日志中记录
- 请求/响应数据加密传输
- 执行结果脱敏存储
---
## 10. 总结
本方案实现了 MCP Server 与 AKS Agent Pod 的集成，主要特点：
1. **智能路由**：根据 Agent 配置自动选择执行位置
2. **故障转移**：K8s 执行失败时可回退到本地
3. **可观测性**：完整的日志、指标和追踪
4. **安全性**：网络隔离和认证机制
5. **可扩展性**：支持多 Pod 负载均衡（未来）
通过此方案，用户可以透明地使用部署在 AKS 中的 Agent，无需关心底层执行细节。
plans/database-migration-plan.md
+166
View File
@@ -0,0 +1,166 @@
# 数据库迁移计划：postgres → taiji
**创建时间**: 2025-12-31
**目标**: 将 postgres 库的表结构和数据完全覆盖到 taiji 库
---
## 📋 任务概述
将 Azure PostgreSQL 服务器上的 `postgres` 数据库（新结构）完全复制到 `taiji` 数据库（旧结构），包括：
- 表结构
- 索引
- 约束
- 数据
- 序列
---
## 🔄 迁移流程图
```mermaid
flowchart TD
A[开始迁移] --> B[连接 postgres 源数据库]
B --> C[连接 taiji 目标数据库]
C --> D[备份 taiji 数据库 - 可选]
D --> E[删除 taiji 中的所有表]
E --> F[从 postgres 获取表结构]
F --> G[在 taiji 中创建表]
G --> H[创建索引和约束]
H --> I[复制数据]
I --> J[同步序列值]
J --> K[验证迁移结果]
K --> L[完成]
```
---
## ✅ 任务清单
### 1. 准备工作
- [ ] 确认数据库连接信息正确
- [ ] 确认 postgres 库中有最新的表结构
- [ ] 备份 taiji 库现有数据（可选但推荐）
### 2. 创建迁移脚本
- [ ] 修改现有 `copy_database.py` 脚本，交换源和目标数据库
- [ ] 或创建新脚本 `sync_postgres_to_taiji.py`
### 3. 脚本功能实现
- [ ] 连接源数据库（postgres）
- [ ] 连接目标数据库（taiji）
- [ ] 获取 postgres 库所有表列表
- [ ] 删除 taiji 库中的所有现有表（CASCADE）
- [ ] 复制表结构（DDL）
- [ ] 复制索引定义
- [ ] 复制数据
- [ ] 同步序列值
### 4. 验证和测试
- [ ] 验证表数量一致
- [ ] 验证数据行数一致
- [ ] 验证索引创建成功
- [ ] 测试应用连接 taiji 库正常工作
---
## 📝 技术细节
### 数据库连接信息
```python
DB_HOST = "taijipda.postgres.database.azure.com"
DB_USER = "taiji"
DB_PASSWORD = "By@123456."
DB_PORT = 5432
# 源数据库（新结构）
SOURCE_DB = "postgres"
# 目标数据库（需要更新）
TARGET_DB = "taiji"
```
### 主要表列表（基于 models.py）
| 表名 | 说明 |
|------|------|
| users | 用户表（租户使用者） |
| agents | Agent表（平台Agent和自定义Agent） |
| tools | 工具表 |
| sessions | 会话表 |
| executions | 执行记录表 |
| api_keys | API密钥表 |
| billing | 计费详情表 |
| balances | 用户余额表 |
| billing_records | 计费记录表 |
| channels | 渠道合作伙伴表 |
| model_providers | 模型供应商表 |
| resource_allocations | 资源分配表 |
| applications | 申请审批表 |
| workflows | 工作流表 |
| audit_logs | 审计日志表 |
| channel_agent_quotas | 渠道Agent配额表 |
| provider_models | 模型提供商表 |
| token_blacklist | Token黑名单表 |
| resource_usage | 资源使用记录表 |
| quota_alerts | 配额预警记录表 |
| model_pricing | 模型定价配置表 |
| provider_health_checks | 供应商健康检查记录表 |
| agent_traces | Agent执行轨迹表 |
| billing_events | 计费事件表 |
| channel_provider_access | 渠道供应商授权表 |
| provider_applications | 供应商使用申请表 |
| gateway_apis | 网关API定义表 |
| data_templates | 数据模板表 |
| recharge_records | 充值记录表 |
---
## ⚠️ 注意事项
1. **数据丢失风险**: 此操作会删除 taiji 库中的所有现有数据，请确保已备份
2. **外键约束**: 删除表时使用 CASCADE 处理外键依赖
3. **序列同步**: 确保序列值正确同步，避免主键冲突
4. **连接中断**: 迁移过程中确保网络稳定
5. **应用停机**: 建议在迁移期间停止连接 taiji 库的应用服务
---
## 🚀 执行步骤
1. **运行迁移脚本**:
```bash
cd /home/taiji/tools/taiji-AI-PAD
python scripts/sync_postgres_to_taiji.py
```
2. **验证迁移结果**:
```bash
# 连接 taiji 库检查表
psql "host=taijipda.postgres.database.azure.com port=5432 dbname=taiji user=taiji password=By@123456. sslmode=require"
# 查看所有表
\dt
# 检查数据行数
SELECT COUNT(*) FROM users;
SELECT COUNT(*) FROM agents;
```
3. **更新应用配置**（如需要）:
确保应用的 DATABASE_URL 指向 taiji 库
---
## 📊 预期结果
迁移完成后：
- taiji 库将拥有与 postgres 库完全相同的表结构
- 所有数据将从 postgres 库复制到 taiji 库
- 索引和约束将正确创建
- 序列值将同步
---
**下一步**: 切换到 Code 模式创建迁移脚本