Claude 系列模型对结构化 Prompt 特别敏感。好的 Agent Prompt 需要具备以下四要素：
| 特性 | Claude | GPT 系列 | 开源模型 |
|---|---|---|---|
| 结构化指令遵循 | ⭐⭐⭐⭐⭐ 极强 | ⭐⭐⭐⭐ 强 | ⭐⭐⭐ 中等 |
| 角色扮演稳定性 | ⭐⭐⭐⭐⭐ 极强 | ⭐⭐⭐⭐ 强 | ⭐⭐ 较弱 |
| 工具调用格式 | 偏好 JSON，容错好 | 支持 Function Calling | 不稳定 |
| 对示例的依赖 | 低（零样本强） | 低 | 高（需要 Few-shot） |
| Prompt 长度敏感 | 适中 | 适中 | 长 Prompt 容易跑偏 |
Claude 最喜欢"说清楚 + 给自由"。即：把规则和约束说得很清楚，但在具体执行上给 Claude 自主发挥的空间。最忌讳的是"每一步都精确控制"（那样不如直接写脚本），和"什么都不说靠 AI 猜"（那样幻觉多）。
AGENT_PROMPT_BASE = """## 角色
你是 {agent_name}，一个专业的 AI 智能助手。你的目标是帮助用户高效完成任务。
## 核心能力
- 理解用户的自然语言指令，提取关键信息
- 选择合适的工具来完成任务
- 在工具不可用时，诚实告知并尝试替代方案
- 记住对话中的重要信息（用户偏好、任务上下文）
## 行为规范
- 对于你确定的事情，自信而简洁地回答
- 对于你不确定的事情，明确说明"我不确定"，不要猜测
- 当需要用户提供更多信息时，友好地追问，而非胡编
- 优先使用中文回复，除非用户要求其他语言
## 工具使用
{tools_description}
## 输出格式
- 普通回复：直接输出答案
- 工具调用：输出 JSON: {{"tool": "工具名", "args": {{"key": "value"}}}}
## 错误处理
- 工具失败时，分析原因并尝试1次修正
- 如果修正后仍失败，向用户说明情况并提供建议
"""ROLE_SPECIFIC_PROMPT = """## 🎭 角色：{role_name}
### 身份背景
{role_background}
### 专业领域
{expertise_domains}
### 说话风格
- 语气: {tone}（如专业严谨、轻松友好、学术正式）
- 用词: {vocabulary_level}（如通俗易懂、专业术语、混合）
### 核心原则
1. {principle_1}
2. {principle_2}
3. {principle_3}
### 当遇到以下情况时：
- {scenario_a}: {response_strategy_a}
- {scenario_b}: {response_strategy_b}
- 不确定的情况: 诚实说明并追问
### 禁止行为
- {forbidden_behavior_1}
- {forbidden_behavior_2}
"""REASONING_PROMPT = """请在回答前先进行逐步推理，按以下格式：
<thinking>
步骤1: 分析用户意图...
步骤2: 确定需要的工具/信息...
步骤3: 制定执行策略...
步骤4: 考虑可能的异常情况...
</thinking>
然后给出最终回答。
注意：<thinking> 部分是对你思考过程的内部记录，
最终回答中不要包含这部分的内容。"""| 技巧 | 实现方式 | 效果 |
|---|---|---|
| 引用约束 | "只使用已提供的信息回答。如果没有相关信息，说'我无法确定'，不要编造。" | 大幅减少编造事实 |
| 不确定性表达 | "对不确定的内容，使用'可能'、'据我了解'等限定词，并标注不确定程度。" | 用户更容易分辨可信度 |
| 输出校验 | "回答后，请自我检查：1) 所有数字是否正确 2) 所有'事实'是否有依据 3) 逻辑是否自洽" | 自我纠错，减少低级错误 |
| Few-shot 示例 | 在 Prompt 中给出2-3个正确格式的示例 | 输出格式更稳定 |
| 负向提示 | "不要用'作为AI'这样的开头。不要说'我无法执行实际的操作'，直接调用工具。" | 减少模板化拒绝 |
整合本阶段四大能力（记忆+工具+规划+Prompt），开发一个个人智能文档整理 Agent —— DocOrganizer。
#!/usr/bin/env python3
"""
DocOrganizer — 个人智能文档整理 Agent
========================================
整合：记忆机制 + 工具调用 + 任务规划 + Prompt工程
运行：python3 doc_organizer.py
"""
import json
import os
import shutil
from datetime import datetime
from pathlib import Path
from collections import Counter
from dotenv import load_dotenv
from anthropic import Anthropic
load_dotenv()
# ================================================================
# Prompt 模板（整合四大能力的系统提示词）
# ================================================================
DOC_ORGANIZER_PROMPT = """## 🎭 角色
你是 DocOrganizer，一个专业的个人文档整理助手。你帮助用户整理、分类、归档文件。
## 📏 核心能力与约束
- 分析文件内容和类型，给出分类建议
- 为文档生成简洁摘要
- 记住用户的整理习惯和偏好
- 仅处理支持的文件类型：.txt .md .py .json .csv .log
## 🔧 可用工具
1. list_files — 列出目录下的文件，参数: directory_path
2. read_file — 读取文件内容，参数: filepath
3. classify_document — 分类文档，参数: filename|content
4. summarize_document — 摘要文档，参数: filename|content
5. move_file — 移动文件到分类目录，参数: filepath|category
6. save_preference — 保存用户偏好，参数: key=value
7. generate_report — 生成整理报告，参数: directory_path
## 📤 输出规范
- 工具调用：严格的 JSON 格式: {"tool": "工具名", "args": {"参数名": "值"}}
- 普通回复：简洁明了，一次不要超过3句话
- 报告：Markdown 格式
## 🧭 思考引导
1. 先理解用户意图（单文件整理？批量整理？查看报告？）
2. 如需批量处理，先列计划再逐步执行
3. 每步执行后验证结果
4. 自动记录用户的整理偏好（如"代码文件归到 projects/ 下"）
"""
class DocOrganizer:
"""文档整理 Agent"""
def __init__(self, workspace: str = "./documents"):
self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
self.model = "claude-sonnet-4-6"
self.workspace = Path(workspace)
self.workspace.mkdir(parents=True, exist_ok=True)
# 记忆：用户偏好
self.preferences_file = self.workspace / ".organizer_prefs.json"
self.preferences = self._load_preferences()
# 分类目录
self.categories = ["notes", "code", "logs", "reports", "data", "archived"]
for cat in self.categories:
(self.workspace / cat).mkdir(exist_ok=True)
# 操作日志
self.operation_log: list[dict] = []
print(f"📂 工作目录: {self.workspace.absolute()}")
print(f"📁 分类目录: {', '.join(self.categories)}")
def _load_preferences(self) -> dict:
if self.preferences_file.exists():
return json.loads(self.preferences_file.read_text("utf-8"))
return {"rules": {}, "favorite_categories": [], "stats": {}}
def _save_preferences(self):
self.preferences_file.write_text(
json.dumps(self.preferences, ensure_ascii=False, indent=2),
encoding="utf-8"
)
# --- 工具：列出文件 ---
def tool_list_files(self, directory: str = ".") -> str:
target = self.workspace / directory if directory != "." else self.workspace
if not target.exists():
return f"❌ 目录不存在: {target}"
files = []
for f in target.iterdir():
if f.is_file() and not f.name.startswith("."):
size_kb = f.stat().st_size / 1024
files.append(f" 📄 {f.name} ({size_kb:.1f}KB)")
if not files:
return "📂 目录为空（无可整理文件）"
return "📂 文件列表:\n" + "\n".join(sorted(files))
# --- 工具：分类文档 ---
def tool_classify(self, filepath: str) -> str:
"""用简单规则 + Claude 辅助分类"""
path = self.workspace / filepath
if not path.exists():
return f"❌ 文件不存在: {filepath}"
ext = path.suffix.lower()
# 先按扩展名快速分类
ext_map = {".py": "code", ".js": "code", ".json": "data",
".csv": "data", ".log": "logs", ".md": "notes", ".txt": "notes"}
if ext in ext_map:
category = ext_map[ext]
self.preferences.setdefault("ext_rules", {})[ext] = category
return f"🏷️ {path.name} → [自动] {category} (按扩展名 {ext})"
# 扩展名未知，读取内容让 Claude 判断
try:
content = path.read_text("utf-8")[:500]
response = self.client.messages.create(
model=self.model,
max_tokens=50,
messages=[{"role": "user", "content":
f"将以下文件内容分类为: notes/code/logs/reports/data/archived。只输出类别名。\n\n{content}"}],
temperature=0.1,
)
category = response.content[0].text.strip().lower()
if category in self.categories:
return f"🏷️ {path.name} → [AI判断] {category}"
except Exception:
pass
return f"🏷️ {path.name} → [默认] archived"
# --- 工具：生成摘要 ---
def tool_summarize(self, filepath: str) -> str:
path = self.workspace / filepath
if not path.exists():
return f"❌ 文件不存在: {filepath}"
try:
content = path.read_text("utf-8")[:3000]
response = self.client.messages.create(
model=self.model,
max_tokens=150,
messages=[{"role": "user", "content":
f"请用1-2句话总结以下文档的核心内容：\n\n{content}"}],
temperature=0.2,
)
summary = response.content[0].text.strip()
return f"📝 {path.name} 摘要: {summary}"
except Exception as e:
return f"❌ 摘要生成失败: {e}"
# --- 工具：移动文件 ---
def tool_move_file(self, filepath: str, category: str) -> str:
if category not in self.categories:
return f"❌ 无效分类: {category}。可选: {self.categories}"
path = self.workspace / filepath
if not path.exists():
return f"❌ 文件不存在: {filepath}"
dest_dir = self.workspace / category
dest = dest_dir / path.name
# 处理重名
if dest.exists():
dest = dest_dir / f"{path.stem}_{datetime.now().strftime('%H%M%S')}{path.suffix}"
shutil.move(str(path), str(dest))
self.operation_log.append({"action": "move", "file": filepath, "to": str(dest)})
return f"✅ {path.name} → {category}/"
# --- 工具：生成报告 ---
def tool_generate_report(self, directory: str = ".") -> str:
target = self.workspace / directory if directory != "." else self.workspace
report_lines = [
f"# 📊 文档整理报告",
f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}",
f"扫描目录: {target.absolute()}",
f"",
]
# 按分类统计
category_counts = Counter()
total_size = 0
for cat in self.categories:
cat_dir = self.workspace / cat
if cat_dir.exists():
files = list(cat_dir.iterdir())
category_counts[cat] = len([f for f in files if not f.name.startswith(".")])
total_size += sum(f.stat().st_size for f in files if f.is_file())
report_lines.append("## 分类统计")
for cat in self.categories:
count = category_counts.get(cat, 0)
bar = "█" * min(count, 20)
report_lines.append(f"- **{cat}**: {count} 个文件 {bar}")
report_lines.append(f"\n**总计**: {sum(category_counts.values())} 个文件, {total_size/1024:.1f}KB")
report_lines.append("\n## 最近操作")
for op in self.operation_log[-10:]:
report_lines.append(f"- {op['action']}: {op.get('file', '')} → {op.get('to', '')}")
if self.preferences.get("rules"):
report_lines.append("\n## 用户偏好规则")
for rule, value in self.preferences["rules"].items():
report_lines.append(f"- {rule}: {value}")
report_text = "\n".join(report_lines)
report_path = self.workspace / f"report_{datetime.now().strftime('%Y%m%d')}.md"
report_path.write_text(report_text, encoding="utf-8")
return f"📊 报告已生成: {report_path.name}\n\n{report_text[:500]}"
# --- 核心运行逻辑 ---
def _call_claude(self, user_input: str, context: str = "") -> str:
system = DOC_ORGANIZER_PROMPT + f"\n\n当前工作目录: {self.workspace.absolute()}"
if context:
system += f"\n\n上次操作结果: {context}"
try:
response = self.client.messages.create(
model=self.model,
max_tokens=1024,
system=system,
messages=[{"role": "user", "content": user_input}],
temperature=0.2,
)
return response.content[0].text.strip()
except Exception as e:
return f'{{"tool": "error", "args": {{"message": "{str(e)}"}}}}'
def _execute_tool(self, tool_name: str, args: dict) -> str:
tools = {
"list_files": lambda a: self.tool_list_files(a.get("directory", ".")),
"read_file": lambda a: self.tool_read_file(a.get("filepath", "")),
"classify_document": lambda a: self.tool_classify(a.get("filepath", "")),
"summarize_document": lambda a: self.tool_summarize(a.get("filepath", "")),
"move_file": lambda a: self.tool_move_file(a.get("filepath", ""), a.get("category", "archived")),
"save_preference": lambda a: self._save_pref(a),
"generate_report": lambda a: self.tool_generate_report(a.get("directory", ".")),
}
func = tools.get(tool_name)
if func:
return func(args)
return f"❌ 未知工具: {tool_name}"
def _save_pref(self, args: dict) -> str:
for k, v in args.items():
self.preferences["rules"][k] = v
self._save_preferences()
return f"✅ 偏好已保存: {args}"
def tool_read_file(self, filepath: str) -> str:
path = self.workspace / filepath
if not path.exists():
return f"❌ 文件不存在: {filepath}"
try:
return path.read_text("utf-8")[:2000]
except Exception as e:
return f"❌ {e}"
def run(self, user_input: str):
"""运行一轮文档整理操作"""
print(f"\n👤 用户: {user_input}")
claude_reply = self._call_claude(user_input)
# 尝试解析工具调用
try:
json_str = claude_reply
for marker in ["```json", "```"]:
if marker in claude_reply:
json_str = claude_reply.split(marker)[1].split("```")[0]
break
parsed = json.loads(json_str.strip())
if "tool" in parsed:
print(f"🔧 调用: {parsed['tool']}({parsed.get('args', {})})")
result = self._execute_tool(parsed["tool"], parsed.get("args", {}))
print(f"🤖 DocOrganizer: {result}")
# 多步任务的循环处理
if "继续" in user_input.lower() or "批量" in user_input.lower():
print(" 📋 批量任务可能需要多步处理...")
return result
except (json.JSONDecodeError, KeyError):
pass
print(f"🤖 DocOrganizer: {claude_reply}")
return claude_reply
def main():
print("=" * 60)
print("📂 DocOrganizer — 个人智能文档整理 Agent")
print("=" * 60)
print("试试这些命令：")
print(" • 列出当前目录所有文件")
print(" • 整理所有文件（自动分类+归档）")
print(" • 分析 test.py 并归类")
print(" • 生成整理报告")
print(" • 偏好: 把所有 .md 文件归到 notes")
print(" • quit → 退出")
print("=" * 60)
if not os.getenv("ANTHROPIC_API_KEY"):
print("❌ 请先配置 ANTHROPIC_API_KEY！")
return
# 创建示例文档
workspace = Path("./documents")
workspace.mkdir(exist_ok=True)
(workspace / "学习笔记.md").write_text("# AI Agent 学习笔记\n\n1. Agent 有记忆、工具、规划能力\n2. Claude Code 是优秀的 Agent 开发平台\n", encoding="utf-8")
(workspace / "app.py").write_text("# My App\n\ndef main():\n print('Hello Agent')\n\nif __name__ == '__main__':\n main()\n", encoding="utf-8")
(workspace / "server.log").write_text("2024-01-15 10:00 INFO Server started\n2024-01-15 10:05 ERROR Connection refused\n", encoding="utf-8")
print("📝 已创建示例文档用于学习测试")
agent = DocOrganizer()
while True:
try:
user_input = input("\n👤 你: ").strip()
if not user_input:
continue
if user_input.lower() in ("quit", "exit", "q"):
break
agent.run(user_input)
except KeyboardInterrupt:
break
print(f"\n📊 本次操作数: {len(agent.operation_log)}")
print(f"👋 DocOrganizer 已退出。")
if __name__ == "__main__":
main()
| 能力 | 核心技术 | 实现效果 |
|---|---|---|
| 记忆机制 | ShortTermMemory + LongTermMemory + SessionMemory 三层结构 | Agent 能记住对话上下文和用户偏好，跨会话持久化 |
| 工具调用 | ToolRegistry 注册中心 + 自动发现 + 失败重试 | Agent 自主选择工具、自动纠错、多工具协同 |
| 任务规划 | TaskPlanner + 链式思考 + 迭代调整 | 复杂任务自动拆分→分步执行→结果汇总 |
| Prompt 工程 | 四要素模板 + 推理引导 + 幻觉抑制 | Agent 输出稳定、格式规范、可预期 |