minimal-agent.py 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149
  1. #!/usr/bin/env python3
  2. """
  3. 最小 Agent 模板 - 可复制后按需定制。
  4. 这是最简单的可运行 Agent(约 80 行)。
  5. 它包含最基本的三件东西:3 个工具 + 循环。
  6. 用法:
  7. 1. 设置 ANTHROPIC_API_KEY 环境变量
  8. 2. python minimal-agent.py
  9. 3. 输入任务,输入 q 退出
  10. """
  11. from anthropic import Anthropic
  12. from pathlib import Path
  13. import subprocess
  14. import os
  15. # 配置
  16. client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
  17. MODEL = os.getenv("MODEL_NAME", "claude-sonnet-4-20250514")
  18. WORKDIR = Path.cwd()
  19. # 系统提示词:保持简单
  20. SYSTEM = f"""你是位于 {WORKDIR} 的编码 Agent。
  21. 规则:
  22. - 使用工具完成任务
  23. - 优先行动,而不是只解释
  24. - 完成后总结你做了什么"""
  25. # 最小工具集:可按需继续添加
  26. TOOLS = [
  27. {
  28. "name": "bash",
  29. "description": "运行 shell 命令。",
  30. "input_schema": {
  31. "type": "object",
  32. "properties": {"command": {"type": "string"}},
  33. "required": ["command"]
  34. }
  35. },
  36. {
  37. "name": "read_file",
  38. "description": "读取文件内容。",
  39. "input_schema": {
  40. "type": "object",
  41. "properties": {"path": {"type": "string"}},
  42. "required": ["path"]
  43. }
  44. },
  45. {
  46. "name": "write_file",
  47. "description": "向文件写入内容。",
  48. "input_schema": {
  49. "type": "object",
  50. "properties": {
  51. "path": {"type": "string"},
  52. "content": {"type": "string"}
  53. },
  54. "required": ["path", "content"]
  55. }
  56. },
  57. ]
  58. def execute_tool(name: str, args: dict) -> str:
  59. """执行工具并返回结果。"""
  60. if name == "bash":
  61. try:
  62. r = subprocess.run(
  63. args["command"], shell=True, cwd=WORKDIR,
  64. capture_output=True, text=True, timeout=60
  65. )
  66. return (r.stdout + r.stderr).strip() or "(无输出)"
  67. except subprocess.TimeoutExpired:
  68. return "错误:执行超时"
  69. if name == "read_file":
  70. try:
  71. return (WORKDIR / args["path"]).read_text()[:50000]
  72. except Exception as e:
  73. return f"错误:{e}"
  74. if name == "write_file":
  75. try:
  76. p = WORKDIR / args["path"]
  77. p.parent.mkdir(parents=True, exist_ok=True)
  78. p.write_text(args["content"])
  79. return f"已写入 {len(args['content'])} 字节到 {args['path']}"
  80. except Exception as e:
  81. return f"错误:{e}"
  82. return f"未知工具:{name}"
  83. def agent(prompt: str, history: list = None) -> str:
  84. """运行 Agent 循环。"""
  85. if history is None:
  86. history = []
  87. history.append({"role": "user", "content": prompt})
  88. while True:
  89. response = client.messages.create(
  90. model=MODEL,
  91. system=SYSTEM,
  92. messages=history,
  93. tools=TOOLS,
  94. max_tokens=8000,
  95. )
  96. # 构建 assistant 消息
  97. history.append({"role": "assistant", "content": response.content})
  98. # 如果没有工具调用,直接返回文本
  99. if response.stop_reason != "tool_use":
  100. return "".join(b.text for b in response.content if hasattr(b, "text"))
  101. # 执行工具
  102. results = []
  103. for block in response.content:
  104. if block.type == "tool_use":
  105. print(f"> {block.name}: {block.input}")
  106. output = execute_tool(block.name, block.input)
  107. print(f" {output[:100]}...")
  108. results.append({
  109. "type": "tool_result",
  110. "tool_use_id": block.id,
  111. "content": output
  112. })
  113. history.append({"role": "user", "content": results})
  114. if __name__ == "__main__":
  115. print(f"最小 Agent - {WORKDIR}")
  116. print("输入 q 退出。\n")
  117. history = []
  118. while True:
  119. try:
  120. query = input(">> ").strip()
  121. except (EOFError, KeyboardInterrupt):
  122. break
  123. if query in ("q", "quit", "exit", ""):
  124. break
  125. print(agent(query, history))
  126. print()