最近在尝试将一些AI应用从本地Windows环境迁移到更复杂的分布式Agent网络时，遇到了不少环境配置、网络通信和任务编排的难题。网上资料要么是零散的概念介绍，要么是特定框架的简单Demo，缺乏一个从个人开发环境到多Agent协作网络的完整实战路径。本文旨在填补这一空白，通过一个具体的项目案例，手把手带你搭建一个可运行的AI Agent网络，并探讨其向“超级应用”演进的挑战与可能性。无论你是想了解Agent基础，还是希望将现有AI能力扩展到分布式系统，这篇文章都能提供从环境搭建到核心代码的完整参考。

1. 背景与核心概念：从单机AI到协作网络

在深入代码之前，我们有必要厘清几个关键概念。这有助于理解我们正在构建的是什么，以及它为何重要。

AI Agent（智能体） ： 不同于传统的、被动响应请求的AI模型（如ChatGPT对话），一个真正的AI Agent是一个具有自主性、能够感知环境、做出决策并执行行动以实现目标的软件实体。它通常包含几个核心模块：规划（Planning）、记忆（Memory）、工具使用（Tool Use）和行动（Action）。例如，一个数据分析Agent可以自主决定连接数据库、执行查询、分析结果并生成报告。

Agent网络（多智能体系统） ： 当多个具有不同能力的AI Agent通过某种通信机制（如消息队列、HTTP API、共享状态）连接起来，协同完成一个更复杂的任务时，就形成了一个Agent网络。这类似于人类社会的分工协作。一个Agent可能擅长搜索信息，另一个擅长编写代码，第三个擅长审核结果，它们通过协作可以完成单一个体难以胜任的工作。

从Windows到网络 ： “从Windows到Agent网络”这个说法，象征了AI应用形态的演进。过去，许多AI应用（如基于Python脚本的本地模型推理）是“单机版”的，运行在一台Windows或Linux电脑上，能力有限，扩展性差。而Agent网络代表了一种“云端”或“分布式”的思维，将AI能力服务化、模块化，并通过网络连接，实现弹性伸缩和复杂任务编排。本文的实战部分，就会演示如何将原本可能在Windows本地运行的多个AI功能，改造为部署在网络上、可互相调用的Agent。

AI超级应用 ： 这是一个更前瞻的概念。它指的或许不是一个单一的“杀手级应用”，而是一个由无数细粒度、专业化的AI Agent构成的生态系统。在这个系统中，用户可以通过自然语言提出复杂需求，系统自动分解任务，调度合适的Agent网络协同工作，最终返回结果。它模糊了应用的边界，让AI能力像水电一样按需取用。我们当前的技术正处于向这个愿景迈进的早期阶段，而构建可工作的Agent网络是必经之路。

2. 环境准备与版本说明

我们的实战目标是构建一个简单的多Agent协作系统，它包含一个“任务规划Agent”和一个“代码执行Agent”。规划Agent分析用户需求并拆解步骤，代码执行Agent负责运行安全的代码片段（如数据处理）。我们将使用Python作为主要语言，并借助一些成熟的库来简化开发。

核心环境清单：

• 操作系统 ： 本教程示例在 Windows 11 及 Ubuntu 22.04 LTS 上测试通过。Agent网络的核心思想是跨平台，但部分依赖的安装命令可能略有不同，文中会注明。
• Python ： 版本 >= 3.9。这是大多数AI相关库的基础。建议使用 pyenv 、 conda 或官方安装包管理多版本。
• 关键Python库 ：
  • fastapi >= 0.104.0： 用于快速构建Agent的HTTP API接口。
  • uvicorn >= 0.24.0： 用于运行FastAPI应用的ASGI服务器。
  • pydantic >= 2.0： 用于数据验证和设置管理。
  • requests >= 2.31.0： 用于Agent间的HTTP通信。
  • openai >= 1.0.0： 用于为Agent提供大语言模型（LLM）能力。我们将使用其Chat Completion API。你需要准备一个有效的API Key。
  • python-dotenv >= 1.0.0： 管理环境变量，安全存储API Key。
• 辅助工具 ：
  • Docker (可选但推荐) ： 用于容器化每个Agent，实现环境隔离和便捷部署。确保已安装Docker Desktop (Windows/Mac) 或 Docker Engine (Linux)。
  • Git ： 版本控制。
  • curl 或 Postman ： 用于测试API接口。

版本兼容性说明 ： AI生态迭代迅速，特别是OpenAI等库的版本更新可能带来接口变化。本文代码基于上述版本编写，核心逻辑具有通用性。如果你的环境版本不同，遇到API不兼容时，请参考对应库的官方文档进行调整。

项目结构预览 ： 在开始前，我们先规划一下项目目录，这有助于理解后续的代码文件位置。

ai_agent_network/
├── .env                    # 存储环境变量，如OPENAI_API_KEY
├── planner_agent/          # 任务规划Agent
│   ├── main.py            # FastAPI应用入口
│   ├── agent.py           # 规划Agent的核心逻辑
│   ├── schemas.py         # Pydantic数据模型定义
│   └── requirements.txt   # 该Agent的Python依赖
├── code_executor_agent/   # 代码执行Agent
│   ├── main.py
│   ├── agent.py
│   ├── schemas.py
│   └── requirements.txt
├── client/                # 一个简单的客户端，用于发起请求
│   └── demo_client.py
└── docker-compose.yml     # 使用Docker Compose一键启动所有服务（可选）

3. 核心组件与原理拆解

一个最基本的可通信Agent需要包含哪些部分？我们将其抽象为三个核心组件，并解释其工作原理。

3.1 Agent核心逻辑类

这是Agent的“大脑”。它封装了该Agent的特定能力，例如规划任务或执行代码。它通常包含：

• 初始化 ： 加载配置、模型、工具等。
• 处理函数 ： 接收输入，执行业务逻辑，返回输出。
• 工具集 ： Agent可以调用的函数，如计算器、搜索引擎、数据库查询等。

为什么需要这个类？ 它将Agent的能力与Web服务框架（如FastAPI）解耦，使得Agent逻辑可以独立测试，也更容易被其他方式（如消息队列消费者）调用。

3.2 通信接口（API）

这是Agent的“耳朵和嘴巴”。为了让其他Agent或客户端能找到并调用它，我们需要暴露一个标准的接口。HTTP REST API是最通用和简单的方式。我们使用FastAPI来快速创建这个接口。

关键设计点 ：

• 统一的请求/响应格式 ： 通常使用JSON。我们使用Pydantic来定义数据模型，确保传入传出的数据格式正确。
• 清晰的端点 ： 例如 POST /v1/execute 表示执行任务。
• 异步支持 ： 如果Agent任务耗时较长，使用异步处理（ async/await ）可以避免阻塞，提高并发能力。

3.3 任务编排与通信

这是连接多个Agent的“神经网络”。最简单的编排方式是“链式调用”，即一个Agent完成任务后，手动调用下一个Agent的API。更复杂的系统会引入“协调者”（Orchestrator）或“消息代理”（Message Broker，如RabbitMQ, Redis Streams）。

本文采用的简单模式 ： 客户端先将任务发给规划Agent，规划Agent分析后，生成一个任务列表，然后客户端或规划Agent自身按顺序调用代码执行Agent的API。这种模式直观，适合理解基础原理。

4. 完整实战：构建双Agent协作系统

现在，让我们从零开始，一步步实现这个系统。

4.1 创建项目结构与基础配置

首先，创建项目根目录并初始化虚拟环境。

# 创建项目目录
mkdir ai_agent_network && cd ai_agent_network

# 创建各模块目录
mkdir planner_agent code_executor_agent client

# 在项目根目录创建.env文件，用于存放敏感信息
echo "OPENAI_API_KEY=your_openai_api_key_here" > .env
echo "AGENT_NETWORK_HOST=http://localhost" >> .env

重要 ： 将 your_openai_api_key_here 替换为你自己的OpenAI API Key。 .env 文件已被 .gitignore 忽略（你需要自行创建该文件），切勿提交到版本库。

4.2 实现代码执行Agent

我们先实现功能相对明确的代码执行Agent。它接收一段安全的Python代码（字符串），在一个受限的环境中执行，并返回结果或错误。

步骤1： 定义数据模型 ( code_executor_agent/schemas.py )

from pydantic import BaseModel, Field
from typing import Optional

class CodeExecutionRequest(BaseModel):
    """代码执行请求体"""
    code: str = Field(..., description="要执行的Python代码字符串")
    timeout: Optional[int] = Field(default=5, description="执行超时时间（秒）")

class CodeExecutionResponse(BaseModel):
    """代码执行响应体"""
    success: bool
    result: Optional[str] = Field(default=None, description="执行成功时的输出")
    error: Optional[str] = Field(default=None, description="执行失败时的错误信息")
    execution_time: float = Field(..., description="执行耗时（秒）")

步骤2： 实现Agent核心逻辑 ( code_executor_agent/agent.py )

import subprocess
import sys
import tempfile
import os
import time
from .schemas import CodeExecutionRequest, CodeExecutionResponse

class CodeExecutorAgent:
    """一个在安全子进程中执行代码的Agent"""
    
    def execute(self, request: CodeExecutionRequest) -> CodeExecutionResponse:
        start_time = time.time()
        
        # 创建一个临时文件来存放代码
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
            f.write(request.code)
            temp_file_path = f.name
        
        result_output = ""
        error_output = ""
        success = False
        
        try:
            # 使用subprocess在隔离环境中运行代码，设置超时
            # 注意：这只是一个基础沙箱，对于生产环境，需要更严格的安全限制（如Docker容器、资源限制）
            completed_process = subprocess.run(
                [sys.executable, temp_file_path],
                capture_output=True,
                text=True,
                timeout=request.timeout,
                env={**os.environ, 'PYTHONPATH': ''} # 清空PYTHONPATH以限制模块导入
            )
            
            if completed_process.returncode == 0:
                success = True
                result_output = completed_process.stdout
            else:
                error_output = completed_process.stderr
                
        except subprocess.TimeoutExpired:
            error_output = f"Code execution timed out after {request.timeout} seconds."
        except Exception as e:
            error_output = f"An unexpected error occurred: {str(e)}"
        finally:
            # 清理临时文件
            os.unlink(temp_file_path)
        
        execution_time = time.time() - start_time
        
        return CodeExecutionResponse(
            success=success,
            result=result_output.strip() if success else None,
            error=error_output.strip() if not success else None,
            execution_time=execution_time
        )

步骤3： 创建FastAPI应用 ( code_executor_agent/main.py )

from fastapi import FastAPI, HTTPException
from .agent import CodeExecutorAgent
from .schemas import CodeExecutionRequest, CodeExecutionResponse
import uvicorn

app = FastAPI(title="Code Executor Agent", version="1.0.0")
agent = CodeExecutorAgent()

@app.post("/v1/execute", response_model=CodeExecutionResponse)
async def execute_code(request: CodeExecutionRequest):
    """执行代码的端点"""
    try:
        response = agent.execute(request)
        return response
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Agent internal error: {str(e)}")

@app.get("/health")
async def health_check():
    """健康检查端点"""
    return {"status": "healthy", "service": "code_executor_agent"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8001) # 为这个Agent分配端口8001

步骤4： 编写依赖文件 ( code_executor_agent/requirements.txt )

fastapi==0.104.1
uvicorn[standard]==0.24.0
pydantic==2.5.0
python-dotenv==1.0.0

4.3 实现任务规划Agent

规划Agent负责理解用户目标，并将其分解为可执行步骤。这里我们利用大语言模型（LLM）的推理能力。

步骤1： 定义数据模型 ( planner_agent/schemas.py )

from pydantic import BaseModel, Field
from typing import List, Optional

class PlanStep(BaseModel):
    """计划中的一个步骤"""
    agent_name: str = Field(..., description="执行此步骤的Agent名称，如 ‘code_executor‘")
    action: str = Field(..., description="要执行的动作描述")
    input_data: dict = Field(default_factory=dict, description="传递给该Agent的输入参数")

class PlanningRequest(BaseModel):
    """规划请求体"""
    user_objective: str = Field(..., description="用户的原始目标描述")
    available_agents: List[str] = Field(default_factory=lambda: ["code_executor"], description="当前可用的Agent列表")

class PlanningResponse(BaseModel):
    """规划响应体"""
    steps: List[PlanStep] = Field(..., description="分解后的步骤计划")
    reasoning: Optional[str] = Field(default=None, description="模型的推理过程")

步骤2： 实现Agent核心逻辑 ( planner_agent/agent.py )

import os
from openai import OpenAI
from dotenv import load_dotenv
from .schemas import PlanningRequest, PlanningResponse, PlanStep
import json

# 加载环境变量，获取OpenAI API Key
load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), '../../.env'))

class PlanningAgent:
    """利用LLM进行任务规划的Agent"""
    
    def __init__(self):
        api_key = os.getenv("OPENAI_API_KEY")
        if not api_key:
            raise ValueError("OPENAI_API_KEY not found in environment variables.")
        self.client = OpenAI(api_key=api_key)
        self.model = "gpt-3.5-turbo" # 可根据需要切换为 gpt-4 等
        
    def plan(self, request: PlanningRequest) -> PlanningResponse:
        """根据用户目标和可用Agent列表，生成执行计划"""
        
        # 构建给LLM的提示词（Prompt）
        system_prompt = """你是一个任务规划专家。你的目标是将用户复杂的自然语言请求，分解为一系列可以由特定AI Agent执行的步骤。
        每个步骤必须指定使用哪个Agent，以及给该Agent的明确输入。
        当前可用的Agent有：{agents}。
        请以JSON格式输出，包含一个‘steps‘数组，每个元素是一个对象，包含‘agent_name‘, ‘action‘, ‘input_data‘字段。
        在‘input_data‘中，如果涉及代码执行，请将代码放在‘code‘字段里。
        同时，请提供简要的推理过程‘reasoning‘。
        """.format(agents=", ".join(request.available_agents))
        
        user_prompt = f"用户目标：{request.user_objective}"
        
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_prompt}
                ],
                temperature=0.2, # 较低的温度使输出更确定
                response_format={"type": "json_object"} # 要求返回JSON
            )
            
            llm_output = response.choices[0].message.content
            result_dict = json.loads(llm_output)
            
            # 将JSON解析为我们的数据模型
            steps = [PlanStep(**step) for step in result_dict.get("steps", [])]
            reasoning = result_dict.get("reasoning")
            
            return PlanningResponse(steps=steps, reasoning=reasoning)
            
        except json.JSONDecodeError as e:
            raise ValueError(f"Failed to parse LLM output as JSON: {llm_output}. Error: {e}")
        except Exception as e:
            raise RuntimeError(f"Error during planning with LLM: {str(e)}")

步骤3： 创建FastAPI应用 ( planner_agent/main.py )

from fastapi import FastAPI, HTTPException
from .agent import PlanningAgent
from .schemas import PlanningRequest, PlanningResponse
import uvicorn

app = FastAPI(title="Planning Agent", version="1.0.0")
agent = PlanningAgent()

@app.post("/v1/plan", response_model=PlanningResponse)
async def create_plan(request: PlanningRequest):
    """创建执行计划的端点"""
    try:
        response = agent.plan(request)
        return response
    except ValueError as e:
        raise HTTPException(status_code=400, detail=f"Invalid request or LLM output: {str(e)}")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Agent internal error: {str(e)}")

@app.get("/health")
async def health_check():
    return {"status": "healthy", "service": "planning_agent"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000) # 规划Agent使用端口8000

步骤4： 编写依赖文件 ( planner_agent/requirements.txt )

fastapi==0.104.1
uvicorn[standard]==0.24.0
pydantic==2.5.0
openai==1.6.1
python-dotenv==1.0.0
requests==2.31.0 # 用于后续调用其他Agent

4.4 编写客户端进行端到端测试

现在，我们创建一个客户端脚本，模拟用户发起请求，并驱动整个Agent网络工作。

客户端脚本 ( client/demo_client.py )

import requests
import json
import time
from typing import Dict, Any

# 定义两个Agent的地址（假设都在本地运行）
PLANNER_AGENT_URL = "http://localhost:8000/v1/plan"
CODE_EXECUTOR_AGENT_URL = "http://localhost:8001/v1/execute"

def call_planner(user_objective: str) -> Dict[str, Any]:
    """调用规划Agent"""
    print(f"\n[Client] 用户目标: {user_objective}")
    payload = {
        "user_objective": user_objective,
        "available_agents": ["code_executor"]
    }
    print("[Client] 正在请求规划Agent生成计划...")
    try:
        resp = requests.post(PLANNER_AGENT_URL, json=payload, timeout=30)
        resp.raise_for_status()
        plan = resp.json()
        print(f"[Client] 规划完成。推理: {plan.get('reasoning')}")
        print(f"[Client] 生成步骤数: {len(plan['steps'])}")
        for i, step in enumerate(plan['steps']):
            print(f"  步骤{i+1}: 使用Agent [{step['agent_name']}] 执行 [{step['action']}]")
        return plan
    except requests.exceptions.RequestException as e:
        print(f"[Client] 调用规划Agent失败: {e}")
        return {}

def execute_plan(plan: Dict[str, Any]):
    """按计划执行每个步骤"""
    if not plan or 'steps' not in plan:
        print("[Client] 无效的计划，终止执行。")
        return
    
    for i, step in enumerate(plan['steps']):
        print(f"\n[Client] 正在执行步骤 {i+1}...")
        agent_name = step['agent_name']
        input_data = step['input_data']
        
        if agent_name == "code_executor":
            # 调用代码执行Agent
            code_to_run = input_data.get('code', '')
            print(f"[Client] 发送代码给代码执行Agent:\n```python\n{code_to_run}\n```")
            payload = {"code": code_to_run, "timeout": 10}
            try:
                resp = requests.post(CODE_EXECUTOR_AGENT_URL, json=payload, timeout=30)
                resp.raise_for_status()
                result = resp.json()
                if result['success']:
                    print(f"[Client] 代码执行成功！结果:\n{result['result']}")
                    print(f"[Client] 耗时: {result['execution_time']:.2f}秒")
                else:
                    print(f"[Client] 代码执行失败！错误:\n{result['error']}")
            except requests.exceptions.RequestException as e:
                print(f"[Client] 调用代码执行Agent失败: {e}")
        else:
            print(f"[Client] 未知的Agent: {agent_name}，跳过此步骤。")
        time.sleep(0.5) # 短暂间隔，便于观察

if __name__ == "__main__":
    # 测试用例1：一个简单的数据处理任务
    objective_1 = "请计算斐波那契数列的前10个数字，并将结果以列表形式输出。"
    plan_1 = call_planner(objective_1)
    execute_plan(plan_1)
    
    # 测试用例2：一个需要多步代码的任务（规划Agent可能会拆解）
    # objective_2 = “首先生成一个包含10个随机数的列表，然后计算它们的平均值和总和，最后把结果打印出来。”
    # plan_2 = call_planner(objective_2)
    # execute_plan(plan_2)

4.5 运行与验证

现在，让我们启动整个系统并观察其运行。

步骤1： 启动两个Agent服务 打开两个终端窗口，分别进入对应目录，安装依赖并启动服务。

终端1 (规划Agent) :

cd path/to/ai_agent_network/planner_agent
pip install -r requirements.txt
python main.py

服务启动后，访问 http://localhost:8000/docs 可以看到自动生成的API文档。

终端2 (代码执行Agent) :

cd path/to/ai_agent_network/code_executor_agent
pip install -r requirements.txt
python main.py

访问 http://localhost:8001/docs 查看其API文档。

步骤2： 运行客户端 打开第三个终端窗口。

cd path/to/ai_agent_network/client
python demo_client.py

预期输出示例：

[Client] 用户目标: 请计算斐波那契数列的前10个数字，并将结果以列表形式输出。
[Client] 正在请求规划Agent生成计划...
[Client] 规划完成。推理: 用户请求计算斐波那契数列的前10个数字并以列表输出。这需要一个能执行Python代码的Agent。我将生成一段计算斐波那契数列的代码。
[Client] 生成步骤数: 1
  步骤1: 使用Agent [code_executor] 执行 [Calculate the first 10 Fibonacci numbers and output as a list]

[Client] 正在执行步骤 1...
[Client] 发送代码给代码执行Agent:
```python
def fibonacci(n):
    fib_sequence = [0, 1]
    for i in range(2, n):
        fib_sequence.append(fib_sequence[-1] + fib_sequence[-2])
    return fib_sequence[:n]

print(fibonacci(10))

[Client] 代码执行成功！结果: [0, 1, 1, 2, 3, 5, 8, 13, 21, 34] [Client] 耗时: 0.12秒

恭喜！你已经成功运行了一个由两个AI Agent通过网络协作完成的简单任务。规划Agent理解了你的自然语言请求，并将其转化为一段可执行的Python代码，然后代码执行Agent安全地运行了这段代码并返回了结果。

## 5. 常见问题与排查思路

在搭建和运行此类Agent系统时，你可能会遇到以下典型问题。

| 问题现象 | 常见原因 | 解决思路 |
| :--- | :--- | :--- |
| **启动Agent服务时提示端口被占用** | 端口8000或8001已被其他程序使用。 | 1. 修改 `main.py` 中 `uvicorn.run` 的 `port` 参数。<br>2. 使用命令 `netstat -ano \| findstr :8000` (Windows) 或 `lsof -i :8000` (Linux/Mac) 查找占用进程并终止。 |
| **调用规划Agent返回400或422错误** | 请求体不符合Pydantic模型定义，或LLM返回的JSON无法解析。 | 1. 检查客户端发送的JSON格式，确保包含 `user_objective` 等必需字段。<br>2. 查看规划Agent日志，检查LLM返回的原始内容，调整提示词（Prompt）使其输出更稳定的JSON。 |
| **代码执行Agent执行超时或报错** | 1. 执行的代码本身有语法错误或逻辑错误。<br>2. 代码包含危险操作（如无限循环、导入受限模块）。<br>3. 系统资源不足。 | 1. 检查规划Agent生成的代码是否正确。可以在本地Python环境中先测试。<br>2. 强化 `CodeExecutorAgent` 的安全沙箱，例如使用 `docker run` 在独立容器中运行代码，并设置CPU/内存限制。<br>3. 增加 `timeout` 参数。 |
| **OpenAI API调用失败** | 1. API Key未设置或错误。<br>2. 网络连接问题。<br>3. 账户余额不足或速率限制。 | 1. 确认 `.env` 文件中的 `OPENAI_API_KEY` 正确，且被成功加载。<br>2. 检查网络代理设置。<br>3. 登录OpenAI平台检查用量和余额。 |
| **Agent间通信失败** | 1. 目标Agent服务未启动。<br>2. 客户端中配置的URL或端口错误。<br>3. 防火墙或安全组策略阻止了连接。 | 1. 使用 `curl http://localhost:8000/health` 测试每个Agent的健康端点。<br>2. 确保客户端代码中的URL与Agent实际监听的地址一致。<br>3. 检查本地防火墙或Docker网络设置。 |
| **规划步骤不合理** | LLM的提示词（Prompt）不够精确，导致对任务拆解或Agent选择有误。 | 1. 优化 `PlanningAgent` 中的 `system_prompt`，更清晰地描述每个Agent的能力和输入格式。<br>2. 在 `PlanningRequest` 中提供更详细的 `available_agents` 描述，而不仅仅是名字。<br>3. 考虑使用更强大的模型（如GPT-4）或在提示词中加入少量示例（Few-shot）。 |

## 6. 进阶：工程化与最佳实践

上面的示例是一个最小可行产品（MVP）。要将其用于更严肃的项目或向“超级应用”迈进，需要考虑以下工程化实践。

### 6.1 通信与编排升级
*   **消息队列**： 使用RabbitMQ、Redis Streams或Apache Kafka作为Agent间的通信总线。Agent作为生产者/消费者，实现解耦、异步处理和流量削峰。
*   **专用协调者**： 引入一个独立的“协调者”服务，负责接收用户任务，管理整个工作流（Workflow）的状态，根据计划调用相应Agent，并处理错误和重试。像**LangGraph**、**AutoGen** 这类框架就是为此而生。
*   **服务发现与注册**： 当Agent数量增多时，手动配置URL不可行。需要像Consul、Etcd或Nacos这样的服务发现机制，让Agent能动态地找到彼此。

### 6.2 安全性强化
*   **代码沙箱**： 当前的 `subprocess` 方案非常脆弱。生产环境必须使用更隔离的环境，如：
    ```python
    # 使用Docker运行代码（概念示例）
    docker run --rm -v /tmp/code.py:/code.py:ro --memory="100m" --cpus="0.5" python:3.9-slim python /code.py
    ```
*   **输入验证与过滤**： 对传入的代码进行严格的静态分析，禁止导入 `os`, `subprocess`, `sys` 等危险模块，或使用如 `RestrictedPython` 这样的工具。
*   **API认证与授权**： 为Agent的API接口添加API Key、JWT Token等认证机制，防止未授权访问。

### 6.3 可观测性与稳定性
*   **集中式日志**： 使用ELK（Elasticsearch, Logstash, Kibana）或Loki+Grafana收集所有Agent的日志，便于排查问题。
*   **指标监控**： 为每个Agent暴露Prometheus格式的指标（如请求数、耗时、错误率），并用Grafana展示。
*   **分布式追踪**： 使用Jaeger或Zipkin来追踪一个用户请求在所有Agent间的调用链路，直观看到性能瓶颈。
*   **优雅上下线与健康检查**： 实现`/health`、`/ready`端点，并在编排器中集成健康检查，实现故障转移。

### 6.4 性能与扩展性
*   **Agent无状态化**： 确保Agent本身不保存会话状态，状态由外部数据库或协调者管理。这样便于水平扩展。
*   **异步处理**： 对于耗时任务，FastAPI端点应使用`async def`，并在Agent内部使用`asyncio`或任务队列，避免阻塞。
*   **模型缓存与批处理**： 如果多个规划Agent实例频繁调用LLM，可以考虑共享一个模型缓存，或将多个小请求批处理成一个大的API调用以节省成本。

## 7. 总结：从原型到超级应用之路

通过本文的实践，我们完成了一个微型的、但五脏俱全的AI Agent网络。它演示了从自然语言需求到最终结果的全自动流程，涵盖了规划、执行、网络通信等核心环节。

**我们掌握了什么？**
1.  **Agent的核心概念**： 理解了作为具有自主性、工具使用能力的智能体与普通API服务的区别。
2.  **多Agent系统架构**： 体验了通过HTTP API进行Agent间通信的简单编排模式。
3.  **关键技术栈**： 学会了使用FastAPI构建Agent服务，利用OpenAI API赋予Agent规划能力，以及如何安全地执行动态代码。
4.  **完整开发流程**： 从环境搭建、模块设计、代码实现到联调测试的完整闭环。

**通向“超级应用”的挑战**：
本文的示例距离理想的“AI超级应用”还有很长的路，主要挑战在于：
*   **复杂性管理**： 当Agent数量成百上千时，编排、监控、调试的复杂度呈指数级增长。
*   **可靠性**： 如何保证长链条任务中某个Agent的失败不会导致整个任务崩溃？需要完善的错误处理、重试和补偿机制。
*   **评估与优化**： 如何定量评估整个Agent网络的任务完成质量？如何优化提示词和Agent协作策略？
*   **成本控制**： 大量调用LLM和计算资源，成本可能很高。需要精细化的流量控制和优化策略。

**下一步可以做什么？**
*   **探索成熟框架**： 深入学习 **LangChain**、**LangGraph**、**AutoGen**、**CrewAI** 等框架，它们提供了更强大的Agent模板、工具集成和编排能力。
*   **集成更多工具**： 为你的Agent增加搜索、数据库查询、文件操作、调用第三方API等能力。
*   **尝试本地模型**： 使用Ollama、LM Studio等工具部署本地LLM（如Llama 3、Qwen），降低对云端API的依赖和成本。
*   **设计复杂工作流**： 尝试实现一个需要多个Agent循环协作、有条件分支的任务，例如“调研某个主题并撰写一份报告”。

构建AI Agent网络就像搭建乐高。本文为你提供了最基础的那些积木块——规划器、执行器、通信接口。如何用这些积木，结合更强大的框架和工具，搭建出能解决实际复杂问题的“超级应用”，将是接下来更令人兴奋的旅程。动手修改代码，添加新的Agent，解决你遇到的实际问题，是学习这一切的最佳方式。