AI Agent 开发实战：从工具调用到自动化工作流

用 Claude API 搭自动化周报生成器的时候踩了个坑——单次 Prompt 调用根本搞不定多步骤任务。工具调用（Tool Use）加上循环执行，才是让 AI 真正"干事"的关键。

问题是什么

需求其实很简单：每周自动从 Jira 拉项目数据，分析进度风险，生成带图表的周报，再发到企业微信。

一开始我就写了一个 Prompt 调 Claude API，塞一堆数据进去让它一次性输出。结果惨不忍睹——Token 全花在记数据上，分析浅得像流水账，图表更是完全没法看。

痛点很明确：LLM 的单次调用是无状态的。你给数据，它吐回答，然后就没了。它不会说"等等，我先查一下昨天的数据"，也不会说"让我再验证一下这个数字对不对"。要让 AI 搞定复杂的多步骤任务，必须上 Agent 模式——让模型能主动调工具、看结果、自己决定下一步。

解决思路

我对比了三种方案：

选了第三种——手写 Agent Loop。原因很简单：我要完全控制工具的调用逻辑和错误重试策略，不想被框架的黑盒卡住。生产环境自己管错误处理和状态，心里更有底。

操作步骤

步骤 1：定义工具 Schema

Agent 的根基是工具调用。先把需要用到的工具定义成 JSON Schema，这是 Anthropic API 的标准格式：

tools = [
    {
        "name": "fetch_jira_sprint_data",
        "description": "获取当前 Sprint 的 Jira 任务数据，包括任务状态、负责人、预估工时",
        "input_schema": {
            "type": "object",
            "properties": {
                "sprint_id": {
                    "type": "string",
                    "description": "Sprint ID"
                }
            },
            "required": ["sprint_id"]
        }
    },
    {
        "name": "analyze_risk",
        "description": "分析任务列表中的进度风险，返回高风险项",
        "input_schema": {
            "type": "object",
            "properties": {
                "tasks": {
                    "type": "array",
                    "items": {"type": "object"},
                    "description": "任务列表"
                }
            },
            "required": ["tasks"]
        }
    },
    {
        "name": "generate_chart",
        "description": "生成进度的燃尽图（Burndown Chart），返回图片 URL",
        "input_schema": {
            "type": "object",
            "properties": {
                "data": {
                    "type": "object",
                    "description": "燃尽图数据"
                }
            },
            "required": ["data"]
        }
    },
    {
        "name": "send_wecom_message",
        "description": "发送消息到企业微信群机器人",
        "input_schema": {
            "type": "object",
            "properties": {
                "content": {
                    "type": "string",
                    "description": "消息内容（支持 Markdown）"
                }
            },
            "required": ["content"]
        }
    }
]

Tool Schema 里的 description 字段特别关键——模型靠它来判断什么时候调哪个工具，写得越清楚，模型选得越准。

步骤 2：实现工具函数

每个 Schema 对应一个真实的 Python 函数，这些就是 Agent 的"手和脚"：

import json
import requests
from datetime import datetime, timedelta

def fetch_jira_sprint_data(sprint_id: str) -> dict:
    """从 Jira API 获取 Sprint 数据"""
    # 实际项目中这里用 Jira API 认证调用
    # 这里用模拟数据演示
    return {
        "sprint_id": sprint_id,
        "tasks": [
            {"id": "PROJ-101", "title": "用户登录重构", "status": "进行中", 
             "assignee": "张三", "estimate_hours": 16, "remaining_hours": 10},
            {"id": "PROJ-102", "title": "支付模块对接", "status": "阻塞", 
             "assignee": "李四", "estimate_hours": 24, "remaining_hours": 20,
             "block_reason": "等待第三方接口文档"},
            {"id": "PROJ-103", "title": "首页性能优化", "status": "已完成", 
             "assignee": "王五", "estimate_hours": 8, "remaining_hours": 0},
        ]
    }

def analyze_risk(tasks: list) -> dict:
    """分析任务风险"""
    high_risk = []
    medium_risk = []
    for task in tasks:
        if task.get("status") == "阻塞":
            high_risk.append(task)
        elif task.get("remaining_hours", 0) > task.get("estimate_hours", 1) * 0.7:
            medium_risk.append(task)
    return {
        "high_risk_count": len(high_risk),
        "medium_risk_count": len(medium_risk),
        "high_risk_tasks": high_risk,
        "overall_status": "需关注" if high_risk else "正常"
    }

def generate_chart(data: dict) -> str:
    """生成燃尽图，返回图片 URL（这里用 matplotlib 生成）"""
    import matplotlib.pyplot as plt
    # 实际代码会画图并上传到图床
    # 此处省略具体绘图代码
    return "https://example.com/charts/burndown_sprint_123.png"

def send_wecom_message(content: str) -> dict:
    """发送企业微信消息"""
    webhook_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"
    payload = {
        "msgtype": "markdown",
        "markdown": {"content": content}
    }
    resp = requests.post(webhook_url, json=payload)
    return {"status": "ok" if resp.status_code == 200 else "fail"}

步骤 3：实现 Agent 主循环

核心来了——Agent Loop。每次 API 调用后检查响应里有没有 tool_use 类型的 content block，有就执行对应工具，把结果塞回消息列表，然后继续调 API：

import anthropic

client = anthropic.Anthropic(api_key="sk-ant-xxx")

def run_agent(prompt: str, max_iterations: int = 10):
    """Agent 主循环"""
    messages = [{"role": "user", "content": prompt}]
    
    # 工具名称 -> 函数的映射
    tool_map = {
        "fetch_jira_sprint_data": fetch_jira_sprint_data,
        "analyze_risk": analyze_risk,
        "generate_chart": generate_chart,
        "send_wecom_message": send_wecom_message,
    }
    
    for i in range(max_iterations):
        print(f"
=== Agent 迭代 {i + 1} ===")
        
        response = client.messages.create(
            model="claude-sonnet-4-20250506",
            max_tokens=4000,
            system="你是一个周报生成助手。请按顺序执行："
                   "1. 获取 Jira Sprint 数据 "
                   "2. 分析进度风险 "
                   "3. 生成燃尽图 "
                   "4. 整理周报内容 "
                   "5. 发送到企业微信。每次只调用一个工具。",
            messages=messages,
            tools=tools
        )
        
        # 检查是否有工具调用
        has_tool_call = False
        for block in response.content:
            if block.type == "tool_use":
                has_tool_call = True
                tool_name = block.name
                tool_input = block.input
                tool_use_id = block.id
                
                print(f"→ 调用工具: {tool_name}")
                print(f"  参数: {json.dumps(tool_input, ensure_ascii=False)}")
                
                # 执行工具函数
                try:
                    result = tool_map[tool_name](**tool_input)
                    result_str = json.dumps(result, ensure_ascii=False)
                except Exception as e:
                    result_str = json.dumps({"error": str(e)})
                
                print(f"  结果: {result_str[:200]}...")
                
                # 将工具调用和结果追加到消息列表
                messages.append({
                    "role": "assistant",
                    "content": [block]
                })
                messages.append({
                    "role": "user",
                    "content": [
                        {
                            "type": "tool_result",
                            "tool_use_id": tool_use_id,
                            "content": result_str
                        }
                    ]
                })
            elif block.type == "text" and not has_tool_call:
                # 没有工具调用时，说明 Agent 完成了任务
                return block.text
        
        if not has_tool_call:
            break
    
    # 返回最后一次响应的文本
    for block in reversed(response.content):
        if block.type == "text":
            return block.text
    return "Agent 未生成有效响应"

步骤 4：添加状态管理和错误重试

生产环境最大的坑——API 可能挂，工具可能抛异常。我加了两层保护：

import time
from functools import wraps

def retry_on_failure(max_retries=3, delay=1):
    """API 调用重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except anthropic.RateLimitError:
                    wait = delay * (2 ** attempt)
                    print(f"⚠ 触发限流，{wait}秒后重试...")
                    time.sleep(wait)
                except anthropic.APIError as e:
                    print(f"⚠ API 错误: {e}")
                    if attempt == max_retries - 1:
                        raise
                    time.sleep(delay)
            return None
        return wrapper
    return decorator

@retry_on_failure()
def call_llm(messages, tools):
    return client.messages.create(
        model="claude-sonnet-4-20250506",
        max_tokens=4000,
        system="你是一个周报生成助手。",
        messages=messages,
        tools=tools
    )

限流在生产环境很常见，指数退避是最基础的重试策略，配合请求排队和并发控制就更稳了。

步骤 5：组装完整工作流

把上面所有模块串起来：

def weekly_report_workflow():
    """完整的周报生成工作流"""
    print("🚀 启动周报生成工作流")
    
    prompt = """请生成本周 Sprint 周报：
1. 先获取当前 Sprint 数据（Sprint ID: SPRINT-23）
2. 分析任务进度风险
3. 生成燃尽图
4. 整理成周报 Markdown 格式
5. 发送到企业微信"""

    result = run_agent(prompt)
    
    if result:
        print("
✅ 周报生成完成！")
        print(result)
    else:
        print("
❌ 周报生成失败")

if __name__ == "__main__":
    weekly_report_workflow()

结果与总结

这套 Agent 架构上线跑了 6 周，效果比单次 Prompt 调用好太多：

踩到的坑：

延伸思考

这个 Agent Loop 是最朴素的 ReAct 模式，生产级 Agent 还可以做几件事：

工具调用让 LLM 从"聊天机器人"进化成了"能干活的 Agent"，Agent Loop 让 AI 能完成真正的工作流。这个模式不复杂，核心就是两件事：定义好工具接口，写对循环逻辑。框架能帮你省点代码，但理解了底层原理，你才能真正驾驭 Agent。