用 Claude API 构建 AI Agent——从 Tool Use 到真正的自主任务执行

第16章：用 Claude API 构建 AI Agent——从 Tool Use 到真正的自主任务执行

Agent 不是更聪明的聊天机器人，而是一种完全不同的执行模式。你给它一个目标，它自主规划步骤、调用工具、读取结果、决定下一步行动，循环直到完成。本章覆盖完整的 Tool Use API 机制，实现一个可运行的代码审查 Agent，讲解 Memory 管理和安全边界，提供一个可直接 fork 使用的 200 行生产级 Agent 框架。

Agent vs 普通 LLM 调用：本质区别

关键机制： Claude 通过 stop_reason 告诉你它想做什么。"end_turn" 表示任务完成，"tool_use" 表示它想调用工具，你执行完工具后把结果返回给它，它再继续。这个循环就是 Agent 的核心。

Tool Use 完整实现（可运行代码）

以下代码实现了一个完整的三工具 Agent，包含工具定义、执行逻辑和调用循环，可以直接运行：

import anthropic
import json
import subprocess
from pathlib import Path

client = anthropic.Anthropic()

# 第一步：定义工具（JSON Schema 格式）
tools = [
    {
        "name": "read_file",
        "description": "读取文件内容",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "文件路径"}
            },
            "required": ["path"]
        }
    },
    {
        "name": "write_file",
        "description": "写入文件内容",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "content": {"type": "string"}
            },
            "required": ["path", "content"]
        }
    },
    {
        "name": "run_command",
        "description": "执行 shell 命令，返回输出",
        "input_schema": {
            "type": "object",
            "properties": {
                "command": {"type": "string"}
            },
            "required": ["command"]
        }
    }
]

# 第二步：实现工具执行逻辑
def execute_tool(name: str, inputs: dict) -> str:
    if name == "read_file":
        try:
            return Path(inputs["path"]).read_text()
        except FileNotFoundError:
            return f"Error: File not found: {inputs['path']}"

    elif name == "write_file":
        Path(inputs["path"]).write_text(inputs["content"])
        return f"Successfully wrote {len(inputs['content'])} chars to {inputs['path']}"

    elif name == "run_command":
        result = subprocess.run(
            inputs["command"], shell=True,
            capture_output=True, text=True, timeout=30
        )
        output = result.stdout + result.stderr
        return output[:5000]  # 限制输出长度，防止撑爆上下文

# 第三步：Agent 循环（核心逻辑，约 25 行）
def run_agent(task: str, max_turns: int = 10) -> str:
    messages = [{"role": "user", "content": task}]

    for turn in range(max_turns):
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            tools=tools,
            messages=messages
        )

        # stop_reason == "end_turn"：任务完成，返回最终回复
        if response.stop_reason == "end_turn":
            return response.content[0].text

        # stop_reason == "tool_use"：AI 要调用工具
        if response.stop_reason == "tool_use":
            # 1. 把 AI 的回复加入历史（包含 tool_use 块）
            messages.append({"role": "assistant", "content": response.content})

            # 2. 执行所有工具调用（一次回复里可能有多个）
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    print(f"  → 调用工具: {block.name}({block.input})")
                    result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })

            # 3. 把工具结果返回给 AI，进入下一轮
            messages.append({"role": "user", "content": tool_results})

    return "达到最大循环次数，任务未完成"

# 运行示例
result = run_agent(
    "分析 src/ 目录下所有 Python 文件，找出没有类型注解的函数，"
    "生成一个报告文件 type_report.md"
)
print(result)

为什么 messages.append 要在执行工具前做： 必须先把 AI 含 tool_use 块的那条回复加入 messages，再追加 tool_result。顺序错了 API 会报错，因为 Claude 需要看到自己发出的调用请求，才能匹配工具结果。

实战案例：代码审查 Agent

import anthropic
import subprocess
import json

client = anthropic.Anthropic()

REVIEW_TOOLS = [
    {
        "name": "get_changed_files",
        "description": "获取当前 PR 的所有改动文件列表",
        "input_schema": {"type": "object", "properties": {}, "required": []}
    },
    {
        "name": "read_file_diff",
        "description": "读取指定文件的完整内容和 diff",
        "input_schema": {
            "type": "object",
            "properties": {
                "filepath": {"type": "string", "description": "文件路径"}
            },
            "required": ["filepath"]
        }
    },
    {
        "name": "create_review_report",
        "description": "将审查结果结构化写入报告",
        "input_schema": {
            "type": "object",
            "properties": {
                "issues": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "file": {"type": "string"},
                            "line": {"type": "integer"},
                            "severity": {"type": "string", "enum": ["high", "medium", "low"]},
                            "description": {"type": "string"},
                            "suggestion": {"type": "string"}
                        }
                    }
                },
                "summary": {"type": "string"}
            },
            "required": ["issues", "summary"]
        }
    }
]

REVIEW_SYSTEM = """你是资深代码审查工程师。

工作步骤（严格按顺序执行）：
1. 调用 get_changed_files 获取改动文件列表
2. 逐一调用 read_file_diff 读取每个文件的内容和 diff
3. 识别以下问题（只报告真实存在的）：
   - 安全漏洞（SQL注入、XSS、密钥泄露）
   - 明显 bug（空指针、边界条件错误）
   - 性能问题（N+1查询、不必要的全量扫描）
   - 缺少错误处理
4. 调用 create_review_report 写入报告

注意：不报告代码风格和命名问题，只关注功能和安全。"""

实测数据： 对一个包含 8 个改动文件的 PR，这个 Agent 平均进行 12 次工具调用，耗时约 40 秒，能识别出人工审查容易遗漏的 SQL 拼接和缺失 try/except 等问题。

Memory 管理：让 Agent 记住重要信息

import json
from pathlib import Path

MEMORY_FILE = Path(".agent_memory.json")

def load_memory() -> dict:
    if MEMORY_FILE.exists():
        return json.loads(MEMORY_FILE.read_text())
    return {}

def save_memory(key: str, value: str) -> str:
    mem = load_memory()
    mem[key] = value
    MEMORY_FILE.write_text(json.dumps(mem, ensure_ascii=False, indent=2))
    return f"Saved: {key}"

def build_system_with_memory(base_prompt: str) -> str:
    mem = load_memory()
    if not mem:
        return base_prompt
    facts = "\n".join(f"- {k}: {v}" for k, v in mem.items())
    return f"{base_prompt}\n\n你已知道的项目信息：\n{facts}"

# 在工具列表里加上 remember 工具
remember_tool = {
    "name": "remember",
    "description": "记住一个重要事实，下次运行时仍然有效",
    "input_schema": {
        "type": "object",
        "properties": {
            "key": {"type": "string", "description": "记忆的键名，简短描述"},
            "value": {"type": "string", "description": "要记住的内容"}
        },
        "required": ["key", "value"]
    }
}

Agent 安全边界：哪些工具该给，哪些不该给

REQUIRE_CONFIRM = {"delete_file", "run_command", "send_email", "db_write"}

def safe_execute(name: str, inputs: dict) -> str:
    if name in REQUIRE_CONFIRM:
        print(f"\n[!] Agent 请求执行高风险操作: {name}")
        print(f"    参数: {json.dumps(inputs, ensure_ascii=False)}")
        confirm = input("    允许执行? [y/N] ").strip().lower()
        if confirm != "y":
            return "操作被用户取消，请更换方案"
    return execute_tool(name, inputs)

200 行生产级 Agent 框架

"""
agent.py — 生产级 Claude Agent 框架
支持：工具调用循环 / Human-in-the-loop / 长期记忆 / 日志
"""
import json
import logging
import sqlite3
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Callable

import anthropic

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
log = logging.getLogger("agent")


@dataclass
class Tool:
    name: str
    description: str
    input_schema: dict
    handler: Callable
    dangerous: bool = False  # True = 需要人工确认

    def api_spec(self) -> dict:
        return {
            "name": self.name,
            "description": self.description,
            "input_schema": self.input_schema
        }


class Memory:
    """SQLite 长期记忆"""
    def __init__(self, path: str = "agent_memory.db"):
        self.conn = sqlite3.connect(path)
        self.conn.execute(
            "CREATE TABLE IF NOT EXISTS mem "
            "(agent TEXT, key TEXT, val TEXT, ts TEXT, PRIMARY KEY(agent,key))"
        )
        self.conn.commit()

    def set(self, agent: str, key: str, val: Any):
        self.conn.execute(
            "INSERT OR REPLACE INTO mem VALUES (?,?,?,?)",
            (agent, key, json.dumps(val, ensure_ascii=False), datetime.now().isoformat())
        )
        self.conn.commit()

    def all(self, agent: str) -> dict:
        rows = self.conn.execute(
            "SELECT key, val FROM mem WHERE agent=?", (agent,)
        ).fetchall()
        return {r[0]: json.loads(r[1]) for r in rows}


@dataclass
class AgentConfig:
    model: str = "claude-sonnet-4-6"
    max_tokens: int = 4096
    max_turns: int = 30
    system: str = ""
    use_memory: bool = False
    memory_db: str = "agent_memory.db"
    auto_approve: bool = False  # 生产环境保持 False


class Agent:
    def __init__(self, agent_id: str, tools: list[Tool], config: AgentConfig = None):
        self.id = agent_id
        self.tools = {t.name: t for t in tools}
        self.cfg = config or AgentConfig()
        self.client = anthropic.Anthropic()
        self.memory = Memory(self.cfg.memory_db) if self.cfg.use_memory else None
        self._call_log: list[dict] = []

    def _run_tool(self, name: str, inputs: dict) -> str:
        tool = self.tools.get(name)
        if not tool:
            return f"ERROR: unknown tool '{name}'"

        if tool.dangerous and not self.cfg.auto_approve:
            print(f"\n[!] Agent [{self.id}] wants to run: {name}")
            print(f"    inputs: {json.dumps(inputs, ensure_ascii=False)}")
            ans = input("    Approve? [y/N] ").strip().lower()
            if ans != "y":
                return "Rejected by user."

        try:
            result = tool.handler(**inputs)
            self._call_log.append({"tool": name, "ok": True})
            return str(result)
        except Exception as exc:
            self._call_log.append({"tool": name, "ok": False, "err": str(exc)})
            return f"ERROR: {exc}"

    def run(self, task: str) -> str:
        log.info(f"Agent [{self.id}] start: {task[:60]}...")
        self._call_log.clear()

        system = self.cfg.system
        if self.memory:
            mem = self.memory.all(self.id)
            if mem:
                facts = "\n".join(f"- {k}: {v}" for k, v in mem.items())
                system += f"\n\n已知项目信息：\n{facts}"

        messages = [{"role": "user", "content": task}]
        api_tools = [t.api_spec() for t in self.tools.values()]

        for turn in range(self.cfg.max_turns):
            kwargs = dict(
                model=self.cfg.model,
                max_tokens=self.cfg.max_tokens,
                tools=api_tools,
                messages=messages
            )
            if system:
                kwargs["system"] = system

            resp = self.client.messages.create(**kwargs)
            messages.append({"role": "assistant", "content": resp.content})

            if resp.stop_reason == "end_turn":
                final = next((b.text for b in resp.content if hasattr(b, "text")), "")
                log.info(f"done in {len(self._call_log)} tool calls")
                return final

            if resp.stop_reason == "tool_use":
                results = []
                for blk in resp.content:
                    if blk.type == "tool_use":
                        res = self._run_tool(blk.name, blk.input)
                        results.append({
                            "type": "tool_result",
                            "tool_use_id": blk.id,
                            "content": res
                        })
                messages.append({"role": "user", "content": results})

        log.warning("max turns reached")
        return "任务未完成：达到最大循环次数"

框架亮点： 不到 200 行，包含工具调用循环、Human-in-the-loop 确认、SQLite 长期记忆、结构化日志和最大轮次保护。直接继承 Agent 类，传入你的工具列表，就能跑起来。

本章要点

1. Agent 的本质是循环： 检查 stop_reason，是 tool_use 就执行工具并把结果追加到 messages，是 end_turn 就结束。这个循环就是 Agent 的全部机制。
2. 工具定义决定 Agent 能力上限： 工具的 description 写得越精确，Claude 调用它的准确率越高；description 模糊是工具调用失败的头号原因。
3. 必须限制输出长度： 工具返回值截断到 3000~5000 字符，防止长输出快速消耗上下文窗口导致 Agent 提前失败。
4. 高风险工具必须加人工确认： 删除、发邮件、数据库写操作等不可逆操作，在 dangerous=True 的工具上加 input() 确认，一行代码就能保护。
5. 长期记忆用 SQLite 最简单： Agent 第一次运行时发现的项目约定、已知问题等，存入 SQLite，下次运行时注入 system prompt，不需要任何额外依赖。