diff --git a/API_DOCUMENTATION.md b/API_DOCUMENTATION.md index 4992936..adb5e02 100644 --- a/API_DOCUMENTATION.md +++ b/API_DOCUMENTATION.md @@ -44,7 +44,7 @@ Content-Type: application/json "memory_request": "128Mi", // 可选，内存请求量 "memory_limit": "512Mi" // 可选，内存限制 }, - "env": { // 可选，环境变量 + "env_variables": { // 可选，环境变量 "KEY": "value" } } @@ -52,15 +52,72 @@ Content-Type: application/json **支持的模板类型** -| 模板 | 说明 | 类型 | -|------|------|------| -| `echo_agent` | Echo 测试服务 | 平台 | -| `chat_agent` | 聊天服务 | 平台 | -| `code_agent` | 代码执行服务 | 平台 | -| `search_agent` | 搜索服务 | 平台 | -| `jina_search_agent` | Jina 搜索服务 | 平台 | -| `mysql_agent` | MySQL 客户端 | 自定义 | -| `postgresql_agent` | PostgreSQL 客户端 | 自定义 | +| 模板 | 说明 | 类型 | 框架 | +|------|------|------|------| +| `echo_agent` | Echo 测试服务 | 平台 | - | +| `chat_agent` | 聊天服务 | 平台 | - | +| `code_agent` | 代码执行服务 | 平台 | - | +| `search_agent` | 搜索服务 | 平台 | - | +| `jina_search_agent` | Jina 搜索服务 | 平台 | - | +| `mysql_agent` | MySQL 客户端 | 自定义 | - | +| `postgresql_agent` | PostgreSQL 客户端 | 自定义 | - | +| `azure_blob_agent` | Azure Blob Storage 客户端 (LangChain) | 自定义 | LangChain | +| `azure_blob_agent_mcp` | Azure Blob Storage 客户端 (MCP) | 自定义 | MCP | +| `azure_blob_agent_a2a` | Azure Blob Storage 客户端 (A2A) | 自定义 | A2A | + +**Agent 框架说明** + +从 v1.1.0 开始，Agent Manager 支持多种 AI Agent 框架： + +| 框架 | 说明 | 适用场景 | +|------|------|---------| +| **LangChain** | 使用 LangChain + LiteLLM | 复杂推理任务、多步骤处理流程 | +| **MCP** | Model Context Protocol | 标准化工具调用、轻量级集成 | +| **A2A** | Agent-to-Agent | 多 Agent 协作、分布式任务处理 | + +**多框架支持的配置参数** + +创建支持多框架的 Agent 时，可以使用以下额外参数： + +```json +{ + "name": "agent-name", + "template": "azure_blob_agent_mcp", + "config": { + // 基础配置 + "user_id": "user-001", + "tenant_id": "tenant-001", + "namespace": "ai-agents", + + // 框架配置 + "agent_framework": "mcp", // 框架类型: langchain, mcp, a2a + + // 工具配置 + "tools_config": { // 工具配置 JSON + "max_iterations": 5, + "enabled_tools": ["list_containers", "list_blobs"] + }, + "tool_endpoint": "http://tools-api:8080", // 外部工具端点 + "tool_api_key": "tool-key", // 工具 API 密钥 + + // 模型配置 + "model_provider": "openai", // 模型提供商: openai, azure-openai + "model_name": "gpt-4", // 模型名称 + "model_endpoint": "https://api.openai.com/v1", // 模型端点 + "model_api_key": "sk-xxxx", // 模型 API 密钥 + + // 存储配置 (针对 Azure Blob Agent) + "storage_connection_string": "DefaultEndpointsProtocol=https;...", + "storage_account_name": "myaccount", + + // 资源配置 + "cpu_request": "100m", + "cpu_limit": "500m", + "memory_request": "256Mi", + "memory_limit": "512Mi" + } +} +``` **响应** @@ -100,6 +157,7 @@ Content-Type: application/json **示例** +基础示例 - Echo Agent： ```bash curl -X POST http://localhost:8000/agents \ -H "Content-Type: application/json" \ @@ -112,6 +170,277 @@ curl -X POST http://localhost:8000/agents \ }' ``` +Azure Blob Agent (LangChain 版本) - 提供连接字符串： +```bash +curl -X POST http://localhost:8000/agents \ + -H "Content-Type: application/json" \ + -d '{ + "name": "my-azure-blob-agent", + "template": "azure_blob_agent", + "config": { + "user_id": "alice" + }, + "env": { + "LITELLM_API_BASE": "http://litellm-service:4000", + "LITELLM_MODEL": "gpt-4", + "LITELLM_API_KEY": "sk-your-api-key", + "AZURE_STORAGE_CONNECTION_STRING": "DefaultEndpointsProtocol=https;AccountName=youraccount;AccountKey=yourkey;EndpointSuffix=core.windows.net", + "SERVICE_PORT": "8080" + } + }' +``` + +**Azure Blob Agent (MCP 版本) - 标准化工具调用**： +```bash +curl -X POST http://localhost:8000/agents \ + -H "Content-Type: application/json" \ + -d '{ + "name": "my-blob-mcp", + "template": "azure_blob_agent_mcp", + "config": { + "user_id": "alice", + "tenant_id": "tenant-001", + "namespace": "ai-agents", + "agent_framework": "mcp", + "model_provider": "openai", + "model_name": "gpt-4", + "model_api_key": "sk-your-api-key", + "model_endpoint": "https://api.openai.com/v1", + "storage_connection_string": "DefaultEndpointsProtocol=https;AccountName=youraccount;AccountKey=yourkey;EndpointSuffix=core.windows.net", + "tools_config": { + "max_iterations": 5, + "enabled_tools": ["list_containers", "list_blobs", "search_blobs"] + }, + "cpu_request": "100m", + "memory_request": "256Mi" + } + }' +``` + +**Azure Blob Agent (A2A 版本) - Agent 间协作**： +```bash +curl -X POST http://localhost:8000/agents \ + -H "Content-Type: application/json" \ + -d '{ + "name": "my-blob-a2a", + "template": "azure_blob_agent_a2a", + "config": { + "user_id": "alice", + "tenant_id": "tenant-001", + "namespace": "ai-agents", + "agent_framework": "a2a", + "model_provider": "azure-openai", + "model_name": "gpt-4", + "model_api_key": "your-azure-openai-key", + "model_endpoint": "https://your-resource.openai.azure.com", + "storage_connection_string": "DefaultEndpointsProtocol=https;AccountName=youraccount;AccountKey=yourkey;EndpointSuffix=core.windows.net", + "cpu_request": "100m", + "memory_request": "256Mi" + }, + "env": { + "AGENT_ID": "blob-agent-001", + "AGENT_ROLE": "storage_manager", + "AGENT_CAPABILITIES": "[\"blob_storage\", \"file_operations\"]" + } + }' +``` + +**测试部署后的 Agent** + +**LangChain 版本测试：** + +获取 Pod IP 并测试： +```bash +# 1. 检查 Agent 状态 +curl -s http://localhost:8000/agents/my-azure-blob-agent/status | jq '{status, health_status, pod_ip, access_url}' + +# 2. 获取 Pod IP +POD_IP=$(curl -s http://localhost:8000/agents/my-azure-blob-agent/status | jq -r '.pod_ip') +echo "Pod IP: $POD_IP" + +# 3. 测试健康检查 +curl http://$POD_IP:8080/health + +# 4. 查看 Agent 信息 +curl http://$POD_IP:8080/ | jq . + +# 5. 如果启动时未提供连接字符串，可以动态连接 +curl -X POST http://$POD_IP:8080/connect \ + -H "Content-Type: application/json" \ + -d '{ + "connection_string": "DefaultEndpointsProtocol=https;AccountName=youraccount;AccountKey=yourkey;EndpointSuffix=core.windows.net" + }' + +# 6. 检查连接状态 +curl http://$POD_IP:8080/status + +# 7. 执行自然语言查询 - 列出所有容器 +curl -X POST http://$POD_IP:8080/query \ + -H "Content-Type: application/json" \ + -d '{"query": "列出所有容器"}' | jq . + +# 8. 查看指定容器的文件 +curl -X POST http://$POD_IP:8080/query \ + -H "Content-Type: application/json" \ + -d '{"query": "显示 mycontainer 容器中的所有文件"}' | jq . + +# 9. 搜索文件 +curl -X POST http://$POD_IP:8080/query \ + -H "Content-Type: application/json" \ + -d '{"query": "搜索包含 report 的文件"}' | jq . + +# 10. 获取存储统计 +curl -X POST http://$POD_IP:8080/query \ + -H "Content-Type: application/json" \ + -d '{"query": "存储统计"}' | jq . +``` + +**MCP 版本测试：** + +```bash +# 1. 获取 Pod IP +POD_IP=$(curl -s http://localhost:8000/agents/my-blob-mcp/status | jq -r '.pod_ip') + +# 2. 测试健康检查 +curl http://$POD_IP:8080/health | jq . + +# 3. 查看 Agent 信息（包含框架类型） +curl http://$POD_IP:8080/ | jq . + +# 4. 列出所有可用的 MCP 工具 +curl http://$POD_IP:8080/mcp/tools | jq . + +# 5. 调用 MCP 工具 - 列出所有容器 +curl -X POST http://$POD_IP:8080/mcp/call \ + -H "Content-Type: application/json" \ + -d '{ + "tool_name": "list_containers", + "parameters": {} + }' | jq . + +# 6. 调用 MCP 工具 - 列出容器中的文件 +curl -X POST http://$POD_IP:8080/mcp/call \ + -H "Content-Type: application/json" \ + -d '{ + "tool_name": "list_blobs", + "parameters": { + "container_name": "mycontainer" + } + }' | jq . + +# 7. 调用 MCP 工具 - 获取文件信息 +curl -X POST http://$POD_IP:8080/mcp/call \ + -H "Content-Type: application/json" \ + -d '{ + "tool_name": "get_blob_info", + "parameters": { + "container_name": "mycontainer", + "blob_name": "myfile.txt" + } + }' | jq . + +# 8. 调用 MCP 工具 - 搜索文件 +curl -X POST http://$POD_IP:8080/mcp/call \ + -H "Content-Type: application/json" \ + -d '{ + "tool_name": "search_blobs", + "parameters": { + "container_name": "mycontainer", + "keyword": "report" + } + }' | jq . + +# 9. 调用 MCP 工具 - 获取存储统计 +curl -X POST http://$POD_IP:8080/mcp/call \ + -H "Content-Type: application/json" \ + -d '{ + "tool_name": "get_storage_stats", + "parameters": {} + }' | jq . + +# 10. 使用简化的查询接口（规则匹配） +curl -X POST http://$POD_IP:8080/query \ + -H "Content-Type: application/json" \ + -d '{"query": "列出所有容器"}' | jq . +``` + +**A2A 版本测试：** + +```bash +# 1. 获取 Pod IP +POD_IP=$(curl -s http://localhost:8000/agents/my-blob-a2a/status | jq -r '.pod_ip') + +# 2. 测试健康检查（包含 Agent 身份信息） +curl http://$POD_IP:8080/health | jq . + +# 3. 获取 Agent 能力 +curl http://$POD_IP:8080/a2a/capabilities | jq . + +# 4. 发送 A2A 消息 - 列出容器 +curl -X POST http://$POD_IP:8080/a2a/message \ + -H "Content-Type: application/json" \ + -d '{ + "message_id": "msg-001", + "from_agent": "external-caller", + "to_agent": "blob-agent-001", + "message_type": "request", + "action": "list_containers", + "parameters": {} + }' | jq . + +# 5. 发送 A2A 消息 - 列出文件 +curl -X POST http://$POD_IP:8080/a2a/message \ + -H "Content-Type: application/json" \ + -d '{ + "message_id": "msg-002", + "from_agent": "external-caller", + "to_agent": "blob-agent-001", + "message_type": "request", + "action": "list_blobs", + "parameters": { + "container_name": "mycontainer" + } + }' | jq . + +# 6. 发送 A2A 消息 - 获取统计 +curl -X POST http://$POD_IP:8080/a2a/message \ + -H "Content-Type: application/json" \ + -d '{ + "message_id": "msg-003", + "from_agent": "external-caller", + "to_agent": "blob-agent-001", + "message_type": "request", + "action": "get_stats", + "parameters": {} + }' | jq . + +# 7. 注册另一个 Agent（用于协作） +curl -X POST http://$POD_IP:8080/a2a/register \ + -H "Content-Type: application/json" \ + -d '{ + "agent_id": "analytics-agent", + "agent_role": "data_analyzer", + "capabilities": ["data_analysis", "visualization"], + "endpoint": "http://analytics-agent:8080" + }' | jq . + +# 8. 列出已注册的 Agent +curl http://$POD_IP:8080/a2a/agents | jq . + +# 9. 与其他 Agent 协作（需要先注册目标 Agent） +curl -X POST http://$POD_IP:8080/a2a/collaborate \ + -H "Content-Type: application/json" \ + -d '{ + "target_agent_id": "analytics-agent", + "action": "analyze_data", + "parameters": { + "data_source": "blob_storage" + } + }' | jq . +``` + -d '{"query": "统计存储使用情况"}' | jq . +``` + --- ### 2. 查询 Agent 列表 @@ -1028,11 +1357,281 @@ class AgentManagerClient { ### v1.0.0 (2026-01-05) - ✅ 实现 Agent 创建和管理 -- ✅ 支持 7 种 Agent 模板 +- ✅ 支持 10 种 Agent 模板（包含 3 种框架版本） - ✅ 多租户支持（user-id 标签） +- ✅ 多框架支持（LangChain、MCP、A2A） - ✅ Pod ID 返回和归属验证 - ✅ 模板分类查询（平台/自定义） - ✅ 资源监控和状态查询 +- ✅ 可自定义命名空间 + +--- + +## 多框架 Agent 支持 (v1.1.0+) + +### 框架对比 + +| 特性 | LangChain | MCP | A2A | +|------|-----------|-----|-----| +| **实现方式** | LangChain + LiteLLM | Model Context Protocol | Agent-to-Agent Protocol | +| **工具调用** | LangChain Tools | MCP Tool Classes | A2A Action Handlers | +| **主要端点** | `/query` | `/mcp/tools`, `/mcp/call` | `/a2a/capabilities`, `/a2a/message` | +| **协作能力** | ❌ | ❌ | ✅ Agent 注册和通信 | +| **适用场景** | 复杂推理任务 | 标准化工具调用 | 多 Agent 协作 | +| **集成难度** | 中等 | 简单 | 中等 | + +### 配置参数说明 + +#### 通用参数（所有框架） + +| 参数 | 类型 | 必需 | 说明 | 示例 | +|------|------|------|------|------| +| `user_id` | string | ✅ | 用户标识 | `"user-001"` | +| `tenant_id` | string | ❌ | 租户标识 | `"tenant-001"` | +| `namespace` | string | ❌ | Kubernetes 命名空间 | `"ai-agents"` | +| `agent_framework` | string | ❌ | 框架类型 | `"mcp"` 或 `"a2a"` | +| `cpu_request` | string | ❌ | CPU 请求量 | `"100m"` | +| `cpu_limit` | string | ❌ | CPU 限制 | `"500m"` | +| `memory_request` | string | ❌ | 内存请求量 | `"256Mi"` | +| `memory_limit` | string | ❌ | 内存限制 | `"512Mi"` | + +#### 模型配置参数（MCP/A2A） + +| 参数 | 类型 | 必需 | 说明 | 示例 | +|------|------|------|------|------| +| `model_provider` | string | ✅ | 模型提供商 | `"openai"`, `"azure-openai"` | +| `model_name` | string | ✅ | 模型名称 | `"gpt-4"`, `"gpt-3.5-turbo"` | +| `model_api_key` | string | ✅ | 模型 API 密钥 | `"sk-xxxx"` | +| `model_endpoint` | string | ❌ | 模型 API 端点 | `"https://api.openai.com/v1"` | + +#### 工具配置参数（MCP/A2A） + +| 参数 | 类型 | 必需 | 说明 | 示例 | +|------|------|------|------|------| +| `tools_config` | object | ❌ | 工具配置 JSON | `{"max_iterations": 5}` | +| `tool_endpoint` | string | ❌ | 外部工具端点 | `"http://tools-api:8080"` | +| `tool_api_key` | string | ❌ | 工具 API 密钥 | `"tool-key-xxx"` | + +#### 存储配置参数（Azure Blob Agent） + +| 参数 | 类型 | 必需 | 说明 | +|------|------|------|------| +| `storage_connection_string` | string | ❌ | Azure Storage 连接字符串 | +| `storage_account_name` | string | ❌ | 存储账户名称 | + +### MCP 框架 API 端点 + +MCP Agent 部署后提供以下额外端点： + +#### GET /mcp/tools +列出所有可用的 MCP 工具 + +**响应示例：** +```json +{ + "tools": [ + { + "name": "list_containers", + "description": "列出 Azure Blob Storage 中的所有容器", + "inputSchema": { + "type": "object", + "properties": {}, + "required": [] + } + }, + { + "name": "list_blobs", + "description": "列出指定容器中的所有文件", + "inputSchema": { + "type": "object", + "properties": { + "container_name": { + "type": "string", + "description": "容器名称" + } + }, + "required": ["container_name"] + } + } + ], + "count": 5, + "framework": "mcp" +} +``` + +#### POST /mcp/call +调用指定的 MCP 工具 + +**请求：** +```json +{ + "tool_name": "list_blobs", + "parameters": { + "container_name": "mycontainer" + } +} +``` + +**响应：** +```json +{ + "tool": "list_blobs", + "result": { + "success": true, + "container": "mycontainer", + "blobs": [...], + "count": 10, + "total_size_mb": 125.5 + }, + "timestamp": "2026-01-12T14:50:00Z" +} +``` + +### A2A 框架 API 端点 + +A2A Agent 部署后提供以下额外端点： + +#### GET /a2a/capabilities +获取 Agent 的能力信息 + +**响应示例：** +```json +{ + "agent_id": "blob-agent-001", + "agent_role": "storage_manager", + "capabilities": ["blob_storage", "file_operations"], + "supported_actions": [ + "list_containers", + "list_blobs", + "get_blob_info", + "search_blobs", + "get_stats" + ], + "framework": "a2a" +} +``` + +#### POST /a2a/register +注册其他 Agent（用于协作） + +**请求：** +```json +{ + "agent_id": "analytics-agent", + "agent_role": "data_analyzer", + "capabilities": ["data_analysis", "visualization"], + "endpoint": "http://analytics-agent:8080" +} +``` + +#### POST /a2a/message +发送 A2A 消息给 Agent + +**请求：** +```json +{ + "message_id": "msg-001", + "from_agent": "caller-agent", + "to_agent": "blob-agent-001", + "message_type": "request", + "action": "list_containers", + "parameters": {} +} +``` + +**响应：** +```json +{ + "message_id": "msg-001", + "from_agent": "blob-agent-001", + "to_agent": "caller-agent", + "message_type": "response", + "action": "list_containers", + "result": { + "success": true, + "containers": [...], + "count": 5 + }, + "timestamp": "2026-01-12T14:50:00Z" +} +``` + +#### GET /a2a/agents +列出已注册的 Agent + +#### POST /a2a/collaborate +与其他 Agent 协作 + +### 命名空间支持 + +从 v1.1.0 开始，支持自定义 Kubernetes 命名空间： + +```bash +# 在自定义命名空间中创建 Agent +curl -X POST http://localhost:8000/agents \ + -H "Content-Type: application/json" \ + -d '{ + "name": "my-agent", + "template": "azure_blob_agent_mcp", + "config": { + "namespace": "my-namespace", + "user_id": "user-001", + ... + } + }' + +# 查询特定命名空间的 Agent +kubectl get pods -n my-namespace -l app=ai-agent + +# 通过 API 查询时，namespace 会在响应中返回 +curl http://localhost:8000/agents/my-agent | jq '.namespace' +``` + +### 环境变量传递 + +创建 Agent 时，以下配置会自动转换为容器环境变量： + +| 配置参数 | 环境变量名 | +|----------|-----------| +| `agent_framework` | `AGENT_FRAMEWORK` | +| `tools_config` | `TOOLS_CONFIG` (JSON字符串) | +| `tool_endpoint` | `TOOL_ENDPOINT` | +| `tool_api_key` | `TOOL_API_KEY` | +| `model_provider` | `MODEL_PROVIDER` | +| `model_name` | `MODEL_NAME` | +| `model_endpoint` | `MODEL_ENDPOINT` | +| `model_api_key` | `MODEL_API_KEY` | +| `storage_connection_string` | `AZURE_STORAGE_CONNECTION_STRING` | +| `storage_account_name` | `STORAGE_ACCOUNT_NAME` | +| `user_id` | `USER_ID` | +| `tenant_id` | `TENANT_ID` | +| `namespace` | `NAMESPACE` | + +### 故障排查 + +**Agent 创建失败** + +1. 检查模板名称是否正确 +2. 验证必需参数是否提供（如 model_api_key） +3. 查看 agent-manager 日志 + +**MCP 工具调用失败** + +1. 使用 `GET /mcp/tools` 确认工具名称 +2. 检查参数格式是否符合 inputSchema +3. 查看 agent pod 日志 + +**A2A Agent 无法协作** + +1. 确认目标 Agent 已注册 +2. 检查网络连接和端点可访问性 +3. 验证 message 格式是否正确 + +### 更多资源 + +- [多框架使用指南](agent_templates/MULTI_FRAMEWORK_GUIDE.md) +- [快速参考](agent_templates/QUICK_REFERENCE.md) +- [实现总结](MULTI_FRAMEWORK_SUMMARY.md) --- diff --git a/MULTI_FRAMEWORK_SUMMARY.md b/MULTI_FRAMEWORK_SUMMARY.md new file mode 100644 index 0000000..efeb9dc --- /dev/null +++ b/MULTI_FRAMEWORK_SUMMARY.md @@ -0,0 +1,353 @@ +# Azure Blob Agent 多框架实现总结 + +## 📋 概述 + +本次更新为 Azure Blob Storage Agent 实现了三种框架支持： +1. **LangChain 版本** (已有) - 使用 LangChain + LiteLLM +2. **MCP 版本** (新增) - 使用 Model Context Protocol +3. **A2A 版本** (新增) - 使用 Agent-to-Agent 框架 + +## 🆕 新增文件 + +### Agent 实现 + +| 文件 | 说明 | +|------|------| +| `azure_blob_agent_mcp.py` | MCP 框架版本的 Agent 实现 | +| `azure_blob_agent_a2a.py` | A2A 框架版本的 Agent 实现 | + +### Docker 相关 + +| 文件 | 说明 | +|------|------| +| `azure_blob_agent_mcp.Dockerfile` | MCP 版本的 Dockerfile | +| `azure_blob_agent_a2a.Dockerfile` | A2A 版本的 Dockerfile | +| `requirements_mcp.txt` | MCP 版本的依赖 | +| `requirements_a2a.txt` | A2A 版本的依赖 | +| `build_azure_blob_mcp.sh` | MCP 版本构建脚本 | +| `build_azure_blob_a2a.sh` | A2A 版本构建脚本 | + +### 文档和测试 + +| 文件 | 说明 | +|------|------| +| `MULTI_FRAMEWORK_GUIDE.md` | 多框架使用指南 | +| `test_multi_framework.sh` | 多框架集成测试脚本 | + +## 🔄 修改的文件 + +### 数据库层 + +**database.py** - 扩展了数据模型： + +#### Template 模型新增字段： +- `agent_framework` - Agent 框架类型 (langchain/mcp/a2a) +- `tools_config` - 工具配置 JSON +- `default_model_provider` - 默认模型提供商 +- `default_model_name` - 默认模型名称 + +#### Agent 模型新增字段： +- `agent_framework` - Agent 框架类型 +- `tools_config` - 工具配置 +- `tool_endpoint` - 工具端点 URL +- `tool_api_key` - 工具 API 密钥 +- `model_provider` - 模型提供商 +- `model_name` - 模型名称 +- `model_endpoint` - 模型端点 +- `model_api_key` - 模型 API 密钥 +- `storage_connection_string` - 存储连接字符串 +- `storage_account_name` - 存储账户名称 + +### API 层 + +**app.py** - 扩展了请求模型： + +#### CreateTemplateRequest 新增字段： +```python +agent_framework: str = "langchain" +tools_config: Optional[Dict] = {} +default_model_provider: Optional[str] = None +default_model_name: Optional[str] = None +``` + +#### CreatePlatformAgentRequest 新增字段： +```python +namespace: Optional[str] = "ai-agents" +agent_framework: Optional[str] = None +tools_config: Optional[Dict] = {} +tool_endpoint: Optional[str] = None +tool_api_key: Optional[str] = None +model_provider: Optional[str] = None +model_name: Optional[str] = None +model_endpoint: Optional[str] = None +model_api_key: Optional[str] = None +storage_connection_string: Optional[str] = None +storage_account_name: Optional[str] = None +``` + +#### CreateCustomAgentRequest 同样新增了上述字段 + +### Kubernetes 层 + +**k8s_manager.py** - 扩展了部署逻辑： + +#### _generate_pod_manifest 方法更新： +- 支持传递框架类型到容器环境变量 +- 支持传递工具配置 (tools_config, tool_endpoint, tool_api_key) +- 支持传递模型配置 (model_provider, model_name, model_endpoint, model_api_key) +- 支持传递存储配置 (storage_connection_string, storage_account_name) +- 支持传递用户标识 (user_id, tenant_id) +- 支持自定义命名空间 + +#### 新增镜像映射： +```python +"azure_blob_agent_mcp": "agnettaiji.azurecr.io/ai-agents/azure-blob-agent-mcp:latest" +"azure_blob_agent_a2a": "agnettaiji.azurecr.io/ai-agents/azure-blob-agent-a2a:latest" +``` + +#### 新增端口映射： +```python +"azure_blob_agent_mcp": 8080 +"azure_blob_agent_a2a": 8080 +``` + +#### 新增环境变量说明（用于文档） + +## 🏗️ 架构设计 + +### 参数传递流程 + +``` +用户请求 (API) + ↓ +app.py (API 层) + ├─ 验证参数 + ├─ 保存到数据库 (database.py) + └─ 调用 K8sManager + ↓ +k8s_manager.py (K8s 层) + ├─ 构建环境变量 + │ ├─ AGENT_FRAMEWORK + │ ├─ TOOLS_CONFIG + │ ├─ MODEL_* + │ ├─ STORAGE_* + │ └─ USER_ID, TENANT_ID, NAMESPACE + ├─ 创建 Pod/Deployment + └─ 传递到容器 + ↓ +Agent 容器 (azure_blob_agent_*.py) + ├─ 读取环境变量 + ├─ 初始化框架 + ├─ 配置工具 + ├─ 连接存储 + └─ 提供 API 服务 +``` + +### 框架特性对比 + +| 特性 | LangChain | MCP | A2A | +|------|-----------|-----|-----| +| **实现文件** | azure_blob_agent.py | azure_blob_agent_mcp.py | azure_blob_agent_a2a.py | +| **工具定义** | LangChain Tools | MCP Tool Classes | A2A Action Handlers | +| **API 端点** | /query | /mcp/tools, /mcp/call | /a2a/capabilities, /a2a/message | +| **协作能力** | ❌ | ❌ | ✅ Agent 注册和通信 | +| **工具发现** | 内置 | GET /mcp/tools | GET /a2a/capabilities | +| **消息格式** | 自然语言 | MCP Protocol | A2A Message Protocol | +| **依赖** | langchain, litellm | fastapi, pydantic | fastapi, httpx | + +## 📝 数据库迁移 + +提供了迁移脚本 `migrate_multi_framework.py`： + +```bash +python migrate_multi_framework.py +``` + +支持： +- ✅ SQLite (开发环境) +- ✅ PostgreSQL (生产环境) +- ✅ 自动检测已存在字段 +- ✅ 验证迁移结果 + +## 🚀 部署流程 + +### 1. 构建镜像 + +```bash +cd agent_templates + +# 构建 MCP 版本 +./build_azure_blob_mcp.sh + +# 构建 A2A 版本 +./build_azure_blob_a2a.sh +``` + +### 2. 运行数据库迁移 + +```bash +python migrate_multi_framework.py +``` + +### 3. 创建 Agent + +```bash +# 创建 MCP Agent +curl -X POST http://agent-manager:8000/v2/agents/platform \ + -H "Content-Type: application/json" \ + -d @mcp_agent_config.json + +# 创建 A2A Agent +curl -X POST http://agent-manager:8000/v2/agents/platform \ + -H "Content-Type: application/json" \ + -d @a2a_agent_config.json +``` + +### 4. 测试 + +```bash +./test_multi_framework.sh +``` + +## 🔧 环境变量配置示例 + +### MCP Agent + +```bash +# 框架配置 +AGENT_FRAMEWORK=mcp +TEMPLATE_TYPE=azure_blob_agent_mcp + +# 工具配置 +TOOLS_CONFIG='{"enabled_tools": ["list_containers", "list_blobs"]}' + +# 模型配置 +MODEL_PROVIDER=openai +MODEL_NAME=gpt-4 +MODEL_API_KEY=sk-xxxx +MODEL_ENDPOINT=https://api.openai.com/v1 + +# 存储配置 +AZURE_STORAGE_CONNECTION_STRING=DefaultEndpointsProtocol=https;... +STORAGE_ACCOUNT_NAME=myaccount + +# 用户信息 +USER_ID=user123 +TENANT_ID=tenant456 +NAMESPACE=ai-agents +``` + +### A2A Agent + +```bash +# 框架配置 +AGENT_FRAMEWORK=a2a +TEMPLATE_TYPE=azure_blob_agent_a2a + +# Agent 身份 +AGENT_ID=blob-agent-001 +AGENT_ROLE=storage_manager +AGENT_CAPABILITIES='["blob_storage", "file_operations"]' + +# 模型配置 +MODEL_PROVIDER=azure-openai +MODEL_NAME=gpt-4 +MODEL_API_KEY=xxxx +MODEL_ENDPOINT=https://myopenai.openai.azure.com + +# 存储配置 +AZURE_STORAGE_CONNECTION_STRING=DefaultEndpointsProtocol=https;... + +# 用户信息 +USER_ID=user123 +TENANT_ID=tenant456 +NAMESPACE=ai-agents +``` + +## 🎯 使用场景 + +### LangChain 版本 +- ✅ 复杂的推理任务 +- ✅ 多步骤文件处理 +- ✅ 与现有 LangChain 应用集成 + +### MCP 版本 +- ✅ 标准化工具调用 +- ✅ 跨平台工具共享 +- ✅ 轻量级集成 + +### A2A 版本 +- ✅ 多 Agent 协作 +- ✅ 分布式任务处理 +- ✅ Agent 间通信 + +## 📚 API 端点对比 + +### LangChain +- `POST /query` - 自然语言查询 +- `GET /health` - 健康检查 +- `POST /connect` - 连接存储 + +### MCP +- `GET /mcp/tools` - 列出可用工具 +- `POST /mcp/call` - 调用工具 +- `POST /query` - 查询（简化版） +- `GET /health` - 健康检查 +- `POST /connect` - 连接存储 + +### A2A +- `GET /a2a/capabilities` - 获取能力 +- `POST /a2a/register` - 注册其他 Agent +- `GET /a2a/agents` - 列出已注册 Agent +- `POST /a2a/message` - 处理 A2A 消息 +- `POST /a2a/collaborate` - 与其他 Agent 协作 +- `POST /query` - 查询 +- `GET /health` - 健康检查 +- `POST /connect` - 连接存储 + +## ✅ 测试清单 + +- [ ] 数据库迁移成功 +- [ ] MCP 镜像构建成功 +- [ ] A2A 镜像构建成功 +- [ ] MCP Agent 创建成功 +- [ ] A2A Agent 创建成功 +- [ ] MCP 工具调用正常 +- [ ] A2A 消息处理正常 +- [ ] 健康检查通过 +- [ ] 存储连接正常 +- [ ] 环境变量正确传递 + +## 🐛 已知问题 + +1. **LLM 集成**: MCP 和 A2A 版本目前使用简单规则匹配，需要集成实际 LLM 进行意图识别 +2. **安全性**: API 密钥等敏感信息应加密存储 +3. **日志**: 需要统一的日志收集和监控 + +## 🔮 未来改进 + +1. **安全增强** + - 密钥加密存储 + - RBAC 权限控制 + - API 密钥轮换 + +2. **功能扩展** + - 更多 Azure 服务集成 + - 自定义工具注册 + - 工具组合和编排 + +3. **监控和调试** + - 分布式追踪 + - 性能监控 + - 调试工具 + +4. **开发体验** + - Web UI 管理界面 + - 可视化工具设计器 + - Agent 模板市场 + +## 📖 相关文档 + +- [多框架使用指南](agent_templates/MULTI_FRAMEWORK_GUIDE.md) +- [API 文档](API_DOCUMENTATION.md) +- [Azure Blob Agent 原始文档](agent_templates/AZURE_BLOB_AGENT_USAGE.md) diff --git a/ZOMBIE_PROCESS_FIX.md b/ZOMBIE_PROCESS_FIX.md new file mode 100644 index 0000000..b70bb9a --- /dev/null +++ b/ZOMBIE_PROCESS_FIX.md @@ -0,0 +1,126 @@ +# 僵尸进程和 CPU 100% 问题修复方案 + +## 问题诊断 + +**容器**: taiji-mcp-server (ID: 64d363729ff9) +**进程**: PID 3411601, CPU 100% +**根因**: Docker 健康检查导致的僵尸进程泄漏 (800+ defunct curl 进程) + +## 立即修复步骤 + +### 方案 1: 重启容器（最快） + +```bash +# 重启容器，清理僵尸进程 +docker restart taiji-mcp-server + +# 检查状态 +docker ps | grep taiji-mcp-server +``` + +### 方案 2: 临时禁用健康检查 + +```bash +# 停止容器 +docker stop taiji-mcp-server + +# 使用 --no-healthcheck 重新启动 +docker run -d --name taiji-mcp-server-temp \ + --no-healthcheck \ + -p 8002:8000 \ + taiji-ai-pad-mcp-server + +# 或修改 docker-compose.yml，注释掉 healthcheck +``` + +## 长期修复方案 + +### 方案 A: 使用 Python 内置健康检查（推荐） + +不依赖外部 curl 命令，避免子进程问题： + +**Dockerfile 修改**: +```dockerfile +HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \ + CMD python3 -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health').read()" || exit 1 +``` + +### 方案 B: 使用 tini 或 dumb-init（推荐） + +正确处理子进程回收： + +**Dockerfile 修改**: +```dockerfile +# 安装 tini +RUN apt-get update && apt-get install -y tini + +# 使用 tini 作为 init 进程 +ENTRYPOINT ["/usr/bin/tini", "--"] +CMD ["python3", "-m", "uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"] + +# 健康检查保持不变 +HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \ + CMD curl -f http://localhost:8000/health || exit 1 +``` + +### 方案 C: 修改健康检查端点，减少数据库连接 + +**main.py 修改** (假设你有 `/health` 端点): +```python +@app.get("/health") +async def health_check(): + """轻量级健康检查，不连接数据库""" + return {"status": "healthy", "timestamp": datetime.now().isoformat()} + +@app.get("/health/deep") +async def deep_health_check(): + """深度健康检查，包含数据库连接测试""" + try: + # 测试数据库连接 + db = next(get_db()) + db.execute(text("SELECT 1")) + return {"status": "healthy", "database": "connected"} + except Exception as e: + raise HTTPException(status_code=503, detail=f"Unhealthy: {str(e)}") +``` + +### 方案 D: 调整健康检查频率 + +如果服务稳定，可以降低检查频率： + +```dockerfile +HEALTHCHECK --interval=60s --timeout=10s --start-period=40s --retries=3 \ + CMD curl -f http://localhost:8000/health || exit 1 +``` + +## 验证修复 + +```bash +# 1. 检查容器健康状态 +docker ps | grep taiji-mcp-server + +# 2. 检查僵尸进程数量 +docker exec taiji-mcp-server ps aux | grep defunct | wc -l + +# 3. 检查 CPU 占用 +docker stats --no-stream taiji-mcp-server + +# 4. 检查日志 +docker logs --tail 100 taiji-mcp-server +``` + +## 监控建议 + +```bash +# 定期检查僵尸进程 +watch -n 5 'docker exec taiji-mcp-server ps aux | grep defunct | wc -l' + +# 监控资源使用 +docker stats taiji-mcp-server +``` + +## 参考资料 + +- Docker 僵尸进程问题: https://blog.phusion.nl/2015/01/20/docker-and-the-pid-1-zombie-reaping-problem/ +- tini 项目: https://github.com/krallin/tini +- dumb-init: https://github.com/Yelp/dumb-init diff --git a/agent_templates/AZURE_BLOB_AGENT_USAGE.md b/agent_templates/AZURE_BLOB_AGENT_USAGE.md new file mode 100644 index 0000000..1a5f993 --- /dev/null +++ b/agent_templates/AZURE_BLOB_AGENT_USAGE.md @@ -0,0 +1,364 @@ +# Azure Blob Storage AI Agent 使用指南 + +## 概述 + +这是一个基于 LangChain + LiteLLM 的智能 Azure Blob Storage 管理代理，支持： +- 通过 API 动态接收 Azure Storage 连接字符串 +- 使用自然语言查询和管理存储 +- 通过环境变量配置 LLM 模型 + +## 架构说明 + +``` +┌─────────────┐ HTTP API ┌──────────────────┐ Azure SDK ┌─────────────────┐ +│ 客户端 │ ──────────────> │ FastAPI Server │ ──────────────> │ Azure Blob │ +│ │ │ + LangChain │ │ Storage │ +└─────────────┘ │ + LiteLLM │ └─────────────────┘ + └──────────────────┘ + │ + ▼ + ┌──────────────────┐ + │ LiteLLM Server │ + │ (4000端口) │ + └──────────────────┘ +``` + +## 快速开始 + +### 1. 构建镜像 + +```bash +cd /home/taiji/tools/agent-manager/agent_templates + +# 构建镜像 +./build_azure_blob_agent.sh latest + +# 或者手动构建 +docker build -f azure_blob_agent.Dockerfile -t azure-blob-agent:latest . +``` + +### 2. 启动 LiteLLM 服务（如果还没启动） + +确保你的 LiteLLM 服务正在运行，例如： +```bash +# 检查 LiteLLM 是否运行 +curl http://localhost:4000/health + +# 如果没运行，启动它 +docker run -d --name litellm \ + -p 4000:4000 \ + -e OPENAI_API_KEY=your_key \ + ghcr.io/berriai/litellm:latest +``` + +### 3. 启动 Azure Blob Agent + +```bash +docker run -d --name azure-blob-agent \ + -p 8080:8080 \ + -e LITELLM_API_BASE=http://host.docker.internal:4000 \ + -e LITELLM_MODEL=gpt-3.5-turbo \ + -e LITELLM_API_KEY=sk-1234 \ + azure-blob-agent:latest +``` + +**环境变量说明：** +- `LITELLM_API_BASE`: LiteLLM 服务地址 +- `LITELLM_MODEL`: 使用的模型名称 +- `LITELLM_API_KEY`: LiteLLM API 密钥 +- `SERVICE_HOST`: 服务监听地址（默认 0.0.0.0） +- `SERVICE_PORT`: 服务监听端口（默认 8080） + +## API 使用 + +### 1. 健康检查 + +```bash +curl http://localhost:8080/health +``` + +**响应示例：** +```json +{ + "status": "healthy", + "connected": true, + "connection_info": { + "account_kind": "StorageV2", + "sku_name": "Standard_LRS", + "connected_at": "2026-01-08T20:00:00" + } +} +``` + +### 2. 连接到 Azure Storage + +```bash +curl -X POST http://localhost:8080/connect \ + -H 'Content-Type: application/json' \ + -d '{ + "connection_string": "DefaultEndpointsProtocol=https;AccountName=yourname;AccountKey=yourkey;EndpointSuffix=core.windows.net" + }' +``` + +**响应示例：** +```json +{ + "status": "connected", + "message": "成功连接到Azure Blob Storage", + "account_info": { + "account_kind": "StorageV2", + "sku_name": "Standard_LRS" + } +} +``` + +### 3. 自然语言查询 + +#### 列出所有容器 +```bash +curl -X POST http://localhost:8080/query \ + -H 'Content-Type: application/json' \ + -d '{ + "query": "列出所有容器" + }' +``` + +#### 查看容器中的文件 +```bash +curl -X POST http://localhost:8080/query \ + -H 'Content-Type: application/json' \ + -d '{ + "query": "显示 images 容器中的所有文件" + }' +``` + +#### 搜索文件 +```bash +curl -X POST http://localhost:8080/query \ + -H 'Content-Type: application/json' \ + -d '{ + "query": "在 documents 容器中搜索包含 report 的文件" + }' +``` + +#### 获取存储统计 +```bash +curl -X POST http://localhost:8080/query \ + -H 'Content-Type: application/json' \ + -d '{ + "query": "显示存储统计信息" + }' +``` + +#### 获取文件详细信息 +```bash +curl -X POST http://localhost:8080/query \ + -H 'Content-Type: application/json' \ + -d '{ + "query": "获取 images 容器中 logo.png 的详细信息" + }' +``` + +**响应示例：** +```json +{ + "status": "success", + "query": "列出所有容器", + "answer": "当前有3个容器:\n1. images (最后修改: 2026-01-08)\n2. documents (最后修改: 2026-01-07)\n3. backups (最后修改: 2026-01-06)", + "intermediate_steps": "..." +} +``` + +## 在 Kubernetes 中部署 + +### 方法 1: 使用 agent-manager API + +```bash +# 1. 首先确保模板已添加到 k8s_manager.py +# 2. 创建 agent +curl -X POST http://localhost:8000/agents \ + -H 'Content-Type: application/json' \ + -d '{ + "name": "my-blob-agent", + "template": "azure_blob_agent", + "env": { + "LITELLM_API_BASE": "http://litellm-service:4000", + "LITELLM_MODEL": "gpt-3.5-turbo", + "LITELLM_API_KEY": "sk-1234" + } + }' + +# 3. 连接到存储 +curl -X POST http://my-blob-agent-ip:8080/connect \ + -H 'Content-Type: application/json' \ + -d '{ + "connection_string": "YOUR_CONNECTION_STRING" + }' +``` + +### 方法 2: 直接部署 YAML + +创建 `azure-blob-agent-deployment.yaml`: + +```yaml +apiVersion: apps/v1 +kind: Deployment +metadata: + name: azure-blob-agent + namespace: ai-agents +spec: + replicas: 1 + selector: + matchLabels: + app: azure-blob-agent + template: + metadata: + labels: + app: azure-blob-agent + spec: + containers: + - name: azure-blob-agent + image: agnettaiji.azurecr.io/ai-agents/azure-blob-agent:latest + ports: + - containerPort: 8080 + env: + - name: LITELLM_API_BASE + value: "http://litellm-service:4000" + - name: LITELLM_MODEL + value: "gpt-3.5-turbo" + - name: LITELLM_API_KEY + valueFrom: + secretKeyRef: + name: litellm-secret + key: api-key + resources: + requests: + memory: "256Mi" + cpu: "250m" + limits: + memory: "512Mi" + cpu: "500m" +--- +apiVersion: v1 +kind: Service +metadata: + name: azure-blob-agent-service + namespace: ai-agents +spec: + selector: + app: azure-blob-agent + ports: + - port: 8080 + targetPort: 8080 + type: ClusterIP +``` + +部署： +```bash +kubectl apply -f azure-blob-agent-deployment.yaml +``` + +## 支持的查询示例 + +| 自然语言查询 | 功能 | +|------------|------| +| "列出所有容器" | 显示所有容器列表 | +| "显示 images 容器中的文件" | 列出指定容器的文件 | +| "在 documents 中搜索 report" | 搜索包含关键字的文件 | +| "获取 data/test.csv 的信息" | 显示文件详细信息 | +| "显示存储统计" | 显示整体存储使用情况 | +| "images 容器有多少文件" | 统计容器文件数 | +| "查找所有 .pdf 文件" | 按扩展名搜索 | + +## 故障排查 + +### 1. Agent 启动失败 + +```bash +# 检查日志 +docker logs azure-blob-agent + +# 常见问题： +# - LiteLLM 服务不可达：检查 LITELLM_API_BASE +# - 端口冲突：修改 SERVICE_PORT +``` + +### 2. 连接 Azure Storage 失败 + +```bash +# 检查连接字符串格式 +# 正确格式： +DefaultEndpointsProtocol=https;AccountName=myaccount;AccountKey=mykey==;EndpointSuffix=core.windows.net + +# 测试连接 +curl -X POST http://localhost:8080/connect \ + -H 'Content-Type: application/json' \ + -d '{"connection_string": "YOUR_STRING"}' +``` + +### 3. 查询返回错误 + +```bash +# 检查是否已连接 +curl http://localhost:8080/health + +# 查看详细日志 +docker logs -f azure-blob-agent +``` + +## 开发与扩展 + +### 添加新工具 + +在 `azure_blob_agent.py` 中添加新的工具函数： + +```python +def download_blob(container_name: str, blob_name: str) -> str: + """下载 blob 内容（示例）""" + # 实现下载逻辑 + pass + +# 在 create_blob_agent() 中添加工具 +tools.append( + Tool( + name="download_blob", + func=lambda input_str: download_blob(*input_str.split(",")), + description="下载指定的文件。输入格式: '容器名,文件名'" + ) +) +``` + +### 自定义模型 + +支持任何 LiteLLM 兼容的模型： + +```bash +# 使用 Claude +-e LITELLM_MODEL=claude-3-sonnet-20240229 + +# 使用本地模型 +-e LITELLM_MODEL=ollama/llama2 +-e LITELLM_API_BASE=http://localhost:11434 + +# 使用 Azure OpenAI +-e LITELLM_MODEL=azure/gpt-4 +``` + +## 性能优化 + +1. **连接池**: BlobServiceClient 会自动管理连接池 +2. **缓存**: 可以添加 Redis 缓存常用查询结果 +3. **并发**: 使用 `max_workers` 参数提高并发处理能力 + +## 安全建议 + +1. **连接字符串**: 不要在代码中硬编码，使用环境变量或 K8s Secrets +2. **访问控制**: 使用 SAS token 而非完整连接字符串 +3. **网络隔离**: 在 K8s 中使用 NetworkPolicy 限制访问 +4. **日志脱敏**: 避免记录敏感信息 + +## 更多资源 + +- [Azure Blob Storage Python SDK](https://learn.microsoft.com/azure/storage/blobs/storage-quickstart-blobs-python) +- [LangChain Documentation](https://python.langchain.com/docs/get_started/introduction) +- [LiteLLM Documentation](https://docs.litellm.ai/) diff --git a/agent_templates/MULTI_FRAMEWORK_GUIDE.md b/agent_templates/MULTI_FRAMEWORK_GUIDE.md new file mode 100644 index 0000000..0e43606 --- /dev/null +++ b/agent_templates/MULTI_FRAMEWORK_GUIDE.md @@ -0,0 +1,366 @@ +# Azure Blob Agent - 多框架支持使用指南 + +本文档介绍如何使用三种不同框架版本的 Azure Blob Storage AI Agent： +- **LangChain 版本**: 使用 LangChain + LiteLLM +- **MCP 版本**: 使用 Model Context Protocol +- **A2A 版本**: 使用 Agent-to-Agent 框架 + +## 📋 目录 + +1. [框架对比](#框架对比) +2. [部署配置](#部署配置) +3. [API 使用示例](#api-使用示例) +4. [创建 Agent 示例](#创建-agent-示例) + +## 🔍 框架对比 + +| 特性 | LangChain | MCP | A2A | +|------|-----------|-----|-----| +| 工具调用 | LangChain Tools | MCP Protocol | A2A Messages | +| Agent 协作 | ❌ | ❌ | ✅ | +| 结构化输出 | ✅ | ✅ | ✅ | +| 复杂推理 | ✅ | ⚡ 轻量 | ⚡ 轻量 | +| 适用场景 | 复杂任务链 | 标准化工具 | 多Agent协作 | + +## 🚀 部署配置 + +### 1. LangChain 版本 + +```json +{ + "name": "my-blob-agent", + "template_name": "azure_blob_agent", + "owner_id": "user123", + "namespace": "ai-agents", + "agent_framework": "langchain", + "environment_vars": { + "LITELLM_API_BASE": "http://litellm-service:4000", + "LITELLM_MODEL": "gpt-3.5-turbo", + "LITELLM_API_KEY": "sk-xxxx", + "AZURE_STORAGE_CONNECTION_STRING": "DefaultEndpointsProtocol=https;..." + } +} +``` + +### 2. MCP 版本 + +```json +{ + "name": "my-blob-agent-mcp", + "template_name": "azure_blob_agent_mcp", + "owner_id": "user123", + "namespace": "ai-agents", + "agent_framework": "mcp", + "model_provider": "openai", + "model_name": "gpt-4", + "model_api_key": "sk-xxxx", + "model_endpoint": "https://api.openai.com/v1", + "storage_connection_string": "DefaultEndpointsProtocol=https;...", + "tools_config": { + "enabled_tools": ["list_containers", "list_blobs", "search_blobs"] + } +} +``` + +### 3. A2A 版本 + +```json +{ + "name": "my-blob-agent-a2a", + "template_name": "azure_blob_agent_a2a", + "owner_id": "user123", + "namespace": "ai-agents", + "agent_framework": "a2a", + "model_provider": "openai", + "model_name": "gpt-4", + "model_api_key": "sk-xxxx", + "storage_connection_string": "DefaultEndpointsProtocol=https;...", + "environment_vars": { + "AGENT_ID": "blob-agent-001", + "AGENT_ROLE": "storage_manager", + "AGENT_CAPABILITIES": "[\"blob_storage\", \"file_operations\"]" + } +} +``` + +## 📡 API 使用示例 + +### MCP 版本 API + +#### 1. 列出所有可用工具 + +```bash +curl http:///mcp/tools +``` + +响应： +```json +{ + "tools": [ + { + "name": "list_containers", + "description": "列出 Azure Blob Storage 中的所有容器", + "inputSchema": { + "type": "object", + "properties": {}, + "required": [] + } + }, + { + "name": "list_blobs", + "description": "列出指定容器中的所有文件", + "inputSchema": { + "type": "object", + "properties": { + "container_name": { + "type": "string", + "description": "容器名称" + } + }, + "required": ["container_name"] + } + } + ] +} +``` + +#### 2. 调用 MCP 工具 + +```bash +curl -X POST http:///mcp/call \ + -H "Content-Type: application/json" \ + -d '{ + "tool_name": "list_containers", + "parameters": {} + }' +``` + +```bash +curl -X POST http:///mcp/call \ + -H "Content-Type: application/json" \ + -d '{ + "tool_name": "list_blobs", + "parameters": { + "container_name": "my-container" + } + }' +``` + +### A2A 版本 API + +#### 1. 获取 Agent 能力 + +```bash +curl http:///a2a/capabilities +``` + +响应： +```json +{ + "agent_id": "blob-agent-001", + "agent_role": "storage_manager", + "capabilities": ["blob_storage", "file_operations"], + "supported_actions": [ + "list_containers", + "list_blobs", + "get_blob_info", + "search_blobs", + "get_stats" + ] +} +``` + +#### 2. 注册其他 Agent + +```bash +curl -X POST http:///a2a/register \ + -H "Content-Type: application/json" \ + -d '{ + "agent_id": "analytics-agent", + "agent_role": "data_analyzer", + "capabilities": ["data_analysis", "visualization"], + "endpoint": "http://analytics-agent:8080" + }' +``` + +#### 3. 发送 A2A 消息 + +```bash +curl -X POST http:///a2a/message \ + -H "Content-Type: application/json" \ + -d '{ + "message_id": "msg-001", + "from_agent": "external-agent", + "to_agent": "blob-agent-001", + "message_type": "request", + "action": "list_containers", + "parameters": {} + }' +``` + +#### 4. Agent 间协作 + +```bash +curl -X POST http:///a2a/collaborate \ + -H "Content-Type: application/json" \ + -d '{ + "target_agent_id": "analytics-agent", + "action": "analyze_data", + "parameters": { + "data_source": "blob_storage" + } + }' +``` + +## 🛠️ 创建 Agent 示例 + +### 使用 Agent Manager API 创建 + +#### 1. 创建 MCP Agent + +```bash +curl -X POST http://agent-manager:8000/v2/agents/platform \ + -H "Content-Type: application/json" \ + -d '{ + "name": "blob-mcp-001", + "template_name": "azure_blob_agent_mcp", + "owner_id": "user123", + "namespace": "ai-agents", + "agent_framework": "mcp", + "model_provider": "openai", + "model_name": "gpt-4", + "model_api_key": "sk-xxxx", + "storage_connection_string": "DefaultEndpointsProtocol=https;AccountName=myaccount;AccountKey=xxx;EndpointSuffix=core.windows.net", + "tools_config": { + "max_iterations": 5, + "timeout": 30 + } + }' +``` + +#### 2. 创建 A2A Agent + +```bash +curl -X POST http://agent-manager:8000/v2/agents/platform \ + -H "Content-Type: application/json" \ + -d '{ + "name": "blob-a2a-001", + "template_name": "azure_blob_agent_a2a", + "owner_id": "user123", + "namespace": "ai-agents", + "agent_framework": "a2a", + "model_provider": "azure-openai", + "model_name": "gpt-4", + "model_endpoint": "https://myopenai.openai.azure.com", + "model_api_key": "xxxx", + "storage_connection_string": "DefaultEndpointsProtocol=https;...", + "query_params": { + "agent_id": "blob-a2a-001", + "agent_role": "storage_manager", + "agent_capabilities": ["blob_storage", "file_operations"] + } + }' +``` + +## 🔧 参数说明 + +### 通用参数（所有框架） + +| 参数 | 类型 | 必需 | 说明 | +|------|------|------|------| +| `name` | string | ✅ | Agent 名称（唯一） | +| `template_name` | string | ✅ | 模板名称 | +| `owner_id` | string | ✅ | 所有者ID | +| `namespace` | string | ❌ | K8s 命名空间，默认 `ai-agents` | +| `agent_framework` | string | ❌ | 框架类型: `langchain`, `mcp`, `a2a` | +| `storage_connection_string` | string | ❌ | Azure Storage 连接字符串 | +| `storage_account_name` | string | ❌ | 存储账户名称 | + +### 模型配置参数（MCP/A2A） + +| 参数 | 类型 | 必需 | 说明 | +|------|------|------|------| +| `model_provider` | string | ✅ | 模型提供商: `openai`, `azure-openai` | +| `model_name` | string | ✅ | 模型名称: `gpt-4`, `gpt-3.5-turbo` | +| `model_api_key` | string | ✅ | 模型 API 密钥 | +| `model_endpoint` | string | ❌ | 模型 API 端点 | + +### 工具配置参数（MCP/A2A） + +| 参数 | 类型 | 必需 | 说明 | +|------|------|------|------| +| `tools_config` | object | ❌ | 工具配置 JSON | +| `tool_endpoint` | string | ❌ | 外部工具端点 | +| `tool_api_key` | string | ❌ | 工具 API 密钥 | + +### 资源配置参数 + +| 参数 | 类型 | 必需 | 说明 | +|------|------|------|------| +| `cpu_request` | string | ❌ | CPU 请求，如 `100m` | +| `cpu_limit` | string | ❌ | CPU 限制，如 `500m` | +| `memory_request` | string | ❌ | 内存请求，如 `128Mi` | +| `memory_limit` | string | ❌ | 内存限制，如 `512Mi` | + +## 🎯 使用场景 + +### LangChain 版本适用于： +- 需要复杂推理链的任务 +- 多步骤文件处理流程 +- 集成现有 LangChain 生态系统 + +### MCP 版本适用于： +- 标准化工具调用 +- 轻量级集成 +- 跨平台工具共享 + +### A2A 版本适用于： +- 多 Agent 协作场景 +- 分布式任务处理 +- Agent 间通信需求 + +## 📝 数据库迁移 + +如果从旧版本升级，需要运行数据库迁移： + +```sql +-- 添加新字段到 templates 表 +ALTER TABLE templates ADD COLUMN agent_framework VARCHAR(50) DEFAULT 'langchain'; +ALTER TABLE templates ADD COLUMN tools_config JSON; +ALTER TABLE templates ADD COLUMN default_model_provider VARCHAR(100); +ALTER TABLE templates ADD COLUMN default_model_name VARCHAR(200); + +-- 添加新字段到 agents 表 +ALTER TABLE agents ADD COLUMN agent_framework VARCHAR(50) DEFAULT 'langchain'; +ALTER TABLE agents ADD COLUMN tools_config JSON; +ALTER TABLE agents ADD COLUMN tool_endpoint VARCHAR(500); +ALTER TABLE agents ADD COLUMN tool_api_key VARCHAR(500); +ALTER TABLE agents ADD COLUMN model_provider VARCHAR(100); +ALTER TABLE agents ADD COLUMN model_name VARCHAR(200); +ALTER TABLE agents ADD COLUMN model_endpoint VARCHAR(500); +ALTER TABLE agents ADD COLUMN model_api_key VARCHAR(500); +ALTER TABLE agents ADD COLUMN storage_connection_string VARCHAR(1000); +ALTER TABLE agents ADD COLUMN storage_account_name VARCHAR(200); +``` + +## 🐛 故障排查 + +### 问题: MCP 工具调用失败 + +**解决方案**: +1. 检查工具名称是否正确 +2. 验证参数格式 +3. 查看日志: `kubectl logs -n ai-agents` + +### 问题: A2A Agent 无法注册 + +**解决方案**: +1. 确认目标 Agent 可访问 +2. 检查网络策略 +3. 验证 endpoint URL 格式 + +## 📚 更多资源 + +- [LangChain 文档](https://python.langchain.com/) +- [MCP 协议规范](https://modelcontextprotocol.io/) +- [Agent Manager API 文档](../API_DOCUMENTATION.md) diff --git a/agent_templates/QUICKSTART.md b/agent_templates/QUICKSTART.md new file mode 100644 index 0000000..5079037 --- /dev/null +++ b/agent_templates/QUICKSTART.md @@ -0,0 +1,157 @@ +# 🚀 Azure Blob Storage Agent 快速启动 + +## 一键启动命令 + +### 1. 构建镜像 +```bash +cd /home/taiji/tools/agent-manager/agent_templates +./build_azure_blob_agent.sh latest +``` + +### 2. 启动 Agent（本地测试） +```bash +# 方式 A: 启动时提供连接字符串（推荐） +docker run -d --name azure-blob-agent \ + -p 8080:8080 \ + -e LITELLM_API_BASE=http://20.2.70.108:4000 \ + -e LITELLM_MODEL=gpt-3.5-turbo \ + -e LITELLM_API_KEY=sk-1234 \ + -e AZURE_STORAGE_CONNECTION_STRING="DefaultEndpointsProtocol=https;AccountName=xxx;AccountKey=xxx;EndpointSuffix=core.windows.net" \ + azure-blob-agent:latest + +# 方式 B: 稍后通过 API 连接 +docker run -d --name azure-blob-agent \ + -p 8080:8080 \ + -e LITELLM_API_BASE=http://20.2.70.108:4000 \ + -e LITELLM_MODEL=gpt-3.5-turbo \ + -e LITELLM_API_KEY=sk-1234 \ + azure-blob-agent:latest + +# 然后调用 /connect API 连接 + +# 方式 C: 如果 LiteLLM 在另一个容器中 +docker run -d --name azure-blob-agent \ + --network host \ + -e LITELLM_API_BASE=http://localhost:4000 \ + -e LITELLM_MODEL=gpt-3.5-turbo \ + -e LITELLM_API_KEY=sk-1234 \ + -e AZURE_STORAGE_CONNECTION_STRING="YOUR_CONNECTION_STRING" \ + azure-blob-agent:latest +``` + +### 3. 测试 Agent + +#### 方法 1: 使用 Bash 测试脚本 +```bash +./test_azure_blob_agent.sh +``` + +#### 方法 2: 使用 Python 客户端 +```bash +# 设置连接字符串（可选） +export AZURE_STORAGE_CONNECTION_STRING="DefaultEndpointsProtocol=https;AccountName=xxx;..." + +# 运行客户端 +python3 test_client.py +``` + +#### 方法 3: 使用 curl 手动测试 +```bash +# 健康检查 +curl http://localhost:8080/health + +# 连接到 Azure Storage +curl -X POST http://localhost:8080/connect \ + -H 'Content-Type: application/json' \ + -d '{ + "connection_string": "DefaultEndpointsProtocol=https;AccountName=myaccount;AccountKey=mykey;EndpointSuffix=core.windows.net" + }' + +# 执行查询 +curl -X POST http://localhost:8080/query \ + -H 'Content-Type: application/json' \ + -d '{"query": "列出所有容器"}' +``` + +## 推送到 ACR + +```bash +# 登录 ACR +az acr login --name agnettaiji + +# 推送镜像 +docker tag azure-blob-agent:latest agnettaiji.azurecr.io/ai-agents/azure-blob-agent:latest +docker push agnettaiji.azurecr.io/ai-agents/azure-blob-agent:latest +``` + +## 在 K8s 中部署 + +### 使用 agent-manager + +```bash +# 添加到 k8s_manager.py 的 image_map +"azure_blob_agent": "agnettaiji.azurecr.io/ai-agents/azure-blob-agent:latest" + +# 创建 agent +curl -X POST http://localhost:8000/agents \ + -H 'Content-Type: application/json' \ + -d '{ + "name": "my-blob-agent", + "template": "azure_blob_agent", + "env": { + "LITELLM_API_BASE": "http://litellm-service:4000", + "LITELLM_MODEL": "gpt-3.5-turbo", + "LITELLM_API_KEY": "sk-1234" + } + }' +``` + +## 常见问题 + +### Q: 容器启动失败 +```bash +# 查看日志 +docker logs azure-blob-agent + +# 检查 LiteLLM 是否可达 +docker exec azure-blob-agent curl http://host.docker.internal:4000/health +``` + +### Q: 无法连接到 Azure Storage +```bash +# 验证连接字符串格式 +# 正确格式包含: AccountName, AccountKey, EndpointSuffix + +# 测试连接 +curl -X POST http://localhost:8080/connect \ + -H 'Content-Type: application/json' \ + -d '{"connection_string": "YOUR_STRING"}' -v +``` + +### Q: 查询没有响应 +```bash +# 检查是否已连接 +curl http://localhost:8080/health | jq . + +# 查看详细日志 +docker logs -f azure-blob-agent +``` + +## 文件清单 + +``` +agent_templates/ +├── azure_blob_agent.py # 主程序 +├── azure_blob_agent.Dockerfile # Docker 镜像 +├── build_azure_blob_agent.sh # 构建脚本 +├── test_azure_blob_agent.sh # Bash 测试脚本 +├── test_client.py # Python 客户端 +├── AZURE_BLOB_AGENT_USAGE.md # 详细使用文档 +└── QUICKSTART.md # 本文件 +``` + +## 下一步 + +- 阅读 [详细使用文档](AZURE_BLOB_AGENT_USAGE.md) +- 查看 [agent_templates README](../README.md) +- 集成到你的应用中 diff --git a/agent_templates/QUICK_REFERENCE.md b/agent_templates/QUICK_REFERENCE.md new file mode 100644 index 0000000..74596dd --- /dev/null +++ b/agent_templates/QUICK_REFERENCE.md @@ -0,0 +1,199 @@ +# Azure Blob Agent - 快速参考 + +## 🚀 快速开始 + +### 1. 选择框架 + +| 框架 | 使用场景 | 文件 | +|------|---------|------| +| **LangChain** | 复杂推理任务 | `azure_blob_agent.py` | +| **MCP** | 标准化工具调用 | `azure_blob_agent_mcp.py` | +| **A2A** | 多 Agent 协作 | `azure_blob_agent_a2a.py` | + +### 2. 创建 Agent (curl) + +#### MCP 版本 +```bash +curl -X POST http://agent-manager:8000/v2/agents/platform \ + -H "Content-Type: application/json" \ + -d '{ + "name": "my-blob-mcp", + "template_name": "azure_blob_agent_mcp", + "owner_id": "user123", + "agent_framework": "mcp", + "model_provider": "openai", + "model_name": "gpt-4", + "model_api_key": "sk-xxxx", + "storage_connection_string": "DefaultEndpointsProtocol=https;..." + }' +``` + +#### A2A 版本 +```bash +curl -X POST http://agent-manager:8000/v2/agents/platform \ + -H "Content-Type: application/json" \ + -d '{ + "name": "my-blob-a2a", + "template_name": "azure_blob_agent_a2a", + "owner_id": "user123", + "agent_framework": "a2a", + "model_provider": "openai", + "model_name": "gpt-4", + "model_api_key": "sk-xxxx", + "storage_connection_string": "DefaultEndpointsProtocol=https;...", + "query_params": { + "agent_id": "my-blob-a2a", + "agent_role": "storage_manager" + } + }' +``` + +### 3. 使用 Agent + +#### MCP - 列出工具 +```bash +curl http:///mcp/tools +``` + +#### MCP - 调用工具 +```bash +curl -X POST http:///mcp/call \ + -H "Content-Type: application/json" \ + -d '{"tool_name": "list_containers", "parameters": {}}' +``` + +#### A2A - 获取能力 +```bash +curl http:///a2a/capabilities +``` + +#### A2A - 发送消息 +```bash +curl -X POST http:///a2a/message \ + -H "Content-Type: application/json" \ + -d '{ + "message_id": "msg-001", + "from_agent": "caller", + "to_agent": "my-blob-a2a", + "message_type": "request", + "action": "list_containers", + "parameters": {} + }' +``` + +## 🔧 必需参数 + +### MCP Agent +- ✅ `model_provider` - 模型提供商 +- ✅ `model_name` - 模型名称 +- ✅ `model_api_key` - API 密钥 + +### A2A Agent +- ✅ `model_provider` - 模型提供商 +- ✅ `model_name` - 模型名称 +- ✅ `model_api_key` - API 密钥 +- ✅ `query_params.agent_id` - Agent ID +- ✅ `query_params.agent_role` - Agent 角色 + +## 🛠️ 可选参数 + +| 参数 | 说明 | 示例 | +|------|------|------| +| `namespace` | K8s 命名空间 | `"ai-agents"` | +| `tools_config` | 工具配置 | `{"max_iterations": 5}` | +| `tool_endpoint` | 外部工具端点 | `"http://tools-api:8080"` | +| `model_endpoint` | 模型端点 | `"https://api.openai.com/v1"` | +| `storage_account_name` | 存储账户名 | `"myaccount"` | +| `cpu_request` | CPU 请求 | `"100m"` | +| `memory_request` | 内存请求 | `"256Mi"` | + +## 📊 环境变量 (容器内) + +### 框架相关 +- `AGENT_FRAMEWORK` - 框架类型 +- `TEMPLATE_TYPE` - 模板类型 + +### 模型相关 +- `MODEL_PROVIDER` - 模型提供商 +- `MODEL_NAME` - 模型名称 +- `MODEL_API_KEY` - API 密钥 +- `MODEL_ENDPOINT` - 端点 URL + +### 工具相关 +- `TOOLS_CONFIG` - 工具配置 JSON +- `TOOL_ENDPOINT` - 工具端点 +- `TOOL_API_KEY` - 工具密钥 + +### 存储相关 +- `AZURE_STORAGE_CONNECTION_STRING` - 连接字符串 +- `STORAGE_ACCOUNT_NAME` - 账户名 + +### 用户相关 +- `USER_ID` - 用户标识 +- `TENANT_ID` - 租户标识 +- `NAMESPACE` - 命名空间 + +## 🔍 故障排查 + +### Agent 启动失败 +```bash +# 查看日志 +kubectl logs -n ai-agents + +# 查看事件 +kubectl describe pod -n ai-agents +``` + +### 工具调用失败 +```bash +# 检查工具列表 +curl http:///mcp/tools + +# 测试健康检查 +curl http:///health +``` + +### 存储连接失败 +```bash +# 验证连接字符串 +curl -X POST http:///connect \ + -H "Content-Type: application/json" \ + -d '{"connection_string": "DefaultEndpointsProtocol=https;..."}' +``` + +## 📝 工具列表 + +### 共同工具（所有版本） +1. `list_containers` - 列出所有容器 +2. `list_blobs` - 列出容器中的文件 +3. `get_blob_info` - 获取文件详情 +4. `search_blobs` - 搜索文件 +5. `get_storage_stats` - 获取统计信息 + +## 🏗️ 构建镜像 + +```bash +cd agent_templates + +# MCP 版本 +./build_azure_blob_mcp.sh + +# A2A 版本 +./build_azure_blob_a2a.sh +``` + +## 🧪 测试 + +```bash +# 设置环境变量 +export AZURE_STORAGE_CONNECTION_STRING="DefaultEndpointsProtocol=https;..." +export OPENAI_API_KEY="sk-xxxx" + +# 运行测试 +./test_multi_framework.sh +``` + +## 📚 更多文档 + +- 详细指南: [MULTI_FRAMEWORK_GUIDE.md](MULTI_FRAMEWORK_GUIDE.md) +- 实现总结: [MULTI_FRAMEWORK_SUMMARY.md](../MULTI_FRAMEWORK_SUMMARY.md) diff --git a/agent_templates/agent_manager.db b/agent_templates/agent_manager.db new file mode 100644 index 0000000..77813d3 Binary files /dev/null and b/agent_templates/agent_manager.db differ diff --git a/agent_templates/azure_blob_agent.Dockerfile b/agent_templates/azure_blob_agent.Dockerfile new file mode 100644 index 0000000..e322d29 --- /dev/null +++ b/agent_templates/azure_blob_agent.Dockerfile @@ -0,0 +1,34 @@ +FROM python:3.11-slim + +WORKDIR /app + +# 安装系统依赖 +RUN apt-get update && apt-get install -y \ + curl \ + && rm -rf /var/lib/apt/lists/* + +# 安装Python依赖 +RUN pip install --no-cache-dir \ + fastapi==0.109.0 \ + uvicorn[standard]==0.27.0 \ + pydantic==2.5.3 \ + langchain==0.1.0 \ + langchain-community==0.0.10 \ + litellm==1.17.0 \ + azure-storage-blob==12.19.0 \ + azure-identity==1.15.0 + +# 复制agent代码 +COPY azure_blob_agent.py . + +# 设置环境变量 +ENV PYTHONUNBUFFERED=1 +ENV SERVICE_HOST=0.0.0.0 +ENV SERVICE_PORT=8080 + +# 健康检查 - 使用Python避免僵尸进程 +HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \ + CMD python3 -c "import urllib.request; urllib.request.urlopen('http://localhost:8080/health').read()" || exit 1 + +# 运行agent (直接使用Python，避免shell) +CMD ["python3", "-u", "azure_blob_agent.py"] diff --git a/agent_templates/azure_blob_agent.py b/agent_templates/azure_blob_agent.py new file mode 100644 index 0000000..7be6ec1 --- /dev/null +++ b/agent_templates/azure_blob_agent.py @@ -0,0 +1,514 @@ +""" +Azure Blob Storage AI Agent - 使用LangChain + LiteLLM实现 +通过HTTP API接收连接字符串，并提供智能文件操作功能 +""" +import os +import logging +from typing import Optional, Dict, Any, List +from datetime import datetime +from fastapi import FastAPI, HTTPException +from pydantic import BaseModel, Field +from azure.storage.blob import BlobServiceClient, ContainerClient +from langchain.agents import Tool, AgentExecutor, create_react_agent +from langchain.prompts import PromptTemplate +from langchain_community.chat_models import ChatLiteLLM +import uvicorn + +# 配置日志 +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger(__name__) + +# 环境变量配置 +SERVICE_HOST = os.getenv("SERVICE_HOST", "0.0.0.0") +SERVICE_PORT = int(os.getenv("SERVICE_PORT", "8080")) +POD_NAME = os.getenv("POD_NAME", "azure-blob-agent") +TEMPLATE_TYPE = os.getenv("TEMPLATE_TYPE", "azure_blob_agent") + +# LiteLLM配置 +LITELLM_API_BASE = os.getenv("LITELLM_API_BASE", "http://localhost:4000") +LITELLM_MODEL = os.getenv("LITELLM_MODEL", "gpt-3.5-turbo") +LITELLM_API_KEY = os.getenv("LITELLM_API_KEY", "sk-1234") + +# Azure Storage 连接字符串（可选，也可通过API动态传入） +AZURE_STORAGE_CONNECTION_STRING = os.getenv("AZURE_STORAGE_CONNECTION_STRING", "") + +# 全局存储客户端 +blob_service_client: Optional[BlobServiceClient] = None +connection_string: Optional[str] = None + +# FastAPI应用 +app = FastAPI( + title="Azure Blob Storage AI Agent", + description="智能Azure Blob存储管理代理", + version="1.0.0" +) + + +# ==================== 请求/响应模型 ==================== + +class ConnectRequest(BaseModel): + """连接请求""" + connection_string: str = Field(..., description="Azure Storage连接字符串") + + +class QueryRequest(BaseModel): + """查询请求""" + query: str = Field(..., description="自然语言查询或操作指令") + container_name: Optional[str] = Field(None, description="指定容器名称") + + +class HealthResponse(BaseModel): + """健康检查响应""" + status: str + connected: bool + connection_info: Optional[Dict] = None + + +# ==================== Azure Blob Storage 工具函数 ==================== + +def list_containers_tool() -> str: + """列出所有容器""" + global blob_service_client + + if not blob_service_client: + return "错误: 未连接到Azure Blob Storage" + + try: + containers = blob_service_client.list_containers() + container_list = [] + for container in containers: + container_list.append({ + "name": container.name, + "last_modified": str(container.last_modified) + }) + + if not container_list: + return "当前没有容器" + + result = "容器列表:\n" + for i, c in enumerate(container_list, 1): + result += f"{i}. {c['name']} (最后修改: {c['last_modified']})\n" + + return result + except Exception as e: + logger.error(f"列出容器失败: {str(e)}") + return f"错误: {str(e)}" + + +def list_blobs_in_container(container_name: str) -> str: + """列出指定容器中的所有blob""" + global blob_service_client + + if not blob_service_client: + return "错误: 未连接到Azure Blob Storage" + + try: + container_client = blob_service_client.get_container_client(container_name) + blobs = container_client.list_blobs() + + blob_list = [] + for blob in blobs: + blob_list.append({ + "name": blob.name, + "size": blob.size, + "content_type": blob.content_settings.content_type if blob.content_settings else "unknown", + "last_modified": str(blob.last_modified) + }) + + if not blob_list: + return f"容器 '{container_name}' 中没有文件" + + result = f"容器 '{container_name}' 中的文件列表:\n" + total_size = 0 + for i, b in enumerate(blob_list, 1): + size_mb = b['size'] / (1024 * 1024) + result += f"{i}. {b['name']} ({size_mb:.2f}MB, {b['content_type']})\n" + total_size += b['size'] + + result += f"\n总计: {len(blob_list)} 个文件, {total_size / (1024 * 1024):.2f}MB" + + return result + except Exception as e: + logger.error(f"列出blob失败: {str(e)}") + return f"错误: {str(e)}" + + +def get_blob_info(container_name: str, blob_name: str) -> str: + """获取blob的详细信息""" + global blob_service_client + + if not blob_service_client: + return "错误: 未连接到Azure Blob Storage" + + try: + blob_client = blob_service_client.get_blob_client(container_name, blob_name) + properties = blob_client.get_blob_properties() + + info = f"文件信息: {blob_name}\n" + info += f"- 容器: {container_name}\n" + info += f"- 大小: {properties.size / (1024 * 1024):.2f}MB\n" + info += f"- 类型: {properties.content_settings.content_type if properties.content_settings else 'unknown'}\n" + info += f"- 创建时间: {properties.creation_time}\n" + info += f"- 最后修改: {properties.last_modified}\n" + info += f"- ETag: {properties.etag}\n" + + if properties.metadata: + info += f"- 元数据: {properties.metadata}\n" + + return info + except Exception as e: + logger.error(f"获取blob信息失败: {str(e)}") + return f"错误: {str(e)}" + + +def search_blobs(container_name: str, keyword: str) -> str: + """在容器中搜索包含关键字的blob""" + global blob_service_client + + if not blob_service_client: + return "错误: 未连接到Azure Blob Storage" + + try: + container_client = blob_service_client.get_container_client(container_name) + blobs = container_client.list_blobs() + + matched_blobs = [] + for blob in blobs: + if keyword.lower() in blob.name.lower(): + matched_blobs.append({ + "name": blob.name, + "size": blob.size, + "last_modified": str(blob.last_modified) + }) + + if not matched_blobs: + return f"在容器 '{container_name}' 中没有找到包含 '{keyword}' 的文件" + + result = f"搜索结果 (关键字: '{keyword}'):\n" + for i, b in enumerate(matched_blobs, 1): + result += f"{i}. {b['name']} ({b['size'] / 1024:.2f}KB)\n" + + return result + except Exception as e: + logger.error(f"搜索blob失败: {str(e)}") + return f"错误: {str(e)}" + + +def get_storage_stats() -> str: + """获取存储统计信息""" + global blob_service_client + + if not blob_service_client: + return "错误: 未连接到Azure Blob Storage" + + try: + containers = list(blob_service_client.list_containers()) + total_containers = len(containers) + total_blobs = 0 + total_size = 0 + + container_stats = [] + for container in containers: + container_client = blob_service_client.get_container_client(container.name) + blobs = list(container_client.list_blobs()) + blob_count = len(blobs) + container_size = sum(blob.size for blob in blobs) + + total_blobs += blob_count + total_size += container_size + + container_stats.append({ + "name": container.name, + "blobs": blob_count, + "size_mb": container_size / (1024 * 1024) + }) + + result = "存储统计信息:\n" + result += f"- 总容器数: {total_containers}\n" + result += f"- 总文件数: {total_blobs}\n" + result += f"- 总大小: {total_size / (1024 * 1024):.2f}MB\n\n" + + if container_stats: + result += "各容器详情:\n" + for stat in container_stats: + result += f" • {stat['name']}: {stat['blobs']} 个文件, {stat['size_mb']:.2f}MB\n" + + return result + except Exception as e: + logger.error(f"获取统计信息失败: {str(e)}") + return f"错误: {str(e)}" + + +# ==================== 创建LangChain Agent ==================== + +def create_blob_agent() -> Optional[AgentExecutor]: + """创建Azure Blob Storage Agent""" + global blob_service_client + + if not blob_service_client: + logger.warning("尚未连接到Azure Blob Storage") + return None + + # 初始化LiteLLM + try: + llm = ChatLiteLLM( + model=LITELLM_MODEL, + api_base=LITELLM_API_BASE, + api_key=LITELLM_API_KEY, + temperature=0 + ) + logger.info(f"✅ LiteLLM初始化成功: {LITELLM_MODEL} @ {LITELLM_API_BASE}") + except Exception as e: + logger.error(f"❌ LiteLLM初始化失败: {str(e)}") + return None + + # 定义工具 + tools = [ + Tool( + name="list_containers", + func=list_containers_tool, + description="列出所有Azure Blob Storage容器。当用户询问'有哪些容器'、'显示容器列表'时使用此工具。" + ), + Tool( + name="list_blobs", + func=lambda input_str: list_blobs_in_container(input_str), + description="列出指定容器中的所有文件。输入参数是容器名称。当用户询问'容器X中有什么文件'、'列出XXX容器的文件'时使用此工具。" + ), + Tool( + name="get_blob_info", + func=lambda input_str: get_blob_info(*input_str.split(",")), + description="获取特定文件的详细信息。输入格式: '容器名,文件名'。当用户询问'文件XXX的详细信息'、'XXX文件的属性'时使用此工具。" + ), + Tool( + name="search_blobs", + func=lambda input_str: search_blobs(*input_str.split(",", 1)), + description="在容器中搜索文件。输入格式: '容器名,关键字'。当用户询问'搜索包含XXX的文件'、'查找XXX'时使用此工具。" + ), + Tool( + name="get_storage_stats", + func=get_storage_stats, + description="获取存储的统计信息，包括容器数量、文件数量、总大小等。当用户询问'存储统计'、'有多少文件'、'占用多少空间'时使用此工具。" + ), + ] + + # 定义Agent Prompt + template = """你是一个Azure Blob Storage管理助手。你可以帮助用户管理和查询Azure存储中的文件。 + +可用工具: +{tools} + +工具名称: {tool_names} + +回答问题时请使用以下格式: + +Question: 用户的输入问题 +Thought: 你应该思考如何回答这个问题 +Action: 要使用的工具名称，必须是以下之一: [{tool_names}] +Action Input: 传递给工具的输入 +Observation: 工具返回的结果 +... (这个 Thought/Action/Action Input/Observation 可以重复N次) +Thought: 我现在知道最终答案了 +Final Answer: 对用户问题的最终回答 + +重要提示: +- 如果用户只是说"列出容器"或"显示容器"，使用 list_containers 工具 +- 如果用户说"显示XXX容器的文件"，使用 list_blobs 工具，传入容器名 +- 搜索时需要同时提供容器名和关键字 +- 获取文件信息时需要提供容器名和文件名，用逗号分隔 +- 始终用中文回答 + +开始! + +Question: {input} +Thought: {agent_scratchpad}""" + + prompt = PromptTemplate( + template=template, + input_variables=["input", "agent_scratchpad"], + partial_variables={ + "tools": "\n".join([f"- {tool.name}: {tool.description}" for tool in tools]), + "tool_names": ", ".join([tool.name for tool in tools]) + } + ) + + # 创建Agent + agent = create_react_agent(llm, tools, prompt) + + # 创建Agent执行器 + agent_executor = AgentExecutor( + agent=agent, + tools=tools, + verbose=True, + handle_parsing_errors=True, + max_iterations=5 + ) + + logger.info("✅ Azure Blob Storage Agent创建成功") + return agent_executor + + +# ==================== API端点 ==================== + +@app.get("/health", response_model=HealthResponse) +async def health_check(): + """健康检查""" + global blob_service_client, connection_string + + connected = blob_service_client is not None + + connection_info = None + if connected: + try: + # 获取账户信息 + account_info = blob_service_client.get_account_information() + connection_info = { + "account_kind": account_info.get('account_kind', 'unknown'), + "sku_name": account_info.get('sku_name', 'unknown'), + "connected_at": str(datetime.now()) + } + except Exception as e: + logger.error(f"获取账户信息失败: {str(e)}") + + return HealthResponse( + status="healthy" if connected else "not_connected", + connected=connected, + connection_info=connection_info + ) + + +@app.post("/connect") +async def connect_to_storage(request: ConnectRequest): + """连接到Azure Blob Storage""" + global blob_service_client, connection_string + + try: + # 创建BlobServiceClient + blob_service_client = BlobServiceClient.from_connection_string( + request.connection_string + ) + + # 测试连接 + account_info = blob_service_client.get_account_information() + + connection_string = request.connection_string + + logger.info(f"✅ 成功连接到Azure Blob Storage") + + return { + "status": "connected", + "message": "成功连接到Azure Blob Storage", + "account_info": { + "account_kind": account_info.get('account_kind'), + "sku_name": account_info.get('sku_name') + } + } + except Exception as e: + logger.error(f"❌ 连接失败: {str(e)}") + blob_service_client = None + connection_string = None + raise HTTPException(status_code=400, detail=f"连接失败: {str(e)}") + + +@app.post("/query") +async def query_storage(request: QueryRequest): + """使用自然语言查询存储""" + global blob_service_client + + if not blob_service_client: + raise HTTPException( + status_code=400, + detail="未连接到Azure Blob Storage，请先调用 /connect" + ) + + try: + # 创建Agent + agent = create_blob_agent() + + if not agent: + raise HTTPException(status_code=500, detail="Agent创建失败") + + # 执行查询 + logger.info(f"收到查询: {request.query}") + result = agent.invoke({"input": request.query}) + + return { + "status": "success", + "query": request.query, + "answer": result.get("output", "无法生成答案"), + "intermediate_steps": str(result.get("intermediate_steps", [])) + } + except Exception as e: + logger.error(f"查询执行失败: {str(e)}") + raise HTTPException(status_code=500, detail=f"查询失败: {str(e)}") + + +@app.get("/") +async def root(): + """根端点""" + return { + "service": "Azure Blob Storage AI Agent", + "version": "1.0.0", + "pod_name": POD_NAME, + "template": TEMPLATE_TYPE, + "connected": blob_service_client is not None, + "endpoints": { + "health": "/health", + "connect": "POST /connect", + "query": "POST /query" + } + } + + +# ==================== 主函数 ==================== + +def init_storage_connection(): + """启动时初始化存储连接""" + global blob_service_client, connection_string + + if AZURE_STORAGE_CONNECTION_STRING: + try: + logger.info("检测到环境变量中的连接字符串，尝试连接...") + blob_service_client = BlobServiceClient.from_connection_string( + AZURE_STORAGE_CONNECTION_STRING + ) + + # 测试连接 + account_info = blob_service_client.get_account_information() + connection_string = AZURE_STORAGE_CONNECTION_STRING + + logger.info(f"✅ 成功连接到Azure Blob Storage") + logger.info(f" - Account Kind: {account_info.get('account_kind')}") + logger.info(f" - SKU: {account_info.get('sku_name')}") + except Exception as e: + logger.error(f"❌ 启动时连接失败: {str(e)}") + logger.info("💡 提示: 可以稍后通过 /connect API 手动连接") + blob_service_client = None + connection_string = None + else: + logger.info("💡 未设置 AZURE_STORAGE_CONNECTION_STRING，需通过 /connect API 手动连接") + + +def main(): + """启动服务""" + logger.info(f"🚀 启动 Azure Blob Storage AI Agent") + logger.info(f" - Pod名称: {POD_NAME}") + logger.info(f" - 模板类型: {TEMPLATE_TYPE}") + logger.info(f" - LiteLLM: {LITELLM_MODEL} @ {LITELLM_API_BASE}") + logger.info(f" - 服务地址: http://{SERVICE_HOST}:{SERVICE_PORT}") + + # 初始化存储连接 + init_storage_connection() + + uvicorn.run( + app, + host=SERVICE_HOST, + port=SERVICE_PORT, + log_level="info" + ) + + +if __name__ == "__main__": + main() diff --git a/agent_templates/azure_blob_agent_a2a.Dockerfile b/agent_templates/azure_blob_agent_a2a.Dockerfile new file mode 100644 index 0000000..b0d29a0 --- /dev/null +++ b/agent_templates/azure_blob_agent_a2a.Dockerfile @@ -0,0 +1,28 @@ +# Azure Blob Agent - A2A 版本 Dockerfile +FROM python:3.11-slim + +WORKDIR /app + +# 安装系统依赖 +RUN apt-get update && apt-get install -y \ + gcc \ + && rm -rf /var/lib/apt/lists/* + +# 复制requirements文件 +COPY requirements_a2a.txt /app/ + +# 安装Python依赖 +RUN pip install --no-cache-dir -r requirements_a2a.txt + +# 复制应用代码 +COPY azure_blob_agent_a2a.py /app/ + +# 暴露端口 +EXPOSE 8080 + +# 健康检查 +HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \ + CMD python -c "import requests; requests.get('http://localhost:8080/health', timeout=5)" + +# 启动应用 +CMD ["python", "azure_blob_agent_a2a.py"] diff --git a/agent_templates/azure_blob_agent_a2a.py b/agent_templates/azure_blob_agent_a2a.py new file mode 100644 index 0000000..f256993 --- /dev/null +++ b/agent_templates/azure_blob_agent_a2a.py @@ -0,0 +1,652 @@ +""" +Azure Blob Storage AI Agent - A2A (Agent-to-Agent) 版本 +支持 Agent 之间的协作和通信 +""" +import os +import logging +import json +import httpx +from typing import Optional, Dict, Any, List +from datetime import datetime +from fastapi import FastAPI, HTTPException, Header +from pydantic import BaseModel, Field +from azure.storage.blob import BlobServiceClient, ContainerClient +import uvicorn + +# 配置日志 +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger(__name__) + +# 环境变量配置 +SERVICE_HOST = os.getenv("SERVICE_HOST", "0.0.0.0") +SERVICE_PORT = int(os.getenv("SERVICE_PORT", "8080")) +POD_NAME = os.getenv("POD_NAME", "azure-blob-agent-a2a") +TEMPLATE_TYPE = os.getenv("TEMPLATE_TYPE", "azure_blob_agent_a2a") +AGENT_FRAMEWORK = os.getenv("AGENT_FRAMEWORK", "a2a") + +# 工具配置 +TOOLS_CONFIG = json.loads(os.getenv("TOOLS_CONFIG", "{}")) +TOOL_ENDPOINT = os.getenv("TOOL_ENDPOINT", "") +TOOL_API_KEY = os.getenv("TOOL_API_KEY", "") + +# 模型配置 +MODEL_PROVIDER = os.getenv("MODEL_PROVIDER", "openai") +MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4") +MODEL_API_KEY = os.getenv("MODEL_API_KEY", "") +MODEL_ENDPOINT = os.getenv("MODEL_ENDPOINT", "https://api.openai.com/v1") + +# 存储配置 +AZURE_STORAGE_CONNECTION_STRING = os.getenv("AZURE_STORAGE_CONNECTION_STRING", "") +STORAGE_ACCOUNT_NAME = os.getenv("STORAGE_ACCOUNT_NAME", "") + +# 用户标识 +USER_ID = os.getenv("USER_ID", "") +TENANT_ID = os.getenv("TENANT_ID", "") +NAMESPACE = os.getenv("NAMESPACE", "ai-agents") + +# A2A Agent 配置 +AGENT_ID = os.getenv("AGENT_ID", POD_NAME) +AGENT_ROLE = os.getenv("AGENT_ROLE", "storage_manager") +AGENT_CAPABILITIES = json.loads(os.getenv("AGENT_CAPABILITIES", '["blob_storage", "file_operations"]')) + +# 全局存储客户端 +blob_service_client: Optional[BlobServiceClient] = None +connection_string: Optional[str] = None + +# A2A Agent 注册表 (其他可协作的 Agent) +registered_agents: Dict[str, Dict] = {} + +# FastAPI应用 +app = FastAPI( + title="Azure Blob Storage AI Agent (A2A)", + description="支持 Agent-to-Agent 协作的智能 Azure Blob 存储管理代理", + version="1.0.0" +) + + +# ==================== 请求/响应模型 ==================== + +class ConnectRequest(BaseModel): + """连接请求""" + connection_string: str = Field(..., description="Azure Storage连接字符串") + + +class A2AMessage(BaseModel): + """A2A 消息格式""" + message_id: str = Field(..., description="消息ID") + from_agent: str = Field(..., description="发送者 Agent ID") + to_agent: str = Field(..., description="接收者 Agent ID") + message_type: str = Field(..., description="消息类型: request/response/notification") + action: str = Field(..., description="请求的动作") + parameters: Dict[str, Any] = Field(default_factory=dict, description="参数") + context: Optional[Dict] = Field(default_factory=dict, description="上下文") + timestamp: Optional[str] = None + + +class A2AQueryRequest(BaseModel): + """A2A 查询请求""" + query: str = Field(..., description="自然语言查询") + container_name: Optional[str] = None + requester_agent: Optional[str] = Field(None, description="请求者 Agent ID") + context: Optional[Dict] = Field(default_factory=dict) + + +class A2ARegisterRequest(BaseModel): + """A2A Agent 注册请求""" + agent_id: str + agent_role: str + capabilities: List[str] + endpoint: str + + +class HealthResponse(BaseModel): + """健康检查响应""" + status: str + connected: bool + framework: str + agent_id: str + agent_role: str + capabilities: List[str] + user_id: Optional[str] = None + namespace: Optional[str] = None + registered_agents_count: int = 0 + connection_info: Optional[Dict] = None + + +# ==================== A2A 操作处理器 ==================== + +class A2AActionHandler: + """A2A 动作处理器""" + + @staticmethod + async def handle_list_containers(parameters: Dict) -> Dict: + """处理列出容器请求""" + global blob_service_client + + if not blob_service_client: + return {"error": "未连接到 Azure Blob Storage"} + + try: + containers = blob_service_client.list_containers() + container_list = [] + for container in containers: + container_list.append({ + "name": container.name, + "last_modified": str(container.last_modified) + }) + + return { + "success": True, + "containers": container_list, + "count": len(container_list) + } + except Exception as e: + logger.error(f"列出容器失败: {str(e)}") + return {"error": str(e)} + + @staticmethod + async def handle_list_blobs(parameters: Dict) -> Dict: + """处理列出 blob 请求""" + global blob_service_client + + if not blob_service_client: + return {"error": "未连接到 Azure Blob Storage"} + + container_name = parameters.get("container_name") + if not container_name: + return {"error": "缺少参数: container_name"} + + try: + container_client = blob_service_client.get_container_client(container_name) + blobs = container_client.list_blobs() + + blob_list = [] + total_size = 0 + for blob in blobs: + blob_info = { + "name": blob.name, + "size": blob.size, + "size_mb": round(blob.size / (1024 * 1024), 2), + "content_type": blob.content_settings.content_type if blob.content_settings else "unknown", + "last_modified": str(blob.last_modified) + } + blob_list.append(blob_info) + total_size += blob.size + + return { + "success": True, + "container": container_name, + "blobs": blob_list, + "count": len(blob_list), + "total_size_mb": round(total_size / (1024 * 1024), 2) + } + except Exception as e: + logger.error(f"列出 blob 失败: {str(e)}") + return {"error": str(e)} + + @staticmethod + async def handle_get_blob_info(parameters: Dict) -> Dict: + """处理获取 blob 信息请求""" + global blob_service_client + + if not blob_service_client: + return {"error": "未连接到 Azure Blob Storage"} + + container_name = parameters.get("container_name") + blob_name = parameters.get("blob_name") + + if not container_name or not blob_name: + return {"error": "缺少参数: container_name 或 blob_name"} + + try: + blob_client = blob_service_client.get_blob_client(container_name, blob_name) + properties = blob_client.get_blob_properties() + + return { + "success": True, + "blob_name": blob_name, + "container": container_name, + "size": properties.size, + "size_mb": round(properties.size / (1024 * 1024), 2), + "content_type": properties.content_settings.content_type if properties.content_settings else "unknown", + "creation_time": str(properties.creation_time), + "last_modified": str(properties.last_modified), + "etag": properties.etag, + "metadata": properties.metadata if properties.metadata else {} + } + except Exception as e: + logger.error(f"获取 blob 信息失败: {str(e)}") + return {"error": str(e)} + + @staticmethod + async def handle_search_blobs(parameters: Dict) -> Dict: + """处理搜索 blob 请求""" + global blob_service_client + + if not blob_service_client: + return {"error": "未连接到 Azure Blob Storage"} + + container_name = parameters.get("container_name") + keyword = parameters.get("keyword") + + if not container_name or not keyword: + return {"error": "缺少参数: container_name 或 keyword"} + + try: + container_client = blob_service_client.get_container_client(container_name) + blobs = container_client.list_blobs() + + matched_blobs = [] + for blob in blobs: + if keyword.lower() in blob.name.lower(): + matched_blobs.append({ + "name": blob.name, + "size": blob.size, + "size_kb": round(blob.size / 1024, 2), + "last_modified": str(blob.last_modified) + }) + + return { + "success": True, + "container": container_name, + "keyword": keyword, + "results": matched_blobs, + "count": len(matched_blobs) + } + except Exception as e: + logger.error(f"搜索 blob 失败: {str(e)}") + return {"error": str(e)} + + @staticmethod + async def handle_get_stats(parameters: Dict) -> Dict: + """处理获取统计信息请求""" + global blob_service_client + + if not blob_service_client: + return {"error": "未连接到 Azure Blob Storage"} + + try: + containers = list(blob_service_client.list_containers()) + total_containers = len(containers) + total_blobs = 0 + total_size = 0 + + container_stats = [] + for container in containers: + container_client = blob_service_client.get_container_client(container.name) + blobs = list(container_client.list_blobs()) + blob_count = len(blobs) + container_size = sum(blob.size for blob in blobs) + + total_blobs += blob_count + total_size += container_size + + container_stats.append({ + "name": container.name, + "blobs": blob_count, + "size_mb": round(container_size / (1024 * 1024), 2) + }) + + return { + "success": True, + "total_containers": total_containers, + "total_blobs": total_blobs, + "total_size_mb": round(total_size / (1024 * 1024), 2), + "container_stats": container_stats + } + except Exception as e: + logger.error(f"获取统计信息失败: {str(e)}") + return {"error": str(e)} + + +# 动作路由表 +ACTION_HANDLERS = { + "list_containers": A2AActionHandler.handle_list_containers, + "list_blobs": A2AActionHandler.handle_list_blobs, + "get_blob_info": A2AActionHandler.handle_get_blob_info, + "search_blobs": A2AActionHandler.handle_search_blobs, + "get_stats": A2AActionHandler.handle_get_stats, +} + + +# ==================== API 端点 ==================== + +@app.get("/health", response_model=HealthResponse) +async def health_check(): + """健康检查""" + global blob_service_client, connection_string + + connected = blob_service_client is not None + + connection_info = None + if connected: + try: + account_info = blob_service_client.get_account_information() + connection_info = { + "account_kind": account_info.get('account_kind', 'unknown'), + "sku_name": account_info.get('sku_name', 'unknown'), + "connected_at": str(datetime.now()) + } + except Exception as e: + logger.error(f"获取账户信息失败: {str(e)}") + + return HealthResponse( + status="healthy" if connected else "not_connected", + connected=connected, + framework=AGENT_FRAMEWORK, + agent_id=AGENT_ID, + agent_role=AGENT_ROLE, + capabilities=AGENT_CAPABILITIES, + user_id=USER_ID, + namespace=NAMESPACE, + registered_agents_count=len(registered_agents), + connection_info=connection_info + ) + + +@app.post("/connect") +async def connect_to_storage(request: ConnectRequest): + """连接到 Azure Blob Storage""" + global blob_service_client, connection_string + + try: + blob_service_client = BlobServiceClient.from_connection_string( + request.connection_string + ) + + account_info = blob_service_client.get_account_information() + connection_string = request.connection_string + + logger.info(f"✅ 成功连接到 Azure Blob Storage (Agent: {AGENT_ID}, User: {USER_ID})") + + return { + "status": "connected", + "message": "成功连接到 Azure Blob Storage", + "framework": AGENT_FRAMEWORK, + "agent_id": AGENT_ID, + "user_id": USER_ID, + "account_info": { + "account_kind": account_info.get('account_kind'), + "sku_name": account_info.get('sku_name') + } + } + except Exception as e: + logger.error(f"❌ 连接失败: {str(e)}") + blob_service_client = None + connection_string = None + raise HTTPException(status_code=400, detail=f"连接失败: {str(e)}") + + +@app.get("/a2a/capabilities") +async def get_capabilities(): + """获取 Agent 能力""" + return { + "agent_id": AGENT_ID, + "agent_role": AGENT_ROLE, + "capabilities": AGENT_CAPABILITIES, + "supported_actions": list(ACTION_HANDLERS.keys()), + "framework": AGENT_FRAMEWORK + } + + +@app.post("/a2a/register") +async def register_agent(request: A2ARegisterRequest): + """注册其他 Agent""" + global registered_agents + + registered_agents[request.agent_id] = { + "agent_id": request.agent_id, + "agent_role": request.agent_role, + "capabilities": request.capabilities, + "endpoint": request.endpoint, + "registered_at": str(datetime.now()) + } + + logger.info(f"✅ Agent '{request.agent_id}' 注册成功") + + return { + "status": "registered", + "agent_id": request.agent_id, + "message": f"Agent '{request.agent_id}' 已注册" + } + + +@app.get("/a2a/agents") +async def list_registered_agents(): + """列出已注册的 Agent""" + return { + "agents": list(registered_agents.values()), + "count": len(registered_agents) + } + + +@app.post("/a2a/message") +async def handle_a2a_message(message: A2AMessage): + """处理 A2A 消息""" + if not blob_service_client: + raise HTTPException( + status_code=400, + detail="未连接到 Azure Blob Storage，请先调用 /connect" + ) + + # 验证消息目标 + if message.to_agent != AGENT_ID: + raise HTTPException( + status_code=400, + detail=f"消息目标不匹配: 期望 {AGENT_ID}, 收到 {message.to_agent}" + ) + + # 处理消息 + if message.message_type == "request": + action = message.action + + if action not in ACTION_HANDLERS: + return { + "message_id": message.message_id, + "status": "error", + "error": f"不支持的动作: {action}", + "supported_actions": list(ACTION_HANDLERS.keys()) + } + + try: + handler = ACTION_HANDLERS[action] + result = await handler(message.parameters) + + return { + "message_id": message.message_id, + "from_agent": AGENT_ID, + "to_agent": message.from_agent, + "message_type": "response", + "action": action, + "result": result, + "timestamp": str(datetime.now()) + } + except Exception as e: + logger.error(f"处理 A2A 消息失败: {str(e)}") + return { + "message_id": message.message_id, + "status": "error", + "error": str(e) + } + + return { + "message_id": message.message_id, + "status": "info", + "message": f"收到消息类型: {message.message_type}" + } + + +@app.post("/query") +async def query_storage(request: A2AQueryRequest): + """查询存储（支持 A2A 上下文）""" + if not blob_service_client: + raise HTTPException( + status_code=400, + detail="未连接到 Azure Blob Storage，请先调用 /connect" + ) + + try: + query = request.query.lower() + result = None + action_used = None + + # 简单的规则匹配 + if "容器" in query and ("列出" in query or "显示" in query or "有哪些" in query): + result = await A2AActionHandler.handle_list_containers({}) + action_used = "list_containers" + elif "统计" in query or "有多少" in query or "占用" in query: + result = await A2AActionHandler.handle_get_stats({}) + action_used = "get_stats" + elif request.container_name: + if "文件" in query or "blob" in query.lower(): + result = await A2AActionHandler.handle_list_blobs({"container_name": request.container_name}) + action_used = "list_blobs" + + return { + "status": "success" if result else "info", + "query": request.query, + "action": action_used, + "result": result, + "agent_id": AGENT_ID, + "requester": request.requester_agent, + "framework": AGENT_FRAMEWORK + } + except Exception as e: + logger.error(f"查询执行失败: {str(e)}") + raise HTTPException(status_code=500, detail=f"查询失败: {str(e)}") + + +@app.post("/a2a/collaborate") +async def collaborate_with_agent( + target_agent_id: str, + action: str, + parameters: Dict[str, Any] +): + """与其他 Agent 协作""" + if target_agent_id not in registered_agents: + raise HTTPException( + status_code=404, + detail=f"Agent '{target_agent_id}' 未注册" + ) + + target_agent = registered_agents[target_agent_id] + + # 创建 A2A 消息 + message = A2AMessage( + message_id=f"{AGENT_ID}_{datetime.now().timestamp()}", + from_agent=AGENT_ID, + to_agent=target_agent_id, + message_type="request", + action=action, + parameters=parameters, + timestamp=str(datetime.now()) + ) + + try: + # 发送请求到目标 Agent + async with httpx.AsyncClient() as client: + response = await client.post( + f"{target_agent['endpoint']}/a2a/message", + json=message.dict(), + timeout=30.0 + ) + response.raise_for_status() + + return { + "status": "success", + "target_agent": target_agent_id, + "action": action, + "response": response.json() + } + except Exception as e: + logger.error(f"协作失败: {str(e)}") + raise HTTPException(status_code=500, detail=f"协作失败: {str(e)}") + + +@app.get("/") +async def root(): + """根端点""" + return { + "service": "Azure Blob Storage AI Agent", + "version": "1.0.0", + "framework": AGENT_FRAMEWORK, + "agent_id": AGENT_ID, + "agent_role": AGENT_ROLE, + "capabilities": AGENT_CAPABILITIES, + "pod_name": POD_NAME, + "template": TEMPLATE_TYPE, + "user_id": USER_ID, + "namespace": NAMESPACE, + "connected": blob_service_client is not None, + "registered_agents": len(registered_agents), + "endpoints": { + "health": "/health", + "connect": "POST /connect", + "capabilities": "GET /a2a/capabilities", + "register_agent": "POST /a2a/register", + "list_agents": "GET /a2a/agents", + "handle_message": "POST /a2a/message", + "collaborate": "POST /a2a/collaborate", + "query": "POST /query" + } + } + + +# ==================== 主函数 ==================== + +def init_storage_connection(): + """启动时初始化存储连接""" + global blob_service_client, connection_string + + if AZURE_STORAGE_CONNECTION_STRING: + try: + logger.info("检测到环境变量中的连接字符串，尝试连接...") + blob_service_client = BlobServiceClient.from_connection_string( + AZURE_STORAGE_CONNECTION_STRING + ) + + account_info = blob_service_client.get_account_information() + connection_string = AZURE_STORAGE_CONNECTION_STRING + + logger.info(f"✅ 成功连接到 Azure Blob Storage") + logger.info(f" - Account Kind: {account_info.get('account_kind')}") + logger.info(f" - SKU: {account_info.get('sku_name')}") + except Exception as e: + logger.error(f"❌ 启动时连接失败: {str(e)}") + logger.info("💡 提示: 可以稍后通过 /connect API 手动连接") + blob_service_client = None + connection_string = None + else: + logger.info("💡 未设置 AZURE_STORAGE_CONNECTION_STRING，需通过 /connect API 手动连接") + + +def main(): + """启动服务""" + logger.info(f"🚀 启动 Azure Blob Storage AI Agent (A2A)") + logger.info(f" - Framework: {AGENT_FRAMEWORK}") + logger.info(f" - Agent ID: {AGENT_ID}") + logger.info(f" - Agent Role: {AGENT_ROLE}") + logger.info(f" - Capabilities: {AGENT_CAPABILITIES}") + logger.info(f" - Pod名称: {POD_NAME}") + logger.info(f" - 模板类型: {TEMPLATE_TYPE}") + logger.info(f" - User ID: {USER_ID}") + logger.info(f" - Namespace: {NAMESPACE}") + logger.info(f" - 模型: {MODEL_NAME} @ {MODEL_PROVIDER}") + logger.info(f" - 服务地址: http://{SERVICE_HOST}:{SERVICE_PORT}") + + # 初始化存储连接 + init_storage_connection() + + uvicorn.run( + app, + host=SERVICE_HOST, + port=SERVICE_PORT, + log_level="info" + ) + + +if __name__ == "__main__": + main() diff --git a/agent_templates/azure_blob_agent_mcp.Dockerfile b/agent_templates/azure_blob_agent_mcp.Dockerfile new file mode 100644 index 0000000..f529d4b --- /dev/null +++ b/agent_templates/azure_blob_agent_mcp.Dockerfile @@ -0,0 +1,28 @@ +# Azure Blob Agent - MCP 版本 Dockerfile +FROM python:3.11-slim + +WORKDIR /app + +# 安装系统依赖 +RUN apt-get update && apt-get install -y \ + gcc \ + && rm -rf /var/lib/apt/lists/* + +# 复制requirements文件 +COPY requirements_mcp.txt /app/ + +# 安装Python依赖 +RUN pip install --no-cache-dir -r requirements_mcp.txt + +# 复制应用代码 +COPY azure_blob_agent_mcp.py /app/ + +# 暴露端口 +EXPOSE 8080 + +# 健康检查 +HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \ + CMD python -c "import requests; requests.get('http://localhost:8080/health', timeout=5)" + +# 启动应用 +CMD ["python", "azure_blob_agent_mcp.py"] diff --git a/agent_templates/azure_blob_agent_mcp.py b/agent_templates/azure_blob_agent_mcp.py new file mode 100644 index 0000000..06b98fa --- /dev/null +++ b/agent_templates/azure_blob_agent_mcp.py @@ -0,0 +1,622 @@ +""" +Azure Blob Storage AI Agent - MCP (Model Context Protocol) 版本 +使用 MCP 协议实现智能文件操作功能 +""" +import os +import logging +import json +from typing import Optional, Dict, Any, List +from datetime import datetime +from fastapi import FastAPI, HTTPException +from pydantic import BaseModel, Field +from azure.storage.blob import BlobServiceClient, ContainerClient +import uvicorn +import asyncio + +# 配置日志 +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger(__name__) + +# 环境变量配置 +SERVICE_HOST = os.getenv("SERVICE_HOST", "0.0.0.0") +SERVICE_PORT = int(os.getenv("SERVICE_PORT", "8080")) +POD_NAME = os.getenv("POD_NAME", "azure-blob-agent-mcp") +TEMPLATE_TYPE = os.getenv("TEMPLATE_TYPE", "azure_blob_agent_mcp") +AGENT_FRAMEWORK = os.getenv("AGENT_FRAMEWORK", "mcp") + +# 工具配置 (从环境变量传入的 JSON) +TOOLS_CONFIG = json.loads(os.getenv("TOOLS_CONFIG", "{}")) +TOOL_ENDPOINT = os.getenv("TOOL_ENDPOINT", "") +TOOL_API_KEY = os.getenv("TOOL_API_KEY", "") + +# 模型配置 +MODEL_PROVIDER = os.getenv("MODEL_PROVIDER", "openai") +MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4") +MODEL_API_KEY = os.getenv("MODEL_API_KEY", "") +MODEL_ENDPOINT = os.getenv("MODEL_ENDPOINT", "https://api.openai.com/v1") + +# 存储配置 +AZURE_STORAGE_CONNECTION_STRING = os.getenv("AZURE_STORAGE_CONNECTION_STRING", "") +STORAGE_ACCOUNT_NAME = os.getenv("STORAGE_ACCOUNT_NAME", "") + +# 用户标识 +USER_ID = os.getenv("USER_ID", "") +TENANT_ID = os.getenv("TENANT_ID", "") +NAMESPACE = os.getenv("NAMESPACE", "ai-agents") + +# 全局存储客户端 +blob_service_client: Optional[BlobServiceClient] = None +connection_string: Optional[str] = None + +# MCP 工具注册表 +mcp_tools: Dict[str, Any] = {} + +# FastAPI应用 +app = FastAPI( + title="Azure Blob Storage AI Agent (MCP)", + description="基于 MCP 协议的智能 Azure Blob 存储管理代理", + version="1.0.0" +) + + +# ==================== 请求/响应模型 ==================== + +class ConnectRequest(BaseModel): + """连接请求""" + connection_string: str = Field(..., description="Azure Storage连接字符串") + + +class MCPToolRequest(BaseModel): + """MCP 工具调用请求""" + tool_name: str = Field(..., description="工具名称") + parameters: Dict[str, Any] = Field(default_factory=dict, description="工具参数") + + +class MCPQueryRequest(BaseModel): + """MCP 查询请求""" + query: str = Field(..., description="自然语言查询或操作指令") + container_name: Optional[str] = Field(None, description="指定容器名称") + context: Optional[Dict] = Field(default_factory=dict, description="上下文信息") + + +class HealthResponse(BaseModel): + """健康检查响应""" + status: str + connected: bool + framework: str + user_id: Optional[str] = None + namespace: Optional[str] = None + connection_info: Optional[Dict] = None + + +# ==================== MCP 工具定义 ==================== + +class MCPTool: + """MCP 工具基类""" + + def __init__(self, name: str, description: str, parameters_schema: Dict): + self.name = name + self.description = description + self.parameters_schema = parameters_schema + + async def execute(self, parameters: Dict[str, Any]) -> Dict[str, Any]: + """执行工具""" + raise NotImplementedError + + def to_mcp_spec(self) -> Dict: + """转换为 MCP 工具规范""" + return { + "name": self.name, + "description": self.description, + "inputSchema": { + "type": "object", + "properties": self.parameters_schema, + "required": list(self.parameters_schema.keys()) + } + } + + +class ListContainersTool(MCPTool): + """列出所有容器工具""" + + def __init__(self): + super().__init__( + name="list_containers", + description="列出 Azure Blob Storage 中的所有容器", + parameters_schema={} + ) + + async def execute(self, parameters: Dict[str, Any]) -> Dict[str, Any]: + global blob_service_client + + if not blob_service_client: + return {"error": "未连接到 Azure Blob Storage"} + + try: + containers = blob_service_client.list_containers() + container_list = [] + for container in containers: + container_list.append({ + "name": container.name, + "last_modified": str(container.last_modified) + }) + + return { + "success": True, + "containers": container_list, + "count": len(container_list) + } + except Exception as e: + logger.error(f"列出容器失败: {str(e)}") + return {"error": str(e)} + + +class ListBlobsTool(MCPTool): + """列出容器中的 blob 工具""" + + def __init__(self): + super().__init__( + name="list_blobs", + description="列出指定容器中的所有文件", + parameters_schema={ + "container_name": { + "type": "string", + "description": "容器名称" + } + } + ) + + async def execute(self, parameters: Dict[str, Any]) -> Dict[str, Any]: + global blob_service_client + + if not blob_service_client: + return {"error": "未连接到 Azure Blob Storage"} + + container_name = parameters.get("container_name") + if not container_name: + return {"error": "缺少参数: container_name"} + + try: + container_client = blob_service_client.get_container_client(container_name) + blobs = container_client.list_blobs() + + blob_list = [] + total_size = 0 + for blob in blobs: + blob_info = { + "name": blob.name, + "size": blob.size, + "size_mb": round(blob.size / (1024 * 1024), 2), + "content_type": blob.content_settings.content_type if blob.content_settings else "unknown", + "last_modified": str(blob.last_modified) + } + blob_list.append(blob_info) + total_size += blob.size + + return { + "success": True, + "container": container_name, + "blobs": blob_list, + "count": len(blob_list), + "total_size_mb": round(total_size / (1024 * 1024), 2) + } + except Exception as e: + logger.error(f"列出 blob 失败: {str(e)}") + return {"error": str(e)} + + +class GetBlobInfoTool(MCPTool): + """获取 blob 信息工具""" + + def __init__(self): + super().__init__( + name="get_blob_info", + description="获取特定文件的详细信息", + parameters_schema={ + "container_name": { + "type": "string", + "description": "容器名称" + }, + "blob_name": { + "type": "string", + "description": "文件名称" + } + } + ) + + async def execute(self, parameters: Dict[str, Any]) -> Dict[str, Any]: + global blob_service_client + + if not blob_service_client: + return {"error": "未连接到 Azure Blob Storage"} + + container_name = parameters.get("container_name") + blob_name = parameters.get("blob_name") + + if not container_name or not blob_name: + return {"error": "缺少参数: container_name 或 blob_name"} + + try: + blob_client = blob_service_client.get_blob_client(container_name, blob_name) + properties = blob_client.get_blob_properties() + + return { + "success": True, + "blob_name": blob_name, + "container": container_name, + "size": properties.size, + "size_mb": round(properties.size / (1024 * 1024), 2), + "content_type": properties.content_settings.content_type if properties.content_settings else "unknown", + "creation_time": str(properties.creation_time), + "last_modified": str(properties.last_modified), + "etag": properties.etag, + "metadata": properties.metadata if properties.metadata else {} + } + except Exception as e: + logger.error(f"获取 blob 信息失败: {str(e)}") + return {"error": str(e)} + + +class SearchBlobsTool(MCPTool): + """搜索 blob 工具""" + + def __init__(self): + super().__init__( + name="search_blobs", + description="在容器中搜索包含关键字的文件", + parameters_schema={ + "container_name": { + "type": "string", + "description": "容器名称" + }, + "keyword": { + "type": "string", + "description": "搜索关键字" + } + } + ) + + async def execute(self, parameters: Dict[str, Any]) -> Dict[str, Any]: + global blob_service_client + + if not blob_service_client: + return {"error": "未连接到 Azure Blob Storage"} + + container_name = parameters.get("container_name") + keyword = parameters.get("keyword") + + if not container_name or not keyword: + return {"error": "缺少参数: container_name 或 keyword"} + + try: + container_client = blob_service_client.get_container_client(container_name) + blobs = container_client.list_blobs() + + matched_blobs = [] + for blob in blobs: + if keyword.lower() in blob.name.lower(): + matched_blobs.append({ + "name": blob.name, + "size": blob.size, + "size_kb": round(blob.size / 1024, 2), + "last_modified": str(blob.last_modified) + }) + + return { + "success": True, + "container": container_name, + "keyword": keyword, + "results": matched_blobs, + "count": len(matched_blobs) + } + except Exception as e: + logger.error(f"搜索 blob 失败: {str(e)}") + return {"error": str(e)} + + +class GetStorageStatsTool(MCPTool): + """获取存储统计工具""" + + def __init__(self): + super().__init__( + name="get_storage_stats", + description="获取存储的统计信息，包括容器数量、文件数量、总大小等", + parameters_schema={} + ) + + async def execute(self, parameters: Dict[str, Any]) -> Dict[str, Any]: + global blob_service_client + + if not blob_service_client: + return {"error": "未连接到 Azure Blob Storage"} + + try: + containers = list(blob_service_client.list_containers()) + total_containers = len(containers) + total_blobs = 0 + total_size = 0 + + container_stats = [] + for container in containers: + container_client = blob_service_client.get_container_client(container.name) + blobs = list(container_client.list_blobs()) + blob_count = len(blobs) + container_size = sum(blob.size for blob in blobs) + + total_blobs += blob_count + total_size += container_size + + container_stats.append({ + "name": container.name, + "blobs": blob_count, + "size_mb": round(container_size / (1024 * 1024), 2) + }) + + return { + "success": True, + "total_containers": total_containers, + "total_blobs": total_blobs, + "total_size_mb": round(total_size / (1024 * 1024), 2), + "container_stats": container_stats + } + except Exception as e: + logger.error(f"获取统计信息失败: {str(e)}") + return {"error": str(e)} + + +# ==================== MCP 工具注册 ==================== + +def register_tools(): + """注册所有 MCP 工具""" + global mcp_tools + + tools = [ + ListContainersTool(), + ListBlobsTool(), + GetBlobInfoTool(), + SearchBlobsTool(), + GetStorageStatsTool() + ] + + for tool in tools: + mcp_tools[tool.name] = tool + + logger.info(f"✅ 注册了 {len(mcp_tools)} 个 MCP 工具") + + +# ==================== API 端点 ==================== + +@app.get("/health", response_model=HealthResponse) +async def health_check(): + """健康检查""" + global blob_service_client, connection_string + + connected = blob_service_client is not None + + connection_info = None + if connected: + try: + account_info = blob_service_client.get_account_information() + connection_info = { + "account_kind": account_info.get('account_kind', 'unknown'), + "sku_name": account_info.get('sku_name', 'unknown'), + "connected_at": str(datetime.now()) + } + except Exception as e: + logger.error(f"获取账户信息失败: {str(e)}") + + return HealthResponse( + status="healthy" if connected else "not_connected", + connected=connected, + framework=AGENT_FRAMEWORK, + user_id=USER_ID, + namespace=NAMESPACE, + connection_info=connection_info + ) + + +@app.post("/connect") +async def connect_to_storage(request: ConnectRequest): + """连接到 Azure Blob Storage""" + global blob_service_client, connection_string + + try: + blob_service_client = BlobServiceClient.from_connection_string( + request.connection_string + ) + + account_info = blob_service_client.get_account_information() + connection_string = request.connection_string + + logger.info(f"✅ 成功连接到 Azure Blob Storage (User: {USER_ID})") + + return { + "status": "connected", + "message": "成功连接到 Azure Blob Storage", + "framework": AGENT_FRAMEWORK, + "user_id": USER_ID, + "account_info": { + "account_kind": account_info.get('account_kind'), + "sku_name": account_info.get('sku_name') + } + } + except Exception as e: + logger.error(f"❌ 连接失败: {str(e)}") + blob_service_client = None + connection_string = None + raise HTTPException(status_code=400, detail=f"连接失败: {str(e)}") + + +@app.get("/mcp/tools") +async def list_mcp_tools(): + """列出所有可用的 MCP 工具""" + if not blob_service_client: + raise HTTPException( + status_code=400, + detail="未连接到 Azure Blob Storage，请先调用 /connect" + ) + + tools_spec = [tool.to_mcp_spec() for tool in mcp_tools.values()] + + return { + "tools": tools_spec, + "count": len(tools_spec), + "framework": AGENT_FRAMEWORK + } + + +@app.post("/mcp/call") +async def call_mcp_tool(request: MCPToolRequest): + """调用 MCP 工具""" + if not blob_service_client: + raise HTTPException( + status_code=400, + detail="未连接到 Azure Blob Storage，请先调用 /connect" + ) + + tool_name = request.tool_name + if tool_name not in mcp_tools: + raise HTTPException( + status_code=404, + detail=f"工具 '{tool_name}' 不存在" + ) + + try: + tool = mcp_tools[tool_name] + result = await tool.execute(request.parameters) + + return { + "tool": tool_name, + "result": result, + "timestamp": str(datetime.now()) + } + except Exception as e: + logger.error(f"工具调用失败: {str(e)}") + raise HTTPException(status_code=500, detail=f"工具调用失败: {str(e)}") + + +@app.post("/query") +async def query_storage(request: MCPQueryRequest): + """使用自然语言查询存储 (简化版 - 实际应集成 LLM)""" + if not blob_service_client: + raise HTTPException( + status_code=400, + detail="未连接到 Azure Blob Storage，请先调用 /connect" + ) + + try: + query = request.query.lower() + result = None + + # 简单的规则匹配 (实际应使用 LLM 进行意图识别) + if "容器" in query and ("列出" in query or "显示" in query or "有哪些" in query): + tool = mcp_tools["list_containers"] + result = await tool.execute({}) + elif "统计" in query or "有多少" in query or "占用" in query: + tool = mcp_tools["get_storage_stats"] + result = await tool.execute({}) + elif request.container_name: + if "文件" in query or "blob" in query.lower(): + tool = mcp_tools["list_blobs"] + result = await tool.execute({"container_name": request.container_name}) + + if result: + return { + "status": "success", + "query": request.query, + "result": result, + "framework": AGENT_FRAMEWORK + } + else: + return { + "status": "info", + "query": request.query, + "message": "未能匹配到合适的工具，请使用 /mcp/tools 查看可用工具", + "available_tools": list(mcp_tools.keys()) + } + except Exception as e: + logger.error(f"查询执行失败: {str(e)}") + raise HTTPException(status_code=500, detail=f"查询失败: {str(e)}") + + +@app.get("/") +async def root(): + """根端点""" + return { + "service": "Azure Blob Storage AI Agent", + "version": "1.0.0", + "framework": AGENT_FRAMEWORK, + "pod_name": POD_NAME, + "template": TEMPLATE_TYPE, + "user_id": USER_ID, + "namespace": NAMESPACE, + "connected": blob_service_client is not None, + "tools_count": len(mcp_tools), + "endpoints": { + "health": "/health", + "connect": "POST /connect", + "list_tools": "GET /mcp/tools", + "call_tool": "POST /mcp/call", + "query": "POST /query" + } + } + + +# ==================== 主函数 ==================== + +def init_storage_connection(): + """启动时初始化存储连接""" + global blob_service_client, connection_string + + if AZURE_STORAGE_CONNECTION_STRING: + try: + logger.info("检测到环境变量中的连接字符串，尝试连接...") + blob_service_client = BlobServiceClient.from_connection_string( + AZURE_STORAGE_CONNECTION_STRING + ) + + account_info = blob_service_client.get_account_information() + connection_string = AZURE_STORAGE_CONNECTION_STRING + + logger.info(f"✅ 成功连接到 Azure Blob Storage") + logger.info(f" - Account Kind: {account_info.get('account_kind')}") + logger.info(f" - SKU: {account_info.get('sku_name')}") + except Exception as e: + logger.error(f"❌ 启动时连接失败: {str(e)}") + logger.info("💡 提示: 可以稍后通过 /connect API 手动连接") + blob_service_client = None + connection_string = None + else: + logger.info("💡 未设置 AZURE_STORAGE_CONNECTION_STRING，需通过 /connect API 手动连接") + + +def main(): + """启动服务""" + logger.info(f"🚀 启动 Azure Blob Storage AI Agent (MCP)") + logger.info(f" - Framework: {AGENT_FRAMEWORK}") + logger.info(f" - Pod名称: {POD_NAME}") + logger.info(f" - 模板类型: {TEMPLATE_TYPE}") + logger.info(f" - User ID: {USER_ID}") + logger.info(f" - Namespace: {NAMESPACE}") + logger.info(f" - 模型: {MODEL_NAME} @ {MODEL_PROVIDER}") + logger.info(f" - 服务地址: http://{SERVICE_HOST}:{SERVICE_PORT}") + + # 注册 MCP 工具 + register_tools() + + # 初始化存储连接 + init_storage_connection() + + uvicorn.run( + app, + host=SERVICE_HOST, + port=SERVICE_PORT, + log_level="info" + ) + + +if __name__ == "__main__": + main() diff --git a/agent_templates/build_azure_blob_a2a.sh b/agent_templates/build_azure_blob_a2a.sh new file mode 100755 index 0000000..c772fc5 --- /dev/null +++ b/agent_templates/build_azure_blob_a2a.sh @@ -0,0 +1,41 @@ +#!/bin/bash + +# 构建并推送 Azure Blob Agent A2A 版本到 ACR +# 用法: ./build_azure_blob_a2a.sh + +set -e + +echo "🚀 构建 Azure Blob Agent (A2A版本)..." + +# Azure Container Registry 配置 +ACR_NAME="agnettaiji" +ACR_LOGIN_SERVER="${ACR_NAME}.azurecr.io" +IMAGE_NAME="ai-agents/azure-blob-agent-a2a" +IMAGE_TAG="latest" + +# 完整镜像名称 +FULL_IMAGE_NAME="${ACR_LOGIN_SERVER}/${IMAGE_NAME}:${IMAGE_TAG}" + +echo "📦 镜像名称: ${FULL_IMAGE_NAME}" + +# 构建镜像 +echo "🔨 构建 Docker 镜像 (ARM64)..." +docker buildx build \ + --platform linux/arm64 \ + -f azure_blob_agent_a2a.Dockerfile \ + -t ${FULL_IMAGE_NAME} \ + --load \ + . + +echo "✅ 镜像构建成功" + +# 登录到 ACR +echo "🔐 登录到 Azure Container Registry..." +az acr login --name ${ACR_NAME} + +# 推送镜像 +echo "📤 推送镜像到 ACR..." +docker push ${FULL_IMAGE_NAME} + +echo "✅ 镜像推送成功" +echo "🎉 完成！镜像: ${FULL_IMAGE_NAME}" diff --git a/agent_templates/build_azure_blob_agent.sh b/agent_templates/build_azure_blob_agent.sh new file mode 100755 index 0000000..506d3a0 --- /dev/null +++ b/agent_templates/build_azure_blob_agent.sh @@ -0,0 +1,88 @@ +#!/bin/bash + +# Azure Blob Storage Agent 构建和推送脚本 +# 使用方法: ./build_azure_blob_agent.sh [TAG] + +set -e + +# 默认配置 +ACR_NAME="${ACR_NAME:-agnettaiji.azurecr.io}" +IMAGE_NAME="ai-agents/azure-blob-agent" +TAG="${1:-latest}" +FULL_IMAGE="${ACR_NAME}/${IMAGE_NAME}:${TAG}" + +echo "==========================================" +echo "构建 Azure Blob Storage Agent" +echo "==========================================" +echo "镜像: ${FULL_IMAGE}" +echo "" + +# 构建镜像 +echo "📦 开始构建镜像..." +docker build \ + -f azure_blob_agent.Dockerfile \ + -t "${FULL_IMAGE}" \ + . + +echo "" +echo "✅ 镜像构建成功: ${FULL_IMAGE}" +echo "" + +# 询问是否推送 +read -p "是否推送到 ACR? (y/N): " -n 1 -r +echo +if [[ $REPLY =~ ^[Yy]$ ]]; then + echo "🚀 推送镜像到 ACR..." + + # 登录 ACR (如果需要) + echo "登录到 ACR..." + az acr login --name $(echo ${ACR_NAME} | cut -d'.' -f1) + + # 推送镜像 + docker push "${FULL_IMAGE}" + + echo "" + echo "✅ 镜像推送成功!" +else + echo "⏭️ 跳过推送" +fi + +echo "" +echo "==========================================" +echo "本地测试命令:" +echo "==========================================" +echo "" +echo "# 启动容器 (需要 LiteLLM 服务)" +echo "docker run -d --name azure-blob-agent \\" +echo " -p 8080:8080 \\" +echo " -e LITELLM_API_BASE=http://host.docker.internal:4000 \\" +echo " -e LITELLM_MODEL=gpt-3.5-turbo \\" +echo " -e LITELLM_API_KEY=sk-1234 \\" +echo " -e AZURE_STORAGE_CONNECTION_STRING='YOUR_CONNECTION_STRING' \\" +echo " ${FULL_IMAGE}" +echo "" +echo "# 或者不提供连接字符串，稍后通过 API 连接" +echo "docker run -d --name azure-blob-agent \\" +echo " -p 8080:8080 \\" +echo " -e LITELLM_API_BASE=http://host.docker.internal:4000 \\" +echo " -e LITELLM_MODEL=gpt-3.5-turbo \\" +echo " -e LITELLM_API_KEY=sk-1234 \\" +echo " ${FULL_IMAGE}" +echo "" +echo "# 检查健康状态" +echo "curl http://localhost:8080/health" +echo "" +echo "# 连接到 Azure Storage" +echo "curl -X POST http://localhost:8080/connect \\" +echo " -H 'Content-Type: application/json' \\" +echo " -d '{\"connection_string\": \"YOUR_CONNECTION_STRING\"}'" +echo "" +echo "# 执行自然语言查询" +echo "curl -X POST http://localhost:8080/query \\" +echo " -H 'Content-Type: application/json' \\" +echo " -d '{\"query\": \"列出所有容器\"}'" +echo "" +echo "# 查看日志" +echo "docker logs -f azure-blob-agent" +echo "" +echo "==========================================" diff --git a/agent_templates/build_azure_blob_mcp.sh b/agent_templates/build_azure_blob_mcp.sh new file mode 100755 index 0000000..ae9662c --- /dev/null +++ b/agent_templates/build_azure_blob_mcp.sh @@ -0,0 +1,41 @@ +#!/bin/bash + +# 构建并推送 Azure Blob Agent MCP 版本到 ACR +# 用法: ./build_azure_blob_mcp.sh + +set -e + +echo "🚀 构建 Azure Blob Agent (MCP版本)..." + +# Azure Container Registry 配置 +ACR_NAME="agnettaiji" +ACR_LOGIN_SERVER="${ACR_NAME}.azurecr.io" +IMAGE_NAME="ai-agents/azure-blob-agent-mcp" +IMAGE_TAG="latest" + +# 完整镜像名称 +FULL_IMAGE_NAME="${ACR_LOGIN_SERVER}/${IMAGE_NAME}:${IMAGE_TAG}" + +echo "📦 镜像名称: ${FULL_IMAGE_NAME}" + +# 构建镜像 +echo "🔨 构建 Docker 镜像 (ARM64)..." +docker buildx build \ + --platform linux/arm64 \ + -f azure_blob_agent_mcp.Dockerfile \ + -t ${FULL_IMAGE_NAME} \ + --load \ + . + +echo "✅ 镜像构建成功" + +# 登录到 ACR +echo "🔐 登录到 Azure Container Registry..." +az acr login --name ${ACR_NAME} + +# 推送镜像 +echo "📤 推送镜像到 ACR..." +docker push ${FULL_IMAGE_NAME} + +echo "✅ 镜像推送成功" +echo "🎉 完成！镜像: ${FULL_IMAGE_NAME}" diff --git a/agent_templates/requirements_a2a.txt b/agent_templates/requirements_a2a.txt new file mode 100644 index 0000000..7960c0a --- /dev/null +++ b/agent_templates/requirements_a2a.txt @@ -0,0 +1,6 @@ +# Requirements for Azure Blob Agent - A2A Version +fastapi==0.109.0 +uvicorn[standard]==0.27.0 +pydantic==2.5.3 +azure-storage-blob==12.19.0 +httpx==0.26.0 diff --git a/agent_templates/requirements_mcp.txt b/agent_templates/requirements_mcp.txt new file mode 100644 index 0000000..f09fc74 --- /dev/null +++ b/agent_templates/requirements_mcp.txt @@ -0,0 +1,5 @@ +# Requirements for Azure Blob Agent - MCP Version +fastapi==0.109.0 +uvicorn[standard]==0.27.0 +pydantic==2.5.3 +azure-storage-blob==12.19.0 diff --git a/agent_templates/test_azure_blob_agent.sh b/agent_templates/test_azure_blob_agent.sh new file mode 100755 index 0000000..11b5199 --- /dev/null +++ b/agent_templates/test_azure_blob_agent.sh @@ -0,0 +1,148 @@ +#!/bin/bash + +# Azure Blob Storage Agent 本地测试脚本 +# 使用方法: ./test_azure_blob_agent.sh + +set -e + +AGENT_HOST="localhost" +AGENT_PORT="8080" +BASE_URL="http://${AGENT_HOST}:${AGENT_PORT}" + +echo "==========================================" +echo "Azure Blob Storage Agent 测试脚本" +echo "==========================================" +echo "" + +# 颜色定义 +GREEN='\033[0;32m' +RED='\033[0;31m' +YELLOW='\033[1;33m' +NC='\033[0m' # No Color + +# 测试函数 +test_endpoint() { + local name=$1 + local method=$2 + local endpoint=$3 + local data=$4 + + echo -e "${YELLOW}测试: ${name}${NC}" + echo "请求: ${method} ${endpoint}" + + if [ -z "$data" ]; then + response=$(curl -s -w "\n%{http_code}" -X ${method} "${BASE_URL}${endpoint}") + else + response=$(curl -s -w "\n%{http_code}" -X ${method} "${BASE_URL}${endpoint}" \ + -H 'Content-Type: application/json' \ + -d "${data}") + fi + + http_code=$(echo "$response" | tail -n1) + body=$(echo "$response" | sed '$d') + + if [ "$http_code" -eq 200 ] || [ "$http_code" -eq 201 ]; then + echo -e "${GREEN}✅ 成功 (HTTP $http_code)${NC}" + echo "响应: $body" | jq '.' 2>/dev/null || echo "$body" + else + echo -e "${RED}❌ 失败 (HTTP $http_code)${NC}" + echo "响应: $body" + fi + + echo "" +} + +# 1. 检查容器是否运行 +echo "1️⃣ 检查容器状态..." +if docker ps | grep -q azure-blob-agent; then + echo -e "${GREEN}✅ 容器正在运行${NC}" +else + echo -e "${RED}❌ 容器未运行${NC}" + echo "请先启动容器：" + echo "docker run -d --name azure-blob-agent -p 8080:8080 \\" + echo " -e LITELLM_API_BASE=http://host.docker.internal:4000 \\" + echo " -e LITELLM_MODEL=gpt-3.5-turbo \\" + echo " -e LITELLM_API_KEY=sk-1234 \\" + echo " azure-blob-agent:latest" + exit 1 +fi +echo "" + +# 2. 等待服务就绪 +echo "2️⃣ 等待服务就绪..." +max_attempts=30 +attempt=0 +while [ $attempt -lt $max_attempts ]; do + if curl -s "${BASE_URL}/health" > /dev/null 2>&1; then + echo -e "${GREEN}✅ 服务已就绪${NC}" + break + fi + attempt=$((attempt + 1)) + echo -n "." + sleep 1 +done + +if [ $attempt -eq $max_attempts ]; then + echo -e "${RED}❌ 服务启动超时${NC}" + exit 1 +fi +echo "" + +# 3. 健康检查 +test_endpoint "健康检查" "GET" "/health" + +# 4. 根端点 +test_endpoint "根端点" "GET" "/" + +# 5. 连接到 Azure Storage（需要用户提供连接字符串） +echo -e "${YELLOW}==========================================" +echo "连接到 Azure Storage" +echo "==========================================${NC}" +echo "" +echo "请输入 Azure Storage 连接字符串：" +echo "(格式: DefaultEndpointsProtocol=https;AccountName=xxx;AccountKey=xxx;EndpointSuffix=core.windows.net)" +echo "" +read -r CONNECTION_STRING + +if [ -z "$CONNECTION_STRING" ]; then + echo -e "${YELLOW}⏭️ 跳过连接测试${NC}" +else + connect_data="{\"connection_string\": \"${CONNECTION_STRING}\"}" + test_endpoint "连接 Azure Storage" "POST" "/connect" "$connect_data" + + # 6. 查询测试（仅在连接成功后） + echo -e "${YELLOW}==========================================" + echo "自然语言查询测试" + echo "==========================================${NC}" + echo "" + + # 列出容器 + query_data='{"query": "列出所有容器"}' + test_endpoint "查询: 列出所有容器" "POST" "/query" "$query_data" + + # 获取统计信息 + query_data='{"query": "显示存储统计信息"}' + test_endpoint "查询: 存储统计" "POST" "/query" "$query_data" + + # 自定义查询 + echo -e "${YELLOW}输入自定义查询（按Enter跳过）:${NC}" + read -r CUSTOM_QUERY + + if [ ! -z "$CUSTOM_QUERY" ]; then + query_data="{\"query\": \"${CUSTOM_QUERY}\"}" + test_endpoint "自定义查询" "POST" "/query" "$query_data" + fi +fi + +echo "" +echo "==========================================" +echo -e "${GREEN}测试完成！${NC}" +echo "==========================================" +echo "" +echo "查看日志：" +echo " docker logs -f azure-blob-agent" +echo "" +echo "停止容器：" +echo " docker stop azure-blob-agent" +echo " docker rm azure-blob-agent" +echo "" diff --git a/agent_templates/test_client.py b/agent_templates/test_client.py new file mode 100755 index 0000000..c4a16bd --- /dev/null +++ b/agent_templates/test_client.py @@ -0,0 +1,164 @@ +#!/usr/bin/env python3 +""" +Azure Blob Storage Agent 客户端示例 +演示如何使用 Python 调用 agent API +""" +import requests +import json +import os +import sys + +# Agent 配置 +AGENT_BASE_URL = os.getenv("AGENT_URL", "http://localhost:8080") + +class AzureBlobAgentClient: + """Azure Blob Storage Agent 客户端""" + + def __init__(self, base_url: str = AGENT_BASE_URL): + self.base_url = base_url.rstrip('/') + self.session = requests.Session() + self.connected = False + + def health_check(self) -> dict: + """健康检查""" + response = self.session.get(f"{self.base_url}/health") + response.raise_for_status() + return response.json() + + def connect(self, connection_string: str) -> dict: + """连接到 Azure Storage""" + response = self.session.post( + f"{self.base_url}/connect", + json={"connection_string": connection_string} + ) + response.raise_for_status() + result = response.json() + self.connected = True + return result + + def query(self, query_text: str, container_name: str = None) -> dict: + """执行自然语言查询""" + if not self.connected: + raise Exception("未连接到 Azure Storage，请先调用 connect()") + + payload = {"query": query_text} + if container_name: + payload["container_name"] = container_name + + response = self.session.post( + f"{self.base_url}/query", + json=payload + ) + response.raise_for_status() + return response.json() + + def get_info(self) -> dict: + """获取 agent 信息""" + response = self.session.get(f"{self.base_url}/") + response.raise_for_status() + return response.json() + + +def print_response(title: str, response: dict): + """格式化打印响应""" + print(f"\n{'='*60}") + print(f"📋 {title}") + print('='*60) + print(json.dumps(response, indent=2, ensure_ascii=False)) + + +def main(): + """主函数""" + print("🚀 Azure Blob Storage Agent 客户端") + print(f"连接到: {AGENT_BASE_URL}\n") + + # 创建客户端 + client = AzureBlobAgentClient() + + try: + # 1. 健康检查 + print("1️⃣ 执行健康检查...") + health = client.health_check() + print_response("健康检查", health) + + # 2. 获取 agent 信息 + print("\n2️⃣ 获取 Agent 信息...") + info = client.get_info() + print_response("Agent 信息", info) + + # 3. 连接到 Azure Storage + print("\n3️⃣ 连接到 Azure Storage...") + + # 从环境变量获取连接字符串 + connection_string = os.getenv("AZURE_STORAGE_CONNECTION_STRING") + + if not connection_string: + print("\n⚠️ 未设置 AZURE_STORAGE_CONNECTION_STRING 环境变量") + print("请输入 Azure Storage 连接字符串:") + connection_string = input().strip() + + if not connection_string: + print("❌ 未提供连接字符串，退出") + sys.exit(1) + + connect_result = client.connect(connection_string) + print_response("连接结果", connect_result) + + # 4. 执行查询 + print("\n4️⃣ 执行自然语言查询...\n") + + queries = [ + "列出所有容器", + "显示存储统计信息", + ] + + for query_text in queries: + print(f"\n💬 查询: {query_text}") + result = client.query(query_text) + print(f"\n✅ 答案:\n{result.get('answer', 'N/A')}") + print(f"\n状态: {result.get('status')}") + + # 5. 交互式查询 + print("\n5️⃣ 交互式查询") + print("="*60) + print("输入自然语言查询（输入 'quit' 或 'exit' 退出）:") + print("例如:") + print(" - 列出所有容器") + print(" - 显示 images 容器中的文件") + print(" - 在 documents 容器中搜索 report") + print(" - 获取存储统计信息") + print("="*60) + + while True: + try: + query_text = input("\n💬 > ").strip() + + if query_text.lower() in ['quit', 'exit', 'q']: + print("👋 再见!") + break + + if not query_text: + continue + + result = client.query(query_text) + print(f"\n✅ 答案:\n{result.get('answer', 'N/A')}") + + except KeyboardInterrupt: + print("\n\n👋 再见!") + break + except Exception as e: + print(f"\n❌ 查询失败: {str(e)}") + + except requests.exceptions.ConnectionError: + print(f"\n❌ 无法连接到 Agent: {AGENT_BASE_URL}") + print("请确保 Agent 正在运行") + sys.exit(1) + except Exception as e: + print(f"\n❌ 错误: {str(e)}") + import traceback + traceback.print_exc() + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/agent_templates/test_multi_framework.sh b/agent_templates/test_multi_framework.sh new file mode 100755 index 0000000..ddb7a63 --- /dev/null +++ b/agent_templates/test_multi_framework.sh @@ -0,0 +1,172 @@ +#!/bin/bash + +# 测试 Azure Blob Agent 多框架版本 +# 用法: ./test_multi_framework.sh + +set -e + +echo "🧪 测试 Azure Blob Agent 多框架版本" +echo "======================================" + +# 配置 +AGENT_MANAGER_URL="http://localhost:8000" +OWNER_ID="test-user" +NAMESPACE="ai-agents" + +# Azure Storage 连接字符串（从环境变量获取） +STORAGE_CONN_STRING="${AZURE_STORAGE_CONNECTION_STRING}" + +if [ -z "$STORAGE_CONN_STRING" ]; then + echo "❌ 错误: 请设置环境变量 AZURE_STORAGE_CONNECTION_STRING" + exit 1 +fi + +# 模型配置（从环境变量获取） +MODEL_API_KEY="${OPENAI_API_KEY:-sk-test}" + +echo "" +echo "📋 配置信息:" +echo " - Agent Manager: $AGENT_MANAGER_URL" +echo " - Owner ID: $OWNER_ID" +echo " - Namespace: $NAMESPACE" +echo "" + +# 测试函数 +test_agent() { + local framework=$1 + local template=$2 + local agent_name=$3 + local extra_config=$4 + + echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━" + echo "🧪 测试 $framework 版本" + echo "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━" + + # 构建请求 JSON + local request_json=$(cat <