在开发者社区,空谈趋势不如上手实践。AI Agent 的颠覆性不仅在于概念,更在于其能直接写入开发流程的工具属性。本文将通过 ** 多智能体协作开发一个 "用户行为分析 API"** 的实战案例,带开发者直观感受 "人类指挥 + Agent 执行" 的全新工作流,并提供可直接复用的代码框架。

一、核心场景:用 3 个 Agent 协作开发 API 接口

假设需求是:构建一个接收用户行为日志(点击 / 停留 / 跳转)的 API,需完成数据校验、存储、实时统计(UV/PV)三个核心功能。传统开发需前端传参设计→后端接口编写→数据库建模→缓存优化等多环节,而多智能体协作可将流程压缩至 "需求定义→结果验收" 两步。

我们将部署 3 个各司其职的 Agent:

  • 需求解析 Agent:将自然语言需求转化为技术规格(API 文档、数据模型、校验规则)
  • 代码生成 Agent:基于规格生成可运行的 Python 代码(FastAPI+SQLAlchemy+Redis)
  • 测试优化 Agent:自动生成测试用例、压测并优化性能(如添加连接池、缓存策略)

二、实战代码:多智能体协作框架搭建

以下是基于开源框架AutoGen(微软推出的多智能体协作工具)的核心实现,开发者可直接替换user需求快速复用。

1. 环境准备

# 安装依赖(需Python 3.10+)
pip install pyautogen fastapi uvicorn sqlalchemy redis pytest locust

2. 智能体定义与协作逻辑

from autogen import Agent, UserProxyAgent, config_list_from_json
from typing import List, Dict

# 1. 配置LLM(支持GPT-4o/DeepSeek-R1等强推理模型)
config_list = config_list_from_json(env_or_file="OAI_CONFIG_LIST")  # 配置文件含API密钥

# 2. 定义需求解析Agent(擅长将自然语言转为技术规格)
spec_agent = Agent(
    name="需求解析专家",
    system_message="""你是资深技术分析师,需将用户需求转化为:
    - API文档(路径、方法、参数、返回值)
    - 数据模型(表结构、字段类型、索引)
    - 校验规则(必填项、格式限制、业务约束)
    输出格式需结构化(如Markdown表格),方便代码生成Agent直接解析。""",
    llm_config={"config_list": config_list}
)

# 3. 定义代码生成Agent(专注后端开发)
code_agent = Agent(
    name="代码生成工程师",
    system_message="""你是FastAPI专家,根据需求解析Agent输出的规格,生成:
    - main.py(API接口实现)
    - models.py(数据库模型)
    - requirements.txt(依赖清单)
    代码必须可运行,包含异常处理、日志记录,优先使用异步操作提升性能。""",
    llm_config={"config_list": config_list}
)

# 4. 定义测试优化Agent(负责质量保障)
test_agent = Agent(
    name="测试优化专家",
    system_message="""你需完成三件事:
    1. 生成pytest测试用例(覆盖正常/异常场景)
    2. 用locust编写压测脚本(模拟1000并发用户)
    3. 输出优化建议(如添加Redis缓存、数据库连接池配置)
    最终提交可直接运行的测试代码和优化后的程序。""",
    llm_config={"config_list": config_list}
)

# 5. 用户代理(人类仅需输入需求,无需干预中间流程)
user_proxy = UserProxyAgent(
    name="人类开发者",
    system_message="仅提供需求,验收最终结果",
    code_execution_config={"work_dir": "user_behavior_api"},  # 代码输出目录
    human_input_mode="NEVER"  # 全程自动运行,无需人工输入
)

# 6. 定义协作流程(链式调用)
def run_workflow(user需求: str):
    # 第一步:需求解析
    spec_message = user_proxy.initiate_chat(
        spec_agent,
        message=user需求
    )
    
    # 第二步:代码生成(基于解析结果)
    code_message = user_proxy.initiate_chat(
        code_agent,
        message=f"请根据以下规格生成代码:\n{spec_message}"
    )
    
    # 第三步:测试优化(基于生成的代码)
    final_result = user_proxy.initiate_chat(
        test_agent,
        message=f"请测试并优化以下代码:\n{code_message}"
    )
    
    return final_result

# 运行流程(输入用户需求)
if __name__ == "__main__":
    user需求 = """
    构建用户行为分析API:
    1. 接收用户ID(user_id)、行为类型(action: click/停留/跳转)、页面路径(path)、时间戳(timestamp)
    2. 需校验user_id不为空、action在允许范围内、timestamp格式正确
    3. 数据存入MySQL,同时用Redis统计当天UV(去重用户数)和PV(总访问量)
    4. 提供一个GET接口 /stats 返回当天UV和PV
    """
    result = run_workflow(user需求)
    print("最终成果:\n", result)

三、Agent 输出物解析:从代码到测试的全链路交付

运行上述代码后,user_behavior_api目录会生成可直接部署的完整项目,以下是核心输出物解析:

1. 需求解析 Agent 生成的 API 规格(节选)

接口路径 方法 参数名 类型 校验规则
/log POST user_id str 非空,长度≤32
/log POST action str 必须为 click / 停留 / 跳转
/stats GET - - -

数据模型:

CREATE TABLE user_behavior (
    id INT AUTO_INCREMENT PRIMARY KEY,
    user_id VARCHAR(32) NOT NULL,
    action ENUM('click', '停留', '跳转') NOT NULL,
    path VARCHAR(255) NOT NULL,
    timestamp DATETIME NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_user_time (user_id, timestamp)
);

2. 代码生成 Agent 的核心实现(main.py 节选)

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, validator
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
import redis
import datetime

app = FastAPI()

# 数据库配置(异步连接)
engine = create_async_engine("mysql+aiomysql://user:pass@localhost/db")
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

# Redis配置(统计UV/PV)
r = redis.Redis(host='localhost', port=6379, db=0)

# 数据模型(Pydantic校验)
class BehaviorLog(BaseModel):
    user_id: str
    action: str
    path: str
    timestamp: datetime.datetime

    @validator('action')
    def action_must_be_valid(cls, v):
        if v not in ['click', '停留', '跳转']:
            raise ValueError('action必须是click/停留/跳转')
        return v

    @validator('user_id')
    def user_id_cannot_be_empty(cls, v):
        if not v.strip():
            raise ValueError('user_id不能为空')
        return v

# 接口实现
@app.post("/log")
async def receive_log(log: BehaviorLog):
    # 1. 写入数据库
    async with AsyncSessionLocal() as db:
        db.execute(
            "INSERT INTO user_behavior (user_id, action, path, timestamp) VALUES (:u, :a, :p, :t)",
            {"u": log.user_id, "a": log.action, "p": log.path, "t": log.timestamp}
        )
        await db.commit()
    
    # 2. 更新Redis统计(UV用集合去重,PV用计数器)
    today = datetime.date.today().isoformat()
    r.sadd(f"uv:{today}", log.user_id)  # UV统计
    r.incr(f"pv:{today}")  # PV统计
    
    return {"status": "success"}

@app.get("/stats")
async def get_stats():
    today = datetime.date.today().isoformat()
    uv = r.scard(f"uv:{today}")  # 获取去重用户数
    pv = r.get(f"pv:{today}") or 0
    return {"uv": uv, "pv": int(pv)}

3. 测试优化 Agent 的输出(性能优化建议)

# 压测脚本(locustfile.py)
from locust import HttpUser, task, between

class BehaviorUser(HttpUser):
    wait_time = between(0.1, 0.5)  # 模拟用户操作间隔

    @task(3)  # 权重3,更频繁
    def send_click_log(self):
        self.client.post("/log", json={
            "user_id": f"test_{self.user_id}",
            "action": "click",
            "path": "/home",
            "timestamp": "2025-11-03T12:00:00"
        })

    @task(1)
    def get_stats(self):
        self.client.get("/stats")

优化建议(已自动写入代码):

  • 数据库连接池:engine = create_async_engine(..., pool_size=20, max_overflow=10)
  • Redis 缓存过期:r.expire(f"uv:{today}", 86400*7)(保留 7 天数据)
  • 接口限流:添加slowapi中间件,限制单 IP 每秒 50 次请求

四、开发者价值:从 "写代码" 到 "定义协作规则"

这个案例揭示了 AI Agent 时代的开发逻辑转变:

  1. 效率提升:传统需 2-3 小时的 API 开发,Agent 协作仅需 15 分钟(含测试),且代码规范度、异常处理覆盖率超人工平均水平
  2. 能力迁移:开发者的核心工作从 "实现功能" 变为 "定义智能体的分工边界"(如修改system_message约束 Agent 行为)
  3. 扩展可能:可通过添加 "安全扫描 Agent"(集成 OWASP 规则)、"文档 Agent"(自动生成 Swagger)进一步完善流程

注意:当前 Agent 生成的代码在复杂业务逻辑(如分布式事务)上仍需人工介入,但 80% 的常规开发工作已可托管。

五、工具链扩展:2025 年必知的 Agent 开发生态

除了AutoGen,这些工具能帮开发者快速落地 Agent 工作流:

  • LangChain Agents:适合需要深度定制工具调用逻辑的场景(如接入私有代码库)
  • MetaGPT:按 "产品经理→架构师→开发者" 的角色拆分 Agent,模拟完整开发流程
  • 华为 AgentStudio:支持本地化部署,解决企业数据隐私问题,提供可视化协作编辑器

开发者可根据场景选择:轻量需求用AutoGen,复杂项目用MetaGPT,企业级部署用AgentStudio

结语:代码即指令,Agent 即团队

当 AI Agent 能像团队成员一样理解需求、编写代码、执行测试时,软件开发的效率瓶颈将被彻底打破。对开发者而言,真正的竞争力不再是 "写代码的速度",而是 "定义协作规则的智慧"—— 知道如何拆分任务、约束 Agent 行为、把控最终质量。

不妨从本文的代码框架开始,用一个实际需求测试 AI Agent 的能力边界。或许你会发现,未来的开发团队,可能是 "1 个开发者 + N 个 Agent" 的最小作战单元。

Logo

火山引擎开发者社区是火山引擎打造的AI技术生态平台,聚焦Agent与大模型开发,提供豆包系列模型(图像/视频/视觉)、智能分析与会话工具,并配套评测集、动手实验室及行业案例库。社区通过技术沙龙、挑战赛等活动促进开发者成长,新用户可领50万Tokens权益,助力构建智能应用。

更多推荐