12-Factor Agents

概述

12-Factor Agent 方法论提供了一套构建可扩展、可维护、可移植的智能体 AI 应用的最佳实践。该方法论脱胎于 12-Factor App 理念，旨在确保智能体能够在不同环境中保持一致的部署行为。

十二要素

I. 代码库

一份受版本控制的代码库，多处部署

• 使用版本控制（Git）管理智能体代码
• 针对不同环境采用分支策略
• 将同一份代码库部署至开发、预发布和生产环境
• 避免出现特定于某一环境的代码分支

# Example: Agent configuration structure
agent/
├── src/           # Core agent logic
├── config/        # Environment configurations
├── tests/         # Test suites
└── deploy/        # Deployment scripts

II. 依赖

显式声明并隔离依赖

• 使用依赖管理工具（pip、npm、poetry）
• 锁定 LLM API 及框架的具体版本号
• 通过虚拟环境或容器隔离依赖
• 不依赖系统全局安装的软件包

# requirements.txt
langchain==0.1.0
openai==1.3.0
pinecone-client==2.2.4

III. 配置

将配置存储在环境中

• 将配置与代码分离
• 通过环境变量管理 API 密钥和端点地址
• 不将密钥提交至版本控制系统
• 为不同环境提供独立的配置支持

import os

class AgentConfig:
    OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
    MODEL_NAME = os.getenv('MODEL_NAME', 'gpt-4')
    TEMPERATURE = float(os.getenv('TEMPERATURE', '0.7'))

IV. 后端服务

将后端服务视为附加资源

• 通过 URL 连接数据库、API 及各类服务
• 不区分本地服务与第三方服务
• 使用服务发现和配置管理机制
• 支持轻松替换服务实现

# Service abstraction
class VectorStore:
    def __init__(self, connection_string):
        self.client = self._create_client(connection_string)

    def _create_client(self, url):
        if url.startswith('pinecone://'):
            return PineconeClient(url)
        elif url.startswith('weaviate://'):
            return WeaviateClient(url)

V. 构建、发布、运行

严格分离构建与运行阶段

• 构建：将代码转换为可执行产物
• 发布：将构建产物与配置合并
• 运行：在运行时环境中执行智能体
• 使用带有唯一标识符的不可变发布版本

# Build stage
docker build -t agent:v1.2.3 .

# Release stage
docker tag agent:v1.2.3 registry/agent:v1.2.3
docker push registry/agent:v1.2.3

# Run stage
docker run -e OPENAI_API_KEY=$API_KEY registry/agent:v1.2.3

VI. 进程

以一个或多个无状态进程运行智能体

• 将智能体设计为无状态
• 将持久化数据存储至后端服务
• 采用无共享（shared-nothing）架构
• 通过进程复制实现水平扩展

class StatelessAgent:
    def __init__(self, config):
        self.llm = LLM(config.model_name)
        self.memory = ExternalMemory(config.memory_url)

    def process_request(self, request):
        # No local state - all data from request or external services
        context = self.memory.get_context(request.session_id)
        response = self.llm.generate(request.prompt, context)
        self.memory.store_interaction(request.session_id, request, response)
        return response

VII. 端口绑定

通过端口绑定对外暴露服务

• 智能体应自包含，并通过端口暴露服务
• 使用 Web 框架对外提供 HTTP API
• 支持服务间通信
• 兼容负载均衡与服务发现机制

from flask import Flask, request, jsonify

app = Flask(__name__)
agent = Agent()

@app.route('/chat', methods=['POST'])
def chat():
    message = request.json['message']
    response = agent.process(message)
    return jsonify({'response': response})

if __name__ == '__main__':
    port = int(os.getenv('PORT', 8000))
    app.run(host='0.0.0.0', port=port)

VIII. 并发

通过进程模型横向扩展

• 通过运行多个智能体进程实现扩展
• 针对不同工作负载类型使用进程管理器
• 实现合理的资源隔离
• 面向水平扩展进行设计

# Process types
web: gunicorn app:app --workers 4
worker: celery worker -A agent.tasks
scheduler: celery beat -A agent.tasks

IX. 易处理性

以快速启动与优雅关闭保障系统健壮性

• 最小化启动时间，以便快速扩容
• 优雅处理 SIGTERM 信号，实现干净关闭
• 遵循"崩溃即重启"（crash-only software）设计原则
• 实现合理的清理流程

import signal
import sys

class Agent:
    def __init__(self):
        self.running = True
        signal.signal(signal.SIGTERM, self._shutdown_handler)

    def _shutdown_handler(self, signum, frame):
        print("Received shutdown signal, cleaning up...")
        self.running = False
        self._cleanup()
        sys.exit(0)

    def _cleanup(self):
        # Close connections, save state, etc.
        pass

X. 开发与生产环境一致性

尽量保持开发、预发布与生产环境的一致

• 最小化各环境之间的差异
• 跨环境使用相同的后端服务
• 频繁部署以降低发布风险
• 使用容器化手段保证环境一致性

# Dockerfile for consistent environments
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .
CMD ["python", "agent.py"]

XI. 日志

将日志视为事件流

• 将日志写入 stdout/stderr
• 使用结构化日志（JSON 格式）
• 通过外部系统聚合日志
• 在日志中携带关联 ID 以支持链路追踪

import logging
import json

class StructuredLogger:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        handler = logging.StreamHandler()
        handler.setFormatter(self._json_formatter)
        self.logger.addHandler(handler)

    def _json_formatter(self, record):
        log_entry = {
            'timestamp': record.created,
            'level': record.levelname,
            'message': record.getMessage(),
            'agent_id': getattr(record, 'agent_id', None),
            'session_id': getattr(record, 'session_id', None)
        }
        return json.dumps(log_entry)

XII. 管理进程

以一次性进程运行管理与运维任务

• 管理任务使用与主应用相同的代码库
• 在完全相同的环境中执行管理任务
• 为常见操作提供合适的工具支持
• 自动化例行维护任务

# Management commands
class AgentManager:
    def migrate_memory(self):
        """Migrate agent memory to new format"""
        pass

    def cleanup_old_sessions(self):
        """Remove expired session data"""
        pass

    def health_check(self):
        """Verify agent system health"""
        pass

实施指南

开发工作流

1. 本地开发：使用 docker-compose 管理本地服务
2. 测试：构建完善的测试套件
3. CI/CD：自动化构建、测试与部署流水线
4. 监控：实现健康检查与指标采集

部署模式

• 容器编排：使用 Kubernetes 或 Docker Swarm
• 服务网格：实现服务间通信
• 自动扩缩容：根据指标与负载进行配置
• 蓝绿部署：实现零停机发布

监控与可观测性

• 指标：追踪智能体性能与使用情况
• 链路追踪：实现分布式追踪
• 告警：建立主动监控机制
• 仪表板：提供运维可见性

优势

可扩展性

• 通过无状态进程实现水平扩展
• 各组件可独立扩展
• 高效利用资源

可维护性

• 关注点清晰分离
• 部署实践一致
• 调试与故障排查更简便

可移植性

• 与环境无关的设计
• 跨平台行为一致
• 便于在云厂商之间迁移

参见

• 架构组件选型
• 多智能体系统
• 智能体可观测性
• 最佳实践