Agent/examples/production_template

生产级 Agent 模板

这是一个可直接用于生产环境的 Agent 模板项目。

特性

• ✅ 完整的错误处理
• ✅ 日志记录（文件+控制台）
• ✅ 环境变量配置
• ✅ 工具注册示例
• ✅ Skills 加载
• ✅ 结果持久化
• ✅ 生产级代码结构

快速开始

0. 安装依赖

# 从项目根目录安装
pip install -r requirements.txt
pip install dbutils pymysql  # browser 模块依赖

1. 配置环境变量

创建 .env 文件：

# LLM 配置
OPENROUTER_API_KEY=your-api-key-here
MODEL=anthropic/claude-sonnet-4.5
TEMPERATURE=0.3
MAX_ITERATIONS=30

# 存储配置
TRACE_DIR=.cache/traces
OUTPUT_DIR=.cache/output

# Skills 配置
SKILLS_DIR=./skills
ENABLED_SKILLS=planning,research

2. 运行

# 从项目根目录运行
python examples/production_template/run.py

# 或进入目录运行
cd examples/production_template
python run.py

项目结构

production_template/
├── run.py              # 主程序
├── .env                # 环境变量配置
├── skills/             # Skills 目录
│   └── custom.md       # 自定义 Skill
└── .cache/             # 缓存目录
    ├── traces/         # Trace 存储
    ├── output/         # 结果输出
    └── agent.log       # 日志文件

核心组件

1. AgentConfig - 配置管理

统一管理所有配置项，从环境变量读取：

config = AgentConfig()
print(config.model)  # anthropic/claude-sonnet-4.5

2. 自定义工具

使用 @tool 装饰器定义工具：

@tool(description="你的工具描述")
async def your_tool(param: str, ctx: ToolContext = None) -> ToolResult:
    # 你的业务逻辑
    return ToolResult(title="成功", output="结果")

3. ProductionAgent - Agent 封装

封装了 AgentRunner，提供简洁的 API：

agent = ProductionAgent(config)
result = await agent.run("你的任务")

4. 日志记录

自动记录到文件和控制台：

logger.info("信息日志")
logger.error("错误日志", exc_info=True)

自定义开发

添加新工具

在 run.py 中添加：

@tool(description="你的新工具")
async def new_tool(param: str, ctx: ToolContext = None) -> ToolResult:
    try:
        # 业务逻辑
        result = do_something(param)

        return ToolResult(
            title="成功",
            output=result,
            data={"param": param, "result": result},
        )
    except Exception as e:
        logger.error(f"工具执行失败: {e}", exc_info=True)
        return ToolResult(
            title="失败",
            output=str(e),
            error=True,
        )

添加 Skill

创建 skills/your-skill.md：

---
name: your-skill
description: 你的技能描述
---

# 你的技能

## 何时使用

- 场景1
- 场景2

## 使用指南

1. 步骤1
2. 步骤2

在 .env 中启用：

ENABLED_SKILLS=planning,research,your-skill

修改任务

在 main() 函数中修改 task 变量：

task = """
你的任务描述
"""

result = await agent.run(task)

生产部署

1. Docker 部署

创建 Dockerfile：

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["python", "run.py"]

构建和运行：

docker build -t my-agent .
docker run --env-file .env my-agent

2. 监控和告警

添加监控：

# 在 ProductionAgent 中添加
def _send_alert(self, message: str):
    """发送告警"""
    # 集成你的告警系统（钉钉、企业微信等）
    pass

# 在异常处理中调用
except Exception as e:
    self._send_alert(f"Agent 执行失败: {e}")

3. 性能优化

# 1. 使用更快的模型
MODEL=anthropic/claude-haiku-4.5

# 2. 限制迭代次数
MAX_ITERATIONS=20

# 3. 启用 Prompt Caching
config = RunConfig(enable_prompt_caching=True)

4. 错误重试

async def run_with_retry(self, task: str, max_retries: int = 3):
    """带重试的运行"""
    for attempt in range(max_retries):
        try:
            return await self.run(task)
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            logger.warning(f"重试 {attempt + 1}/{max_retries}: {e}")
            await asyncio.sleep(2 ** attempt)  # 指数退避

最佳实践

1. 环境变量管理：所有配置通过环境变量管理，不要硬编码
2. 日志记录：关键操作都要记录日志
3. 错误处理：所有工具都要有 try-except
4. 结果持久化：重要结果要保存到文件或数据库
5. 监控告警：生产环境要有监控和告警机制

常见问题

Q: 如何调试？

# 设置日志级别为 DEBUG
logging.basicConfig(level=logging.DEBUG)

Q: 如何查看执行过程？

# 启动 API Server
python api_server.py

# 访问
http://localhost:8000/api/traces

Q: 如何处理长时间任务？

# 增加最大迭代次数
MAX_ITERATIONS=100

# 或使用异步任务队列（Celery、RQ 等）

参考

• 快速上手指南
• 框架文档
• 示例项目