Claude 做 AI Agent 实战教程:从零搭建一个能自主执行任务的智能体(2026)
上周我接了个私活,甲方要求做一个"能自己查资料、写报告、发邮件"的自动化助手。说白了就是一个 AI Agent。我一开始想用 LangChain 那套,搭到一半发现链路太长、调试痛苦,后来干脆回归本质——直接用 Claude 的 Tool Use(Function Calling)能力,手搓了一个 Agent 框架。整个过程大概花了两天,效果比我预期好不少。
核心思路是:利用 Claude Opus 4.6 / Sonnet 4.6 的 Tool Use 能力,让模型在对话循环中自主决定调用哪些工具、按什么顺序执行,直到任务完成。不需要复杂的框架,一个 while 循环加几个工具函数就能跑起来。
先说结论
| 维度 | 说明 |
|---|---|
| 核心能力 | Claude 的 Tool Use(Function Calling) |
| 适用模型 | Claude Opus 4.6 / Sonnet 4.6(推荐 Sonnet,性价比高) |
| 框架依赖 | 不需要 LangChain/CrewAI,纯 SDK 就够 |
| 代码量 | 核心循环 < 100 行 |
| 适用场景 | 自动化报告、数据采集、多步骤任务编排 |
Agent 到底是什么?别被概念唬住
热榜上天天说"AI Agent",这词被吹得有点虚。剥开来看,Agent 就是一个循环决策系统:
- 接收用户指令
- 模型决定要不要调用工具
- 调用工具,拿到结果
- 把结果喂回模型,让它决定下一步
- 重复 2-4,直到模型认为任务完成
就这么简单。没有黑科技。
环境准备
pip install openai httpx
没错,用的是 OpenAI SDK。Claude 的 API 兼容 OpenAI 协议,用聚合接口的话一套代码就能跑,不用装 anthropic 那个包(当然你装也行)。
Python 版本我用的 3.11,3.10+ 都没问题。
第一步:定义工具(Tools)
Agent 的能力边界完全取决于你给它什么工具。这个项目需要三个:搜索网页、读取文件、发送邮件。
先定义工具的 schema:
tools = [
{
"type": "function",
"function": {
"name": "search_web",
"description": "搜索互联网获取最新信息,返回搜索结果摘要",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "read_file",
"description": "读取本地文件内容",
"parameters": {
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "文件路径"
}
},
"required": ["file_path"]
}
}
},
{
"type": "function",
"function": {
"name": "send_email",
"description": "发送邮件给指定收件人",
"parameters": {
"type": "object",
"properties": {
"to": {"type": "string", "description": "收件人邮箱"},
"subject": {"type": "string", "description": "邮件主题"},
"body": {"type": "string", "description": "邮件正文(支持 Markdown)"}
},
"required": ["to", "subject", "body"]
}
}
}
]
然后是工具的实际执行函数:
import json
import httpx
import smtplib
from email.mime.text import MIMEText
def search_web(query: str) -> str:
"""用搜索 API 获取结果,这里用 DuckDuckGo 的免费接口演示"""
try:
resp = httpx.get(
"https://api.duckduckgo.com/",
params={"q": query, "format": "json", "no_html": 1},
timeout=10
)
data = resp.json()
results = []
for topic in data.get("RelatedTopics", [])[:5]:
if "Text" in topic:
results.append(topic["Text"])
return "\n".join(results) if results else "未找到相关结果"
except Exception as e:
return f"搜索出错: {str(e)}"
def read_file(file_path: str) -> str:
"""读取本地文件"""
try:
with open(file_path, "r", encoding="utf-8") as f:
content = f.read()
# 截断太长的文件,避免 token 爆炸
if len(content) > 8000:
content = content[:8000] + "\n...[文件过长,已截断]"
return content
except FileNotFoundError:
return f"文件不存在: {file_path}"
except Exception as e:
return f"读取文件出错: {str(e)}"
def send_email(to: str, subject: str, body: str) -> str:
"""发送邮件(示例用 SMTP,实际项目建议用 SendGrid/Resend)"""
# 这里只做演示,实际使用请配置你的 SMTP
print(f"[模拟发送邮件] 收件人: {to}, 主题: {subject}")
print(f"正文预览: {body[:200]}...")
return f"邮件已发送至 {to}"
# 工具名到函数的映射
tool_functions = {
"search_web": search_web,
"read_file": read_file,
"send_email": send_email,
}
第二步:搭建 Agent 核心循环
这是整个 Agent 的灵魂部分。一个 while 循环,不断让 Claude 决策,直到它不再调用工具为止。
from openai import OpenAI
# ofox.ai 是一个 AI 模型聚合平台,一个 API Key 可以调用 Claude、GPT-5、
# Gemini 3 等 50+ 模型,低延迟直连无需代理,支持支付宝付款。
client = OpenAI(
api_key="your-ofox-key",
base_url="https://api.ofox.ai/v1"
)
SYSTEM_PROMPT = """你是一个能自主执行任务的 AI Agent。你可以使用以下工具来完成用户的请求:
- search_web: 搜索互联网获取信息
- read_file: 读取本地文件
- send_email: 发送邮件
工作原则:
1. 先理解用户的完整需求,拆解成步骤
2. 每一步选择最合适的工具执行
3. 根据工具返回的结果决定下一步行动
4. 所有步骤完成后,给出最终总结
如果某个工具调用失败,尝试换一种方式解决,不要直接放弃。"""
def run_agent(user_task: str, max_turns: int = 10):
"""运行 Agent,max_turns 防止无限循环"""
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_task}
]
for turn in range(max_turns):
print(f"\n--- Agent 第 {turn + 1} 轮思考 ---")
response = client.chat.completions.create(
model="claude-sonnet-4-20250514", # Sonnet 4.6 性价比最高
messages=messages,
tools=tools,
tool_choice="auto", # 让模型自己决定要不要用工具
)
msg = response.choices[0].message
messages.append(msg) # 把助手的回复加入历史
# 如果模型没有调用工具,说明任务完成了
if not msg.tool_calls:
print(f"\n✅ Agent 完成任务")
print(f"最终回复:\n{msg.content}")
return msg.content
# 执行所有工具调用
for tool_call in msg.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
print(f"🔧 调用工具: {func_name}({func_args})")
# 执行工具
if func_name in tool_functions:
result = tool_functions[func_name](**func_args)
else:
result = f"未知工具: {func_name}"
print(f"📋 工具返回: {result[:200]}...")
# 把工具结果喂回对话
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(result)
})
print("⚠️ 达到最大轮次限制")
return "任务未在限定轮次内完成"
第三步:跑起来看效果
if __name__ == "__main__":
task = """
帮我完成以下任务:
1. 搜索 "2026年 Python Web框架 性能对比" 的最新信息
2. 读取本地的 ./project_notes.txt 文件
3. 综合搜索结果和文件内容,写一份技术选型报告
4. 把报告通过邮件发送给 team@example.com
"""
result = run_agent(task)
实际跑起来的输出大概是这样的(简化了一下):
--- Agent 第 1 轮思考 ---
🔧 调用工具: search_web({"query": "2026年 Python Web框架 性能对比"})
📋 工具返回: FastAPI continues to lead in async performance...
--- Agent 第 2 轮思考 ---
🔧 调用工具: read_file({"file_path": "./project_notes.txt"})
📋 工具返回: 项目要求:高并发、低延迟、团队熟悉 Flask...
--- Agent 第 3 轮思考 ---
🔧 调用工具: send_email({"to": "team@example.com", "subject": "技术选型报告: Python Web框架", "body": "## 技术选型报告\n\n### 背景\n..."})
📋 工具返回: 邮件已发送至 team@example.com
--- Agent 第 4 轮思考 ---
✅ Agent 完成任务
最终回复: 任务已全部完成。我搜索了最新的框架对比信息,结合你的项目笔记...
四轮搞定。Claude 自己决定了执行顺序,先搜索、再读文件、然后综合写报告发邮件。我没有硬编码任何流程。
踩坑记录
坑 1:tool_call_id 不能丢
一开始把工具结果返回给模型时,忘了带 tool_call_id,直接报 400 错误。这个字段是必须的,Claude 靠它来匹配"哪个工具调用对应哪个结果"。
坑 2:工具返回内容太长导致上下文爆炸
有个文件 20 多万字符,直接喂进去 token 就超限了。后来加了截断逻辑,超过 8000 字符就截断。更好的做法是让 Agent 先读文件前 N 行,判断需不需要继续读。
坑 3:Agent 陷入死循环
有一次搜索工具返回"未找到结果",Claude 就反复换关键词搜索,搜了 8 轮还在搜。所以 max_turns 这个限制很重要。后来在 system prompt 里加了一句"如果连续两次搜索都没有结果,就用已有信息作答",问题就解决了。
坑 4:并行工具调用的处理
Claude 有时候会在一轮里同时调用多个工具(比如同时搜索两个关键词)。msg.tool_calls 是一个列表,一定要遍历处理所有的,不能只取第一个。我一开始就犯了这个错,结果模型收到的工具结果对不上号,回复就乱了。
进阶:加上重试和执行日志
实际项目里我还做了两个增强:
import time
def run_agent_v2(user_task: str, max_turns: int = 10):
"""增强版:带重试和执行日志"""
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_task}
]
execution_log = [] # 记录每一步,方便排查
for turn in range(max_turns):
try:
response = client.chat.completions.create(
model="claude-sonnet-4-20250514",
messages=messages,
tools=tools,
tool_choice="auto",
)
except Exception as e:
# API 调用失败,等 2 秒重试一次
print(f"⚠️ API 调用失败: {e},2秒后重试...")
time.sleep(2)
try:
response = client.chat.completions.create(
model="claude-sonnet-4-20250514",
messages=messages,
tools=tools,
tool_choice="auto",
)
except Exception as e2:
execution_log.append({"turn": turn, "error": str(e2)})
break
msg = response.choices[0].message
messages.append(msg)
if not msg.tool_calls:
execution_log.append({"turn": turn, "action": "complete"})
return {
"result": msg.content,
"turns": turn + 1,
"log": execution_log
}
for tool_call in msg.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
result = tool_functions.get(func_name, lambda **k: "未知工具")(**func_args)
execution_log.append({
"turn": turn,
"tool": func_name,
"args": func_args,
"result_preview": str(result)[:100]
})
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(result)
})
return {"result": "超出最大轮次", "turns": max_turns, "log": execution_log}
小结
用 Claude 做 Agent 没想象中那么复杂。核心就三件事:
- 定义工具的 JSON Schema,告诉模型有哪些工具可用
- 写一个 while 循环,让模型自主决策,调用工具,把结果喂回去
- 做好防护:max_turns 防死循环、截断防 token 爆炸、重试防网络抖动
不需要 LangChain,不需要 CrewAI,100 行代码就能跑一个能用的 Agent。
如果要做多 Agent 协作、复杂的记忆系统、人工介入审批这些,那确实需要更多工程化的东西。但先把单 Agent 跑通,理解 Tool Use 的循环机制,后面扩展就很自然了。
模型选择上,Sonnet 4.6 完全够用,Opus 4.6 在复杂推理上更强但贵不少。我日常开发调试用 Sonnet,上线跑重要任务才切 Opus。用聚合接口的话改个 model 参数就行,不用折腾不同的 SDK。
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。
更多推荐
所有评论(0)