# 快速实现自己的 Agent - 生产使用指南

## 🚀 三步快速开始

### 第一步：安装依赖

```bash
# 安装基础依赖
pip install -r requirements.txt

# 安装 browser 模块依赖（如果需要浏览器工具）
pip install dbutils pymysql

# 配置环境变量
cp .env.example .env
# 编辑 .env 填入你的 API Key
```

### 第二步：创建你的 Agent

创建 `my_agent/run.py`：

```python
import asyncio
import sys
from pathlib import Path

# 添加项目根目录到 Python 路径
sys.path.insert(0, str(Path(__file__).parent.parent))

from agent import AgentRunner, RunConfig, FileSystemTraceStore
from agent.llm import create_openrouter_llm_call

async def main():
    # 1. 初始化存储
    trace_store = FileSystemTraceStore(base_path=".cache/traces")

    # 2. 初始化 LLM
    llm_call = create_openrouter_llm_call(
        model="anthropic/claude-sonnet-4.5"
    )

    # 3. 创建 Runner
    runner = AgentRunner(
        llm_call=llm_call,
        trace_store=trace_store,
    )

    # 4. 运行 Agent
    async for item in runner.run(
        messages=[{"role": "user", "content": "你的任务描述"}],
        config=RunConfig(
            model="anthropic/claude-sonnet-4.5",
            max_iterations=30,
        ),
    ):
        print(item)

if __name__ == "__main__":
    asyncio.run(main())
```

### 第三步：运行

```bash
python my_agent/run.py
```

## 📦 核心组件

### 1. AgentRunner - 执行引擎

```python
from agent import AgentRunner, RunConfig

runner = AgentRunner(
    llm_call=llm_call,              # LLM 调用函数
    trace_store=trace_store,        # Trace 存储
    memory_store=memory_store,      # 记忆存储（可选）
    skills_dir="./skills",          # Skills 目录（可选）
)
```

### 2. LLM Providers

框架支持三种 LLM 提供商：

```python
# OpenRouter（推荐，支持多种模型）
from agent.llm import create_openrouter_llm_call
llm_call = create_openrouter_llm_call(
    model="anthropic/claude-sonnet-4.5",
    api_key="your-api-key",  # 或从环境变量读取
)

# Gemini
from agent.llm import create_gemini_llm_call
llm_call = create_gemini_llm_call(
    model="gemini-2.0-flash-exp",
    api_key="your-api-key",
)

# Yescode（内部）
from agent.llm import create_yescode_llm_call
llm_call = create_yescode_llm_call(
    model="gpt-4o",
    api_key="your-api-key",
)
```

### 3. 自定义工具

使用 `@tool` 装饰器注册工具：

```python
from agent import tool, ToolResult, ToolContext

@tool(description="查询产品库存")
async def check_inventory(
    product_id: str,
    warehouse: str = "default",
    ctx: ToolContext = None,
) -> ToolResult:
    """
    查询指定仓库的产品库存

    Args:
        product_id: 产品唯一标识符
        warehouse: 仓库编码，默认为主仓库
        ctx: 工具上下文（自动注入）
    """
    # 你的业务逻辑
    stock = await query_database(product_id, warehouse)

    return ToolResult(
        title="库存查询成功",
        output=f"产品 {product_id} 在 {warehouse} 仓库的库存: {stock}",
        data={"product_id": product_id, "stock": stock},
    )
```

**重要**：确保定义工具的模块在 `runner.run()` 之前被 import。

### 4. Skills - Agent 能力定义

创建 `skills/my-skill.md`：

```markdown
---
name: my-skill
description: 我的自定义技能
category: custom
---

# 我的技能

## 何时使用

- 场景1：当需要...
- 场景2：当遇到...

## 使用指南

1. 首先...
2. 然后...
3. 最后...

## 最佳实践

- 建议1
- 建议2
```

加载 Skills：

```python
runner = AgentRunner(
    llm_call=llm_call,
    trace_store=trace_store,
    skills_dir="./skills",  # 指定 skills 目录
)

# 或在 RunConfig 中指定
config = RunConfig(
    skills=["my-skill", "planning"],  # 只加载指定的 skills
)
```

### 5. Agent Presets - 预设配置

使用内置预设：

```python
from agent import RunConfig

# 默认 Agent（全部工具权限）
config = RunConfig(agent_type="default")

# 探索型 Agent（只读权限）
config = RunConfig(agent_type="explore")

# 评估型 Agent（只读+评估）
config = RunConfig(agent_type="evaluate")
```

自定义预设：

```python
from agent import AgentPreset
from agent.core.presets import register_preset

# 定义预设
my_preset = AgentPreset(
    allowed_tools=["read_file", "grep_content", "my_custom_tool"],
    denied_tools=["bash_command"],
    max_iterations=20,
    skills=["my-skill"],
    description="我的自定义 Agent",
)

# 注册预设
register_preset("my-agent", my_preset)

# 使用预设
config = RunConfig(agent_type="my-agent")
```

## 🎯 实战案例

### 案例1：内容寻找 Agent

参考 `examples/content_finder/`：

```python
# 1. 定义工具
@tool(description="从抖音搜索视频")
async def douyin_search(keywords: str, ctx: ToolContext = None) -> ToolResult:
    results = await call_douyin_api(keywords)
    return ToolResult(output=f"找到 {len(results)} 条内容", data=results)

# 2. 定义 Skill
# skills/content-finder.md

# 3. 运行 Agent
runner = AgentRunner(
    llm_call=llm_call,
    trace_store=trace_store,
    skills_dir="./skills",
)

async for item in runner.run(
    messages=[{"role": "user", "content": "搜索美食类视频"}],
    config=RunConfig(skills=["content-finder"]),
):
    if isinstance(item, Message):
        print(item.content)
```

### 案例2：数据分析 Agent

```python
@tool(description="查询数据库")
async def query_db(sql: str, ctx: ToolContext = None) -> ToolResult:
    results = await execute_sql(sql)
    return ToolResult(
        title="查询成功",
        output=f"返回 {len(results)} 条记录",
        data=results,
    )

@tool(description="生成图表")
async def create_chart(data: list, chart_type: str, ctx: ToolContext = None) -> ToolResult:
    chart_url = await generate_chart(data, chart_type)
    return ToolResult(
        title="图表已生成",
        output=f"图表类型: {chart_type}",
        data={"url": chart_url},
    )

# 运行
async for item in runner.run(
    messages=[{"role": "user", "content": "分析最近一周的销售数据并生成图表"}],
    config=RunConfig(
        skills=["data-analysis"],
        max_iterations=50,
    ),
):
    pass
```

### 案例3：自动化测试 Agent

```python
@tool(description="运行测试用例")
async def run_tests(test_path: str, ctx: ToolContext = None) -> ToolResult:
    result = await execute_tests(test_path)
    return ToolResult(
        title="测试完成",
        output=f"通过: {result.passed}, 失败: {result.failed}",
        data=result.to_dict(),
    )

@tool(description="生成测试报告")
async def generate_report(test_results: dict, ctx: ToolContext = None) -> ToolResult:
    report_path = await create_html_report(test_results)
    return ToolResult(
        title="报告已生成",
        output=f"报告路径: {report_path}",
    )
```

## 🔧 高级功能

### 1. 子 Agent

创建子 Agent 处理子任务：

```python
from agent.tools.builtin.subagent import agent

# Agent 会自动调用 agent 工具创建子 Agent
# 在 system prompt 中说明何时使用子 Agent
```

### 2. 记忆系统

使用记忆存储跨会话数据：

```python
from agent.memory.stores import MemoryMemoryStore

memory_store = MemoryMemoryStore()

runner = AgentRunner(
    llm_call=llm_call,
    trace_store=trace_store,
    memory_store=memory_store,
)
```

### 3. Goal Tree - 计划管理

Agent 自动使用 `goal` 工具管理计划：

```python
# Agent 会自动创建和更新 Goal
# 查看 Goal Tree
from agent.trace import FileSystemTraceStore

trace_store = FileSystemTraceStore(base_path=".cache/traces")
trace = await trace_store.get_trace(trace_id)
goal_tree = trace.goal_tree
```

### 4. 续跑和回溯

从指定消息后继续执行：

```python
# 续跑
async for item in runner.run(
    messages=[],
    config=RunConfig(
        trace_id="existing-trace-id",
        after_sequence=10,  # 从第10条消息后继续
    ),
):
    pass

# 回溯重跑
async for item in runner.run(
    messages=[],
    config=RunConfig(
        trace_id="existing-trace-id",
        after_sequence=5,  # 回到第5条消息重新执行
    ),
):
    pass
```

### 5. API Server - 可视化

启动 API Server 查看执行过程：

```bash
python api_server.py
```

访问：
- `http://localhost:8000/api/traces` - 查看所有 Traces
- `http://localhost:8000/api/traces/{id}` - 查看 Trace 详情
- WebSocket: `ws://localhost:8000/api/traces/{id}/watch` - 实时监控

## 📝 最佳实践

### 1. 工具设计原则

- **单一职责**：每个工具只做一件事
- **清晰描述**：description 要准确描述工具功能
- **详细文档**：docstring 要包含参数说明
- **错误处理**：捕获异常并返回友好的错误信息

```python
@tool(description="发送邮件")
async def send_email(
    to: str,
    subject: str,
    body: str,
    ctx: ToolContext = None,
) -> ToolResult:
    """
    发送邮件

    Args:
        to: 收件人邮箱地址
        subject: 邮件主题
        body: 邮件正文
    """
    try:
        await email_service.send(to, subject, body)
        return ToolResult(
            title="邮件发送成功",
            output=f"已发送到 {to}",
        )
    except Exception as e:
        return ToolResult(
            title="邮件发送失败",
            output=f"错误: {str(e)}",
            error=str(e),
        )
```

### 2. Skill 编写原则

- **明确场景**：清楚说明何时使用这个 Skill
- **提供指南**：给出具体的操作步骤
- **包含示例**：展示最佳实践
- **避免冗余**：不要重复框架已有的功能

### 3. 性能优化

```python
# 1. 限制迭代次数
config = RunConfig(max_iterations=30)

# 2. 使用合适的模型
llm_call = create_openrouter_llm_call(
    model="anthropic/claude-haiku-4.5",  # 更快更便宜
)

# 3. 启用 Prompt Caching（Claude 模型）
config = RunConfig(enable_prompt_caching=True)

# 4. 限制工具权限
config = RunConfig(
    agent_type="explore",  # 只读权限，更快
)
```

### 4. 错误处理

```python
try:
    async for item in runner.run(messages=messages, config=config):
        if isinstance(item, Message):
            # 处理消息
            pass
except Exception as e:
    logger.error(f"Agent 执行失败: {e}")
    # 错误处理逻辑
```

### 5. 日志和监控

```python
import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
)

logger = logging.getLogger(__name__)

# 在关键位置添加日志
logger.info("开始执行 Agent")
logger.debug(f"配置: {config}")
```

## 🚨 常见问题

### Q1: 如何调试 Agent？

```python
# 1. 打印所有消息
async for item in runner.run(messages=messages, config=config):
    print(f"[{type(item).__name__}] {item}")

# 2. 查看 Trace
trace = await trace_store.get_trace(trace_id)
for msg in trace.messages:
    print(f"{msg.role}: {msg.content}")

# 3. 启动 API Server 可视化
python api_server.py
```

### Q2: 工具没有被调用？

检查：
1. 工具是否正确注册（使用 `@tool` 装饰器）
2. 工具模块是否被 import
3. 工具描述是否清晰
4. Agent 是否有权限使用该工具（检查 `allowed_tools`）

### Q3: Agent 陷入循环？

```python
# 1. 限制迭代次数
config = RunConfig(max_iterations=20)

# 2. 在 Skill 中明确终止条件
# 3. 检查工具返回是否有明确的成功/失败标识
```

### Q4: 如何处理长时间运行的任务？

```python
# 使用异步处理
import asyncio

async def long_running_task():
    async for item in runner.run(messages=messages, config=config):
        # 定期保存状态
        if isinstance(item, Trace):
            await save_checkpoint(item)

# 支持中断和恢复
config = RunConfig(
    trace_id=last_trace_id,
    after_sequence=last_sequence,
)
```

## 📚 参考资源

- **框架文档**: `agent/README.md`
- **架构设计**: `agent/docs/architecture.md`
- **工具系统**: `agent/docs/tools.md`
- **Skills 指南**: `agent/docs/skills.md`
- **示例项目**: `examples/`
  - `examples/content_finder/` - 内容寻找 Agent
  - `examples/how/` - 完整示例
  - `examples/research/` - 研究型 Agent

## 🎓 下一步

1. **阅读示例代码**: 从 `examples/content_finder/demo.py` 开始
2. **创建第一个工具**: 实现一个简单的业务工具
3. **编写 Skill**: 定义 Agent 的工作方式
4. **测试运行**: 小规模测试验证功能
5. **生产部署**: 添加监控、日志、错误处理