| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149 |
- #!/usr/bin/env python3
- """
- 最小 Agent 模板 - 可复制后按需定制。
- 这是最简单的可运行 Agent(约 80 行)。
- 它包含最基本的三件东西:3 个工具 + 循环。
- 用法:
- 1. 设置 ANTHROPIC_API_KEY 环境变量
- 2. python minimal-agent.py
- 3. 输入任务,输入 q 退出
- """
- from anthropic import Anthropic
- from pathlib import Path
- import subprocess
- import os
- # 配置
- client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
- MODEL = os.getenv("MODEL_NAME", "claude-sonnet-4-20250514")
- WORKDIR = Path.cwd()
- # 系统提示词:保持简单
- SYSTEM = f"""你是位于 {WORKDIR} 的编码 Agent。
- 规则:
- - 使用工具完成任务
- - 优先行动,而不是只解释
- - 完成后总结你做了什么"""
- # 最小工具集:可按需继续添加
- TOOLS = [
- {
- "name": "bash",
- "description": "运行 shell 命令。",
- "input_schema": {
- "type": "object",
- "properties": {"command": {"type": "string"}},
- "required": ["command"]
- }
- },
- {
- "name": "read_file",
- "description": "读取文件内容。",
- "input_schema": {
- "type": "object",
- "properties": {"path": {"type": "string"}},
- "required": ["path"]
- }
- },
- {
- "name": "write_file",
- "description": "向文件写入内容。",
- "input_schema": {
- "type": "object",
- "properties": {
- "path": {"type": "string"},
- "content": {"type": "string"}
- },
- "required": ["path", "content"]
- }
- },
- ]
- def execute_tool(name: str, args: dict) -> str:
- """执行工具并返回结果。"""
- if name == "bash":
- try:
- r = subprocess.run(
- args["command"], shell=True, cwd=WORKDIR,
- capture_output=True, text=True, timeout=60
- )
- return (r.stdout + r.stderr).strip() or "(无输出)"
- except subprocess.TimeoutExpired:
- return "错误:执行超时"
- if name == "read_file":
- try:
- return (WORKDIR / args["path"]).read_text()[:50000]
- except Exception as e:
- return f"错误:{e}"
- if name == "write_file":
- try:
- p = WORKDIR / args["path"]
- p.parent.mkdir(parents=True, exist_ok=True)
- p.write_text(args["content"])
- return f"已写入 {len(args['content'])} 字节到 {args['path']}"
- except Exception as e:
- return f"错误:{e}"
- return f"未知工具:{name}"
- def agent(prompt: str, history: list = None) -> str:
- """运行 Agent 循环。"""
- if history is None:
- history = []
- history.append({"role": "user", "content": prompt})
- while True:
- response = client.messages.create(
- model=MODEL,
- system=SYSTEM,
- messages=history,
- tools=TOOLS,
- max_tokens=8000,
- )
- # 构建 assistant 消息
- history.append({"role": "assistant", "content": response.content})
- # 如果没有工具调用,直接返回文本
- if response.stop_reason != "tool_use":
- return "".join(b.text for b in response.content if hasattr(b, "text"))
- # 执行工具
- results = []
- for block in response.content:
- if block.type == "tool_use":
- print(f"> {block.name}: {block.input}")
- output = execute_tool(block.name, block.input)
- print(f" {output[:100]}...")
- results.append({
- "type": "tool_result",
- "tool_use_id": block.id,
- "content": output
- })
- history.append({"role": "user", "content": results})
- if __name__ == "__main__":
- print(f"最小 Agent - {WORKDIR}")
- print("输入 q 退出。\n")
- history = []
- while True:
- try:
- query = input(">> ").strip()
- except (EOFError, KeyboardInterrupt):
- break
- if query in ("q", "quit", "exit", ""):
- break
- print(agent(query, history))
- print()
|