理解 AI Agent 的核心机制：循环

1
2
3
4
5
6
7

# 极简 Agent 循环
while True:
    response = model(messages, tools)
    if response.stop_reason != "tool_use":
        return response.text
    results = execute(response.tool_calls)
    messages.append(results)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

from anthropic import Anthropic
import subprocess

client = Anthropic(api_key="your-api-key")

# 核心配置：一个工具 + 一个系统提示词
TOOLS = [{
    "name": "bash",
    "description": "执行 shell 命令",
    "input_schema": {
        "type": "object",
        "properties": {"command": {"type": "string"}},
        "required": ["command"]
    }
}]

def agent_loop(messages):
    while True:
        # 阶段 1：模型决策
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            messages=messages,
            tools=TOOLS
        )

        # 阶段 2：检查是否需要工具
        if response.stop_reason != "tool_use":
            return response.content[0].text

        # [关键步骤] 将模型的思考过程和工具调用请求加入历史
        messages.append({"role": "assistant", "content": response.content})

        # 阶段 3：执行工具并反馈
        results = []
        for block in response.content:
            if block.type == "tool_use":
                output = subprocess.run(
                    block.input["command"],
                    shell=True,
                    capture_output=True,
                    text=True
                ).stdout
                results.append({
                    "type": "tool_result",
                    "name": block.name,
                    "tool_use_id": block.id,
                    "content": output
                })
        messages.append({"role": "user", "content": results})

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

TOOLS = [
    # 工具 1：Bash - "手"：系统操作的网关
    {
        "name": "bash",
        "description": "执行 shell 命令：ls, find, grep, git, npm, python 等",
        "input_schema": {
            "type": "object",
            "properties": {"command": {"type": "string"}},
            "required": ["command"]
        }
    },

    # 工具 2：read_file - "眼"：理解代码的窗口
    {
        "name": "read_file",
        "description": "读取文件内容，支持限制行数以处理大文件",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "limit": {"type": "integer", "description": "最大读取行数"}
            },
            "required": ["path"]
        }
    },

    # 工具 3：write_file - 创建与重写
    {
        "name": "write_file",
        "description": "创建新文件或完全覆盖现有文件",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "content": {"type": "string"}
            },
            "required": ["path", "content"]
        }
    },

    # 工具 4：edit_file - "手术刀"：精确修改
    {
        "name": "edit_file",
        "description": "精确替换文件中的某段文本",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "old_text": {"type": "string"},
                "new_text": {"type": "string"}
            },
            "required": ["path", "old_text", "new_text"]
        }
    }
]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

from pathlib import Path
import subprocess

WORKDIR = Path.cwd()

def safe_path(p: str) -> Path:
    """安全路径检查：防止路径逃逸工作目录"""
    path = (WORKDIR / p).resolve()
    if not path.is_relative_to(WORKDIR):
        raise ValueError(f"Path escapes workspace: {p}")
    return path

def run_bash(command: str) -> str:
    """执行 shell 命令，带超时和基本安全检查"""
    dangerous = ["rm -rf /", "sudo", "shutdown"]
    if any(d in command for d in dangerous):
        return "Error: Dangerous command blocked"

    result = subprocess.run(
        command,
        shell=True,
        cwd=WORKDIR,
        capture_output=True,
        text=True,
        timeout=60
    )
    return (result.stdout + result.stderr).strip()[:50000]

def run_read(path: str, limit: int = None) -> str:
    """读取文件内容，可选行数限制"""
    text = safe_path(path).read_text()
    if limit:
        lines = text.splitlines()[:limit]
        lines.append(f"... ({len(text.splitlines()) - limit} more lines)")
        return "\n".join(lines)
    return text

def run_write(path: str, content: str) -> str:
    """写入文件，自动创建父目录"""
    fp = safe_path(path)
    fp.parent.mkdir(parents=True, exist_ok=True)
    fp.write_text(content)
    return f"Wrote {len(content)} bytes to {path}"

def run_edit(path: str, old_text: str, new_text: str) -> str:
    """精确替换文本，只修改第一个匹配项（注意：需完全匹配包括缩进在内的字符）"""
    fp = safe_path(path)
    content = fp.read_text()
    if old_text not in content:
        return f"Error: Text not found in {path}"
    new_content = content.replace(old_text, new_text, 1)
    fp.write_text(new_content)
    return f"Edited {path}"

def execute_tool(name: str, args: dict) -> str:
    """工具调用分派中心"""
    dispatch = {
        "bash": lambda a=args: run_bash(a["command"]),
        "read_file": lambda a=args: run_read(a["path"], a.get("limit")),
        "write_file": lambda a=args: run_write(a["path"], a["content"]),
        "edit_file": lambda a=args: run_edit(a["path"], a["old_text"], a["new_text"])
    }
    if name in dispatch:
        return dispatch[name]()
    return f"Unknown tool: {name}"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

SYSTEM = f"""你是一个专业的Agent ，工作目录是 {WORKDIR}。

## 工作模式

遵循「思考 → 行动 → 报告」的循环：
1. 短暂思考下一步该做什么（Chain of Thought）
2. 使用合适的工具执行
3. 报告结果并继续

## 核心原则

- **行动优先**：优先使用工具而非长篇解释
- **路径安全**：不确定文件路径时，先用 ls/find 查找
- **最小修改**：只做必要的改动，避免过度工程
- **可验证性**：完成后总结变更内容，让用户可以验证

## 能力边界

- 可以读取、创建、编辑文件
- 可以执行任何 shell 命令
- 不能访问工作目录之外的路径"""

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

MODEL = "your model name"

def agent_loop(messages: list) -> list:
    """
    完整的 Agent 循环实现

    核心模式：
        while True:
            response = model(messages, tools)
            if no tool calls: return
            execute tools, append results, continue
    """
    while True:
        # Step 1：模型决策阶段
        response = client.messages.create(
            model=MODEL,
            system=SYSTEM,
            messages=messages,
            tools=TOOLS,
            max_tokens=8000
        )

        # Step 2：解析响应，区分文本输出和工具调用
        tool_calls = []
        for block in response.content:
            if hasattr(block, "text"):
                print(block.text)  # 显示模型的思考过程
            if block.type == "tool_use":
                tool_calls.append(block)

        # Step 3：检查是否完成（无工具调用 = 任务结束）
        if response.stop_reason != "tool_use":
            messages.append({"role": "assistant", "content": response.content})
            return messages

        # Step 4：执行所有工具调用
        results = []
        for tc in tool_calls:
            print(f"\n> {tc.name}: {tc.input}")  # 显示执行动作

            output = execute_tool(tc.name, tc.input)
            preview = output[:200] + "..." if len(output) > 200 else output
            print(f"  {preview}")  # 显示结果预览

            results.append({
                "type": "tool_result",
                "name": tc.name,
                "tool_use_id": tc.id,
                "content": output
            })

        # Step 5：将结果追加到对话历史，继续循环
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": results})

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

# 代理类型注册表：定义不同类型子代理的能力边界
AGENT_TYPES = {
    "explore": {
        "description": "只读代理，用于搜索和分析代码库",
        "tools": ["bash", "read_file"],  # 权限收敛：不能写入
        "prompt": "你是探索者。广泛搜索、深入分析，但绝不修改任何文件。返回结构化的简洁摘要。"
    },
    "code": {
        "description": "完整代理，用于实现功能",
        "tools": "*",  # 全权限
        "prompt": "你是实现者。高效执行代码变更，遵循最佳实践，确保代码质量。"
    },
    "plan": {
        "description": "规划代理，用于设计实施方案",
        "tools": ["bash", "read_file"],  # 只读
        "prompt": "你是架构师。分析问题，设计清晰的实现步骤。输出编号列表，不要修改任何代码。"
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# Task 工具：触发子代理的核心
TASK_TOOL = {
    "name": "Task",
    "description": "为独立子任务生成专用子代理",
    "input_schema": {
        "type": "object",
        "properties": {
            "description": {"type": "string", "description": "简短的任务名"},
            "prompt": {"type": "string", "description": "详细指令"},
            "agent_type": {"type": "string", "enum": ["explore", "code", "plan"]}
        },
        "required": ["description", "prompt", "agent_type"]
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

def run_task(description: str, prompt: str, agent_type: str) -> str:
    """
    执行子代理任务（隔离上下文版本）

    关键设计：
    1. 全新消息历史（不继承父上下文）
    2. 代理特化系统提示词
    3. 基于类型的工具过滤
    4. 只返回最终摘要
    """
    config = AGENT_TYPES[agent_type]

    # 构建子代理专用的系统提示词
    sub_system = f"""你是 {agent_type} 子代理。

{config["prompt"]}

完成任务后，返回清晰简洁的摘要。"""

    # 根据代理类型过滤可用工具
    sub_tools = get_tools_for_agent(agent_type)

    # 【关键】创建隔离的消息历史
    sub_messages = [{"role": "user", "content": prompt}]

    # 运行子代理循环（与主循环相同的模式）
    while True:
        response = client.messages.create(
            model=MODEL,
            system=sub_system,
            messages=sub_messages,
            tools=sub_tools,
            max_tokens=8000
        )

        if response.stop_reason != "tool_use":
            break

        # 执行工具调用
        tool_calls = [b for b in response.content if b.type == "tool_use"]
        results = []
        for tc in tool_calls:
            output = execute_tool(tc.name, tc.input)
            results.append({
                "type": "tool_result",
                "tool_use_id": tc.id,
                "content": output
            })

        # 追加到子代理历史
        sub_messages.append({"role": "assistant", "content": response.content})
        sub_messages.append({"role": "user", "content": results})

    # 只返回最终文本摘要
    for block in response.content:
        if hasattr(block, "text"):
            return block.text
    return "(子代理未返回文本)"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

---
name: pdf
description: 处理 PDF 文件。用于读取、创建或合并 PDF。
---

# PDF 处理技能

## 读取 PDF

使用 pdftotext 快速提取文本：
```bash
pdftotext input.pdf -

1
2
3
4

import fitz  # PyMuPDF
doc = fitz.open("input.pdf")
for page in doc:
    text = page.get_text()

1
2
3
4
5

from PyPDF2 import PdfMerger
merger = PdfMerger()
merger.append("file1.pdf")
merger.append("file2.pdf")
merger.write("output.pdf")

1
2
3
4

while not done:
    response = model(conversation, tools)
    results = execute(response.tool_calls)
    conversation.append(results)