从零构建 AI Agent：工具篇

def run_bash(command: str) -> str:
    """Run a bash command and return its output."""
    result = subprocess.run(
        command, shell=True, text=True, capture_output=True
    )
    output = result.stdout
    if result.stderr:
        output += f"\nSTDERR:\n{result.stderr}"
    return output or "(no output)"

def read_file(path: str, offset: int = 1, limit: int = 200) -> str:
    """Read lines from a file, with optional offset and limit."""
    p = Path(path)
    if not p.exists():
        return f"Error: file not found: {path}"
    lines = p.read_text(errors="replace").splitlines()
    selected = lines[offset - 1: offset - 1 + limit]
    return "\n".join(f"{offset + i}: {line}" for i, line in enumerate(selected))

def glob_files(pattern: str, path: str = ".") -> str:
    """Find files matching a glob pattern inside a directory."""
    matches = glob_module.glob(f"{path}/**/{pattern}", recursive=True)
    matches += glob_module.glob(f"{path}/{pattern}")
    unique = sorted(set(matches))
    return "\n".join(unique) if unique else "(no matches)"

def grep(pattern: str, path: str = ".", include: str = "*") -> str:
    """Search file contents for a regex pattern."""
    results = []
    for filepath in glob_module.glob(f"{path}/**/{include}", recursive=True):
        fp = Path(filepath)
        if not fp.is_file():
            continue
        try:
            for i, line in enumerate(fp.read_text(errors="replace").splitlines(), 1):
                if re.search(pattern, line):
                    results.append(f"{filepath}:{i}: {line}")
        except OSError:
            pass
    return "\n".join(results) if results else "(no matches)"

def write_file(path: str, content: str) -> str:
    """Write content to a file, creating parent directories if needed."""
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(content)
    return f"Wrote {len(content)} bytes to {path}"

def edit_file(path: str, old_string: str, new_string: str) -> str:
    """Replace the first occurrence of old_string with new_string in a file."""
    p = Path(path)
    if not p.exists():
        return f"Error: file not found: {path}"
    original = p.read_text()
    if old_string not in original:
        return f"Error: string not found in {path}"
    p.write_text(original.replace(old_string, new_string, 1))
    return f"Edited {path}"

def webfetch(url: str) -> str:
    """Fetch a URL and return its plain-text content (up to 2 MB)."""
    parsed = urlparse(url)
    if parsed.scheme not in ("http", "https"):
        return f"Error fetching {url}: unsupported scheme '{parsed.scheme}'."
    req = urllib.request.Request(url, headers={"User-Agent": "agent/1.0"})
    with urllib.request.urlopen(req, timeout=15) as resp:
        raw = b"".join(...).decode(charset, errors="replace")
    soup = BeautifulSoup(raw, "html.parser")
    text = soup.get_text(separator="\n", strip=True)
    return re.sub(r"\n{3,}", "\n\n", text).strip()

def get_tool_schemas():
    return [
        {
            "type": "function",
            "function": {
                "name": "run_bash",
                "description": "Run a bash command on the user's machine and return the output.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "command": {
                            "type": "string",
                            "description": "The bash command to execute.",
                        }
                    },
                    "required": ["command"],
                },
            },
        },
        # ... 其他工具的 schema 类似
        {
            "type": "function",
            "function": {
                "name": "webfetch",
                "description": "Fetch a URL and return its plain-text content.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "url": {
                            "type": "string",
                            "description": "The URL to fetch.",
                        }
                    },
                    "required": ["url"],
                },
            },
        },
    ]

def agent_loop():
    messages = []
    tools = get_tool_schemas()
    tool_map = {
        "run_bash": run_bash,
        "read_file": read_file,
        "glob_files": glob_files,
        "grep": grep,
        "write_file": write_file,
        "edit_file": edit_file,
        "webfetch": webfetch,
    }
 
    while True:
        user_input = input("You: ")
        if user_input.lower() in ("exit", "quit"):
            break
        messages.append({"role": "user", "content": user_input})
 
        while True:
            response = client.chat.completions.create(
                model="your-model",
                messages=messages,
                tools=tools,
            )
            choice = response.choices[0]
 
            if choice.finish_reason == "tool_calls":
                for tc in choice.message.tool_calls:
                    fn_name = tc.function.name
                    args = json.loads(tc.function.arguments)
                    result = tool_map[fn_name](**args)
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tc.id,
                        "content": result,
                    })
            else:
                print(f"Agent: {choice.message.content}")
                messages.append(choice.message)
                break