别只会调用大模型 API：用 Python 实现一个能干活的 AI Agent

大模型负责生成文字，AI Agent 负责完成任务。本文不使用复杂框架，从零实现一个能够自主选择工具、读取文件、查询时间并连续执行任务的最小 AI Agent。

前言：为什么 2026 年大家都在谈 Agent？

过去两年，我们写过太多这样的代码：向大模型发送一段提示词，然后打印回答。

response = client.chat.completions.create(
    model="your-model",
    messages=[{"role": "user", "content": "帮我分析这个项目"}],
)
print(response.choices[0].message.content)

这段代码能聊天，却不能真正分析项目，因为模型既看不到项目文件，也不能执行任何操作。它只能根据训练数据和提示词猜测。

AI Agent 的关键变化，是给大模型增加一个可控的“行动层”：模型可以判断当前需要什么信息，选择合适的工具，读取执行结果，再决定下一步做什么。

例如，当用户提出：

阅读 README.md，总结项目用途，并告诉我当前时间。

一个 Agent 可能按下面的顺序工作：

1. 判断需要读取文件；
2. 调用 read_file；
3. 获取文件内容并进行总结；
4. 判断还需要当前时间；
5. 调用 get_current_time；
6. 综合两次工具结果，生成最终回答。

这就是 Agent 最核心的闭环：

用户目标 -> 模型决策 -> 调用工具 -> 观察结果 -> 再次决策 -> 最终回答

 一、先设计两个安全工具

为了让示例容易运行，我们只提供两个工具：读取指定目录内的文本文件，以及获取当前时间。

项目结构如下：

```text
mini-agent/
├── agent.py
└── workspace/
    └── README.md

先实现工具函数：

from datetime import datetime
from pathlib import Path
from zoneinfo import ZoneInfo

WORKSPACE = Path(__file__).parent.joinpath("workspace").resolve()


def read_file(path: str) -> str:
    """读取工作目录内的 UTF-8 文本文件。"""
    target = WORKSPACE.joinpath(path).resolve()

    # 防止 ../../ 等路径穿越访问工作目录之外的文件
    if target != WORKSPACE and WORKSPACE not in target.parents:
        return "错误：只能读取 workspace 目录内的文件"
    if not target.is_file():
        return f"错误：文件不存在：{path}"
    if target.stat().st_size > 100_000:
        return "错误：文件超过 100 KB，拒绝读取"

    try:
        return target.read_text(encoding="utf-8")
    except UnicodeDecodeError:
        return "错误：当前示例只支持 UTF-8 文本文件"


def get_current_time(timezone: str = "Asia/Shanghai") -> str:
    """返回指定 IANA 时区的当前时间。"""
    try:
        now = datetime.now(ZoneInfo(timezone))
    except Exception:
        return f"错误：无效时区：{timezone}"
    return now.isoformat(timespec="seconds")

这里有一个很重要的细节：不要把整个文件系统直接开放给 Agent。

工具参数来自模型，而模型可能受到错误提示词或 Prompt Injection 的影响。因此，工具本身必须检查路径、文件大小和数据类型。安全边界应该写在工具代码里，不能只靠一句“请勿读取敏感文件”的提示词。

二、把工具描述交给大模型

大模型不会自动知道 Python 函数的存在，我们需要用 JSON Schema 描述工具名称、用途和参数。

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "读取 workspace 目录内的 UTF-8 文本文件",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "相对于 workspace 的文件路径",
                    }
                },
                "required": ["path"],
                "additionalProperties": False,
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "获取指定 IANA 时区的当前时间",
            "parameters": {
                "type": "object",
                "properties": {
                    "timezone": {
                        "type": "string",
                        "description": "例如 Asia/Shanghai 或 UTC",
                    }
                },
                "required": [],
                "additionalProperties": False,
            },
        },
    },
]

描述应当短而准确。如果多个工具的描述含糊或相互重叠，模型就更容易选错工具。

三、实现 Agent 的核心循环

下面使用兼容 Chat Completions 与 Function Calling 的 HTTP 接口。通过环境变量可以更换模型服务地址，不需要把密钥写进代码。

先安装依赖：

pip install requests

配置环境变量：

# Linux / macOS
export LLM_API_KEY="你的密钥"
export LLM_BASE_URL="https://你的服务地址/v1"
export LLM_MODEL="支持工具调用的模型名称"

Windows PowerShell：

$env:LLM_API_KEY="你的密钥"
$env:LLM_BASE_URL="https://你的服务地址/v1"
$env:LLM_MODEL="支持工具调用的模型名称"

然后在 agent.py 中加入 Agent 循环：

import json
import os
from typing import Any

import requests


API_KEY = os.environ["LLM_API_KEY"]
BASE_URL = os.environ["LLM_BASE_URL"].rstrip("/")
MODEL = os.environ["LLM_MODEL"]

TOOL_HANDLERS = {
    "read_file": read_file,
    "get_current_time": get_current_time,
}


def call_model(messages: list[dict[str, Any]]) -> dict[str, Any]:
    response = requests.post(
        f"{BASE_URL}/chat/completions",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
        },
        json={
            "model": MODEL,
            "messages": messages,
            "tools": TOOLS,
            "tool_choice": "auto",
            "temperature": 0,
        },
        timeout=60,
    )
    response.raise_for_status()
    return response.json()["choices"][0]["message"]


def execute_tool(name: str, arguments: str) -> str:
    handler = TOOL_HANDLERS.get(name)
    if handler is None:
        return f"错误：未知工具 {name}"

    try:
        kwargs = json.loads(arguments or "{}")
        if not isinstance(kwargs, dict):
            return "错误：工具参数必须是 JSON 对象"
        return str(handler(**kwargs))
    except json.JSONDecodeError:
        return "错误：工具参数不是合法 JSON"
    except TypeError as exc:
        return f"错误：工具参数不正确：{exc}"
    except Exception as exc:
        return f"错误：工具执行失败：{exc}"


def run_agent(user_input: str, max_steps: int = 8) -> str:
    messages: list[dict[str, Any]] = [
        {
            "role": "system",
            "content": (
                "你是一个谨慎的开发助手。根据任务选择工具；"
                "不要猜测工具结果；完成目标后直接给出结论。"
            ),
        },
        {"role": "user", "content": user_input},
    ]

    for step in range(1, max_steps + 1):
        message = call_model(messages)
        messages.append(message)
        tool_calls = message.get("tool_calls") or []

        if not tool_calls:
            return message.get("content") or "模型没有返回内容"

        for tool_call in tool_calls:
            function = tool_call["function"]
            result = execute_tool(function["name"], function.get("arguments", "{}"))
            print(f"[步骤 {step}] 调用 {function['name']} -> {result[:100]}")

            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": tool_call["id"],
                    "content": result,
                }
            )

    return f"任务超过最大执行步数 {max_steps}，已停止"


if __name__ == "__main__":
    question = input("请输入任务：")
    print("\nAgent 回答：")
    print(run_agent(question))

运行程序：

python agent.py

输入任务：

读取 README.md，总结这个项目的用途，然后告诉我上海当前时间。

一次典型的执行日志可能是：

[步骤 1] 调用 read_file -> 这是一个用于演示工具调用的最小 AI Agent 项目……
[步骤 2] 调用 get_current_time -> 2026-07-27T15:30:18+08:00

模型最后会基于真实文件内容和工具返回值组织答案，而不是凭空猜测。这也是 Agent 与普通对话接口最本质的区别。

四、这段代码为什么已经算 Agent？

判断一个程序是不是 Agent，不在于它使用了多少框架，而在于它是否形成了自主决策闭环。

在这个示例中：

• 目标：来自用户输入；
• 决策者：大模型判断是否需要工具以及调用哪个工具；
• 行动：Python 函数访问外部环境；
• 观察：工具结果以 tool 消息返回给模型；
• 循环：模型根据新信息继续决策，直到给出答案。

许多 Agent 框架所做的事情，本质上也是管理这套循环，并在此基础上加入状态持久化、任务规划、重试、并发和可观测性。

图 1：决策、编排和执行三层分离，工具层负责守住真正的权限边界。

五、真实项目中最容易踩的坑

1. Agent 陷入死循环

模型可能反复调用同一个工具。因此必须设置 max_steps，生产环境还应记录相同工具和参数的重复次数。

2. 把模型当成安全边界

系统提示词不是权限系统。文件访问范围、数据库权限、命令白名单和网络域名限制，都必须由程序强制执行。

3. 工具返回内容太多

如果直接把几十万行日志塞回上下文，不仅成本高，还会稀释真正有用的信息。应该在工具层分页、过滤或截断。

4. 允许 Agent 直接执行任意 Shell 命令

这是很多演示项目最危险的设计。删除文件、安装软件、发送消息等高风险操作，至少需要白名单、沙箱以及人工确认。

5. 只看最终答案，不看执行轨迹

Agent 的错误可能来自模型选错工具、参数错误、工具异常或上下文污染。生产系统需要保存每一步调用耗时、参数摘要、结果状态和 Token 消耗。

六、MCP 在这里扮演什么角色？

本文把工具直接写在 Python 程序里，优点是容易理解，缺点是工具和 Agent 紧密耦合。

MCP（Model Context Protocol）试图为模型连接外部工具和数据源提供统一协议。你可以把文件系统、数据库、浏览器或内部平台封装成 MCP Server，让不同的 Agent 客户端用较一致的方式发现和调用它们。

可以把两者简单理解为：

Function Calling：模型如何表达“我要调用这个工具”
MCP：客户端如何发现、连接和使用外部工具服务

图 2：Function Calling 描述调用意图，MCP 解决外部工具服务的发现与连接。

MCP 不会自动解决权限、安全和结果可信度问题。即使接入 MCP，服务端仍然需要进行参数验证和权限控制。

七、下一步可以怎样升级？

这个最小 Agent 还可以沿着四个方向继续扩展：

1. 接入 RAG：让 Agent 检索企业文档或项目知识库；
2. 增加记忆：保存跨会话的用户偏好和任务状态；
3. 接入 MCP：把本地函数改造成可复用的工具服务；
4. 加入评测：准备固定任务集，统计成功率、调用次数和成本；
5. 人工审批：执行写文件、发消息等操作前请求确认；
6. 多 Agent 协作：让规划、编码和审查角色各自负责不同阶段。

不过，多 Agent 并不一定比单 Agent 更好。角色越多，调用成本、状态同步和故障定位也越复杂。对多数业务来说，先把单 Agent 的工具、权限和评测做好，通常比急着搭建“AI 团队”更重要。

总结

一个最小可用的 AI Agent，只需要三个核心组件：

• 一个支持工具调用的大模型；
• 一组边界清晰、经过验证的工具；
• 一个不断执行“决策—行动—观察”的循环。

真正困难的部分不是让模型调用函数，而是确保它调用正确的函数、只能访问被授权的数据，并且出错时能够停止和追踪。

当我们开始讨论权限、沙箱、评测、可观测性和人工审批时，AI Agent 才真正从有趣的 Demo 走向可以交付的软件系统。

---

大语言模型MCP

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/2501_91062530/article/details/163252274