开发一个「自主问答 + 简单任务执行」极简 Agent，命名为 MiniAgent。
在项目目录 ~/ai-agent-study/ 下创建 mini_agent.py：
#!/usr/bin/env python3
"""
MiniAgent — 你的第一个 AI Agent
=================================
功能：
1. 接收用户输入
2. 自主判断：简单问答 vs 需要执行任务
3. 问答直接回复；任务调用工具执行
4. 返回执行结果
依赖：anthropic, python-dotenv, requests
运行：python3 mini_agent.py
"""
# ============================================================
# 第1部分：导入依赖与配置
# ============================================================
import os
import sys
import json
from datetime import datetime
from pathlib import Path
from dotenv import load_dotenv
from anthropic import Anthropic
# 加载配置
load_dotenv()
# ============================================================
# 第2部分：Agent 类定义
# ============================================================
class MiniAgent:
"""
极简 AI Agent：
- brain: Claude 大模型（负责"思考"）
- tools: 工具字典（负责"执行"）
- memory: 对话历史列表（负责"记忆"）
"""
def __init__(self, name: str = "MiniAgent"):
"""初始化 Agent：设置大脑、记忆、工具"""
self.name = name
# 🧠 大脑：连接 Claude API
self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
self.model = "claude-sonnet-4-6" # 性价比最优
# 📝 记忆：存储对话历史（短期记忆）
self.memory = []
# 🔧 工具注册：Agent 可以使用的工具
self.tools = {
"calculator": self._tool_calculator,
"get_time": self._tool_get_time,
"read_file": self._tool_read_file,
}
print(f"🤖 {self.name} 初始化完成！")
# ============================================================
# 第3部分：工具定义（Agent 的"双手"）
# ============================================================
def _tool_calculator(self, expression: str) -> str:
"""
🧮 计算器工具：安全地计算数学表达式
输入：数学表达式字符串，如 "2 + 3 * 4"
输出：计算结果
"""
try:
# ⚠️ 安全警告：eval 只能用于学习，生产环境必须用更安全的方式
# 这里限制只允许数字和基本运算符
allowed = set("0123456789+-*/().% ")
if not all(c in allowed for c in expression):
return "❌ 表达式包含不允许的字符，仅支持基本数学运算"
result = eval(expression)
return f"✅ 计算结果: {expression} = {result}"
except Exception as e:
return f"❌ 计算出错: {str(e)}"
def _tool_get_time(self, _: str = "") -> str:
"""
🕐 时间查询工具：获取当前日期时间
输入：忽略（不需要参数）
输出：当前时间的格式化字符串
"""
now = datetime.now()
weekday_names = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
wd = weekday_names[now.weekday()]
return f"🕐 当前时间: {now.strftime('%Y年%m月%d日')} {wd} {now.strftime('%H:%M:%S')}"
def _tool_read_file(self, filepath: str) -> str:
"""
📄 文件读取工具：读取指定文件内容
输入：文件路径
输出：文件内容（最多2000字符）
"""
path = Path(filepath)
if not path.exists():
return f"❌ 文件不存在: {filepath}"
try:
content = path.read_text(encoding="utf-8")
if len(content) > 2000:
content = content[:2000] + "\n... (内容过长，已截断)"
return f"📄 文件内容 ({filepath}):\n{content}"
except Exception as e:
return f"❌ 读取文件出错: {str(e)}"
# ============================================================
# 第4部分：工具选择逻辑
# ============================================================
def _get_tools_description(self) -> str:
"""生成工具描述，供 Claude 理解可用工具有哪些"""
return """
可用工具列表：
1. calculator — 计算数学表达式，参数: expression (如 "2+3*4")
2. get_time — 获取当前日期和时间，无需参数
3. read_file — 读取文件内容，参数: filepath (文件路径)
如果用户的问题需要计算、查时间、读文件，你必须输出如下格式的JSON来调用工具：
{"tool": "工具名", "args": {"参数名": "参数值"}}
如果用户的问题不需要工具（普通问答），直接回复即可，不要输出JSON。
"""
# ============================================================
# 第5部分：核心运行循环
# ============================================================
def run(self, user_input: str) -> str:
"""
Agent 的核心运行方法：
1. 让 Claude 分析用户输入
2. 判断需要直接回复还是调用工具
3. 如需工具，则调用工具并返回结果
"""
print(f"\n{'='*50}")
print(f"👤 用户: {user_input}")
print(f"{'='*50}")
# --- 步骤1：构建给 Claude 的消息 ---
system_prompt = f"""你是 {self.name}，一个智能助手。你的职责是帮助用户解决问题。
{self._get_tools_description()}
重要规则：
- 如果用户问当前时间、需要计算、或要求读取文件，你必须调用工具
- 调用工具时，只输出JSON，不要输出其他内容
- 如果只是普通问答，直接友好地回复
- 回复使用中文"""
messages = [{"role": "user", "content": user_input}]
# 如果有历史对话，也加入（让 Agent 有"记忆"）
if self.memory:
# 只保留最近5轮对话，防止上下文过长
recent = self.memory[-5:]
full_messages = []
for turn in recent:
full_messages.append({"role": "user", "content": turn["user"]})
full_messages.append({"role": "assistant", "content": turn["assistant"]})
full_messages.append({"role": "user", "content": user_input})
messages = full_messages
# --- 步骤2：调用 Claude 获取响应 ---
try:
response = self.client.messages.create(
model=self.model,
max_tokens=1024,
system=system_prompt,
messages=messages,
temperature=0.3, # 低温度让输出更稳定
)
reply = response.content[0].text.strip()
except Exception as e:
return f"❌ Agent 调用失败: {str(e)}"
# --- 步骤3：判断响应是工具调用还是直接回复 ---
result = self._process_response(reply, user_input)
# --- 步骤4：保存到记忆 ---
self.memory.append({
"user": user_input,
"assistant": result,
"timestamp": datetime.now().isoformat()
})
# 控制记忆长度（最多保留20轮）
if len(self.memory) > 20:
self.memory = self.memory[-20:]
print(f"🤖 {self.name}: {result}")
return result
def _process_response(self, reply: str, user_input: str) -> str:
"""
处理 Claude 的响应：
- 如果是 JSON 工具调用 → 执行工具 → 返回结果
- 如果是普通文本 → 直接返回
"""
# 尝试解析 JSON（工具调用）
try:
# 提取 JSON 部分（可能被包裹在代码块中）
json_str = reply
if "```json" in reply:
json_str = reply.split("```json")[1].split("```")[0]
elif "```" in reply:
json_str = reply.split("```")[1].split("```")[0]
tool_call = json.loads(json_str.strip())
if "tool" in tool_call:
tool_name = tool_call["tool"]
args = tool_call.get("args", {})
print(f"🔧 调用工具: {tool_name}")
print(f"📥 参数: {args}")
# 查找并执行工具
if tool_name in self.tools:
tool_func = self.tools[tool_name]
# 工具函数的第一个参数是主要参数
first_arg = list(args.values())[0] if args else ""
tool_result = tool_func(first_arg)
return tool_result
else:
return f"❌ 未知工具: {tool_name}，可用工具: {list(self.tools.keys())}"
except (json.JSONDecodeError, KeyError, IndexError):
pass # 不是工具调用，按普通回复处理
# 普通回复，直接返回
return reply
# ============================================================
# 第6部分：交互式运行入口
# ============================================================
def main():
"""Agent 交互式主循环"""
print("=" * 50)
print("🤖 MiniAgent — 你的第一个 AI Agent")
print("=" * 50)
print("支持功能：")
print(" • 普通问答（任何问题）")
print(" • 数学计算（如 '帮我算 156 * 23 + 89'）")
print(" • 时间查询（如 '现在几点了？'）")
print(" • 文件读取（如 '读取 test.txt 的内容'）")
print(" • 输入 'quit' 或 'exit' 退出")
print("=" * 50)
# 检查 API 密钥
if not os.getenv("ANTHROPIC_API_KEY"):
print("❌ 错误：未找到 ANTHROPIC_API_KEY！")
print(" 请在 .env 文件中配置你的 API 密钥")
sys.exit(1)
# 创建 Agent 实例
agent = MiniAgent()
# 创建测试文件（用于演示文件读取功能）
test_file = Path("test.txt")
if not test_file.exists():
test_file.write_text(
"你好！这是一个测试文件。\n"
"AI Agent 可以读取这个文件的内容。\n"
"试试输入：读取 test.txt 的内容\n",
encoding="utf-8"
)
print(f"📝 已创建测试文件: {test_file.absolute()}")
print("\n开始对话吧！\n")
# 主循环
while True:
try:
user_input = input("👤 你: ").strip()
if not user_input:
continue
if user_input.lower() in ("quit", "exit", "q"):
print(f"\n👋 {agent.name} 已退出。再见！")
break
agent.run(user_input)
except KeyboardInterrupt:
print(f"\n\n👋 {agent.name} 已中断。再见！")
break
except Exception as e:
print(f"❌ 运行出错: {e}")
# 显示会话统计
print(f"\n📊 本次会话统计：共 {len(agent.memory)} 轮对话")
if __name__ == "__main__":
main()
导入所需的 Python 库。核心是 anthropic（连接 Claude API）和 python-dotenv（从 .env 文件加载密钥）。from anthropic import Anthropic 创建 API 客户端。
__init__ 是 Agent 的"出生证明"：创建时完成3件事：
self.client = Anthropic(api_key=...) — 连接 Claude APIself.memory = [] — 一个空列表，后面存对话历史self.tools = {...} — 一个字典，把工具名字映射到实际函数三个工具函数就是普通的 Python 函数：
_tool_calculator — 用 eval() 计算，但加了安全过滤（只允许数字和运算符）_tool_get_time — 用 datetime.now() 获取当前时间_tool_read_file — 用 Path.read_text() 读取文件生产环境中绝对不能用 eval() 直接执行用户输入，这里仅为了简化示例加了字符白名单过滤。后续阶段我们会用更安全的方式。
_get_tools_description() 生成一段"工具说明书"，告诉 Claude 有哪些工具可用、每个工具的参数格式。这是 Agent 能"自主选择工具"的关键——Claude 读了这个说明书就知道什么时候该用什么工具。
run() 是 Agent 的"心跳"——每次用户输入都走这个流程：
self.client.messages.create(...) 把消息发给 Claude，拿到回复_process_response() 判断是工具调用还是直接回复self.memory_process_response() 的核心逻辑：尝试把 Claude 的回复解析成 JSON → 如果成功且包含 "tool" 字段，就调用对应工具 → 否则当普通回复处理。
main() 函数创建一个 Agent 实例，然后进入 while True 无限循环，不断等待用户输入 → 调用 agent.run() → 输出结果。
# 确保在项目目录且虚拟环境已激活
cd ~/ai-agent-study
source venv/bin/activate
# 运行 MiniAgent
python3 mini_agent.py
# 期望看到：
# ==================================================
# 🤖 MiniAgent — 你的第一个 AI Agent
# ==================================================
# 支持功能：...
# 📝 已创建测试文件: /home/user/ai-agent-study/test.txt
# 开始对话吧！
# 👤 你:# 测试1：普通问答
👤 你: 你好，请用一句话介绍你自己
# 测试2：数学计算
👤 你: 帮我计算 156 * 23 + 89
# 测试3：时间查询
👤 你: 现在几点了？
# 测试4：文件读取
👤 你: 读取 test.txt 的内容
# 测试5：带记忆的对话
👤 你: 我刚才问了你什么问题？
# Agent 应该能回忆之前的对话（因为 memory 里有记录）# 创建带详细日志的调试版本
cat > mini_agent_debug.py << 'PYEOF'
# 在原有代码基础上，在 run() 方法中添加日志
# （这里展示关键的调试代码片段）
import logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s [%(levelname)s] %(message)s',
handlers=[
logging.FileHandler('agent_debug.log'),
logging.StreamHandler()
]
)
# 在 run() 方法的各个关键步骤添加日志：
# logging.debug(f"发送给API的消息: {messages}")
# logging.debug(f"API原始响应: {reply}")
# logging.debug(f"当前记忆长度: {len(self.memory)}")
PYEOF
echo "调试版脚本已创建，需要时可自行整合完整代码。"👤 你: 帮我计算 35 * 12 + 100
=== 后台发生了什么？ ===
1️⃣ Agent 构建消息发给 Claude：
"你是 MiniAgent...可用工具：calculator、get_time、read_file..."
"用户说：帮我计算 35 * 12 + 100"
2️⃣ Claude 分析：用户要计算 → 需要调用 calculator 工具
Claude 返回JSON: {"tool": "calculator", "args": {"expression": "35 * 12 + 100"}}
3️⃣ Agent 解析 JSON → 找到 tool="calculator"
→ 调用 self.tools["calculator"]("35 * 12 + 100")
→ Python 计算：35*12=420, 420+100=520
4️⃣ Agent 返回结果给用户：
🤖 MiniAgent: ✅ 计算结果: 35 * 12 + 100 = 520
5️⃣ 保存到 memory（用户问了什么 + Agent 回了什么）run() 中限制了只保留最近5轮对话（self.memory[-5:]），并且总记忆不超过20轮。如果仍然超限，可以减少到3轮或缩短每轮的文本长度。run() 方法的 API 调用后加一行 print(f"DEBUG 原始响应: {reply}")。更规范的方式是用 Python 的 logging 模块（参考3.4节的调试版脚本）。| 知识点 | 在本项目中的体现 |
|---|---|
| Agent 大脑 | Anthropic() 客户端连接 Claude API |
| Agent 记忆 | self.memory 列表存储对话历史 |
| Agent 工具 | self.tools 字典，函数名 → 实际函数 |
| 工具选择 | Claude 分析用户输入 → 输出 JSON → Python 解析 → 执行 |
| 自主闭环 | 用户输入 → Claude思考 → 选择工具(或直接回复) → 执行 → 返回结果 → 记忆留存 |
word_count——输入一段文本，返回字数统计。提示：在 self.tools 字典中添加新条目，实现对应的函数。_process_response，增加 JSON 解析失败时的重试逻辑（让 Agent 重新生成）。