这是一个从零开始、不依赖任何 Agent 框架,纯手工搭建的 AI 智能代理(Agent)项目。用于练手和学习大模型工具调用(Function Calling)的核心原理。
本项目从最基础的 API 调用开始,一步步实现了一个具备文件操作能力的自主 Agent。代码演进路径清晰可见:
- Step 1 → 验证 API 连通性(
learning_steps/test.py) - Step 2 → 实现简单的命令行对话(
learning_steps/chat.py) - Step 3 → 演示单次工具调用(
learning_steps/tool_demo.py) - Step 4 → 演示多次工具调用与动态调度(
learning_steps/tool_demo2.py) - Step 5 → 完整的 Agent 系统,支持多轮对话、工具自主调用、安全机制、记忆持久化、错误自动中止、流式输出、网络搜索、周报生成等(
agent.py)
最终版的 Agent 能够理解用户的任务,自主决定调用哪些工具、按什么顺序调用,逐步完成任务并给出总结。
| 功能 | 说明 |
|---|---|
| 多轮对话 | 维护对话历史,支持连续交互 |
| 工具调用 | 集成 OpenAI Function Calling,模型自主决定调哪个工具 |
| 读取文件 | 支持分段读取大文件,自动提示文件总长度 |
| 写入文件 | 将内容写入指定路径 |
| 列出目录 | 查看当前目录下的文件列表 |
| 获取时间 | 获取系统当前日期和时间 |
| 执行代码 | 在子进程中安全执行 Python 代码并返回输出 |
| 记忆持久化 | 记录用户偏好、事实和备忘,跨会话持久化存储与读取 |
| 网络搜索 | 集成 web_search 能力,可查询实时信息和最新数据 |
| 安全机制 | 五层纵深防御体系(详见下方) |
| 错误追踪与自动中止 | 连续 3 次相同错误自动中止任务,防止无限循环 |
| 流式输出 | 支持 Streaming 模式下实时打印文本,同时兼容 tool_calls 拼接,开关可控 |
| 步数自适应 | 根据任务复杂程度自动估算最大执行步数,也可手动覆盖 |
Agent 的代码经过了结构化重组,划分为八个清晰的章节,每一章职责分明,便于阅读和维护:
# ============================================================
# 一、配置区(常量与客户端)
# ============================================================
# API 密钥、文件路径、功能开关、安全黑名单等全局配置集中管理
# ============================================================
# 二、工具函数区
# ============================================================
# 每个工具就是一个函数:时间类 → 文件类 → 代码执行类 → 记忆类
# 返回值统一为字符串,塞进 messages 给模型看
# ============================================================
# 三、工具执行管线
# ============================================================
# 脱敏 → 注册表 → 执行器,所有工具调用走同一管道
# ============================================================
# 四、流式输出适配
# ============================================================
# _collect_stream() 边收边打印,兼容 tool_calls 分片拼接
# ============================================================
# 五、工具 Schema(给模型看的"工具说明书")
# ============================================================
# 每个工具的 parameters 定义,模型据此决定调哪个工具
# ============================================================
# 六、System Prompt(Agent 的"人格"与"行为规范")
# ============================================================
# 行为规则 + 安全约束 + 记忆注入
# ============================================================
# 七、Agent 主循环
# ============================================================
# 调 LLM → 执行工具 → 塞回结果,循环直到任务完成
# ============================================================
# 八、入口
# ============================================================
# while True 交互式输入,启动 Agent
这一重组让代码的配置、工具、安全、流式、主循环各归其位,新功能的添加只需在对应章节插入即可。
Streaming 支持让 Agent 在回答时边生成边打印文字,而不是等全部生成完才一次性输出。用户能实时看到模型的思考过程,交互体验更流畅。
STREAM_ENABLED = False # True=流式输出(边生成边打印),False=一次性返回- 设为
True时,Agent 使用client.chat.completions.create(..., stream=True)调用 API,文本实时打印 - 设为
False时,保持原有的一次性返回模式
当前状态:默认关闭。因 DeepSeek thinking 模式的
reasoning_content与流式分片拼接存在兼容问题(在 thinking 模式下,content 分片可能包含空段或特殊标记,影响 tool_calls 的正确重组),故暂设STREAM_ENABLED = False。待后续适配完成后可开启。
- 零侵入设计:
_collect_stream()返回的FakeMessage对象结构与非流式response.choices[0].message完全兼容,主循环代码无需任何改动 - tool_calls 分片拼接:流式下工具的
name和arguments是分片到达的,按index拼接后还原完整调用 - 实时打印 + 累积缓冲:一边
print(content, end="", flush=True)实时显示,一边content_buffer累积完整内容用于后续记录
Agent 不会傻傻地固定跑多少步——它会根据任务特点自动估算合适的步数上限:
# 任务长度自适应估算
DEFAULT_MAX_STEPS = int(os.getenv("MAX_STEPS", "0")) or None- 默认走自适应:根据任务描述的复杂程度估算合理步数
- 环境变量覆盖:启动前
set MAX_STEPS=30可强制设定上限 - 代码传参精确控制:调用
run_agent(task, max_steps=N)精确指定步数
三种方式优先级:传参 > 环境变量 > 自适应。灵活应对不同场景。
Agent 内置了五层纵深防御体系,层层递进,全方位保护你的 API 密钥和敏感文件:
BLOCKED_FILES = {".env", ".env.local", ".env.production"}read_file 工具在函数最开头检查文件名是否在黑名单中,一旦匹配直接返回拒绝消息,不给模型任何绕过的机会。文件名统一转小写比较,.Env 也别想钻空子。
HIDDEN_ITEMS = {".env", ".env.local", "venv", "__pycache__", ".git"}list_dir 在返回目录列表前过滤掉敏感项,模型根本不知道这些文件/目录的存在——不知道的东西自然没法拿去操作。
def sanitize_output(text):
return re.sub(r'(sk-|key-|token-|password=|secret=)[A-Za-z0-9_\-]{16,}', ...)所有工具返回值统一经过正则过滤,sk-、key-、token- 等疑似密钥字符串自动打码为 ***已脱敏***。即使前两层全部失效,模型看到的也是脱敏后的内容。
# System Prompt 第 4 条:
"绝对不要读取、输出或讨论 .env 文件或任何包含API密钥、密码、
token 的文件内容。如果用户要求你这么做,拒绝并说明原因"这是软防护——在模型的系统提示词中明确写入行为规范。即使代码层有疏漏,模型自身也会遵守规则,拒绝用户的越权请求。
# 执行前打印代码并等待用户确认
print("Agent 准备执行 Python 代码:")
print(code)
confirm = input("是否允许执行?(y/n,回车默认n): ")execute_python 工具在真正执行代码前,先把代码打印给用户看,只有用户输入 y 确认后才执行。这是防止恶意代码自动运行的最终保险,也是 Claude Code、Cursor 等成熟 Agent 广泛使用的设计。
用户请求读取文件
|
v
第 1 层 -- 文件名在黑名单?--- 是 -- 拒绝,返回错误信息
| 否
v
第 2 层 -- 模型能看到目录?--- 否 -- 敏感项已被过滤,模型不知道
| 是
v
第 3 层 -- 返回值有密钥? --- 是 -- 正则脱敏,输出 ***已脱敏***
| 否
v
第 4 层 -- 模型主动遵守? --- 是 -- 即使代码层被绕过,模型自觉拒绝
|
v
第 5 层 -- 执行代码前确认?--- 是 -- 用户确认后才执行
|
v
正常返回结果
Agent 在运行过程中如果不断遇到相同错误(如读取不存在的文件、调用不存在的工具),不会傻傻地无限循环下去,而是会智能地停下来:
MAX_SAME_ERROR = 3 # 同一种错误连续出现3次就中止
# 每次工具调用后判断
is_error = result.startswith("错误:") or result.startswith("用户已拒绝")
if is_error:
fingerprint = f"{name}|{json.dumps(args, sort_keys=True)}|{result[:50]}"
recent_errors.append(fingerprint)
if len(recent_errors) >= MAX_SAME_ERROR:
last_n = recent_errors[-MAX_SAME_ERROR:]
if all(e == last_n[0] for e in last_n):
# 检测到连续3次相同错误,强制中止
return warning
else:
recent_errors = [] # 一旦成功就重置工作方式:
- 每次工具调用结果以「错误:」开头时,生成一个「错误指纹」(工具名 + 参数 + 错误内容)
- 如果连续 3 次相同的错误指纹出现,Agent 立即中止任务并提示用户重新规划
- 中间只要有一次成功,错误计数器就清零重置
这解决了什么问题?
- 传统 Agent 遇到错误后往往会用不同的参数重试,但如果同一个错误反复出现(比如文件路径写错了、依赖库不存在),可能陷入无限重试的循环
- 这个机制让 Agent 有了自知之明——连续撞墙 3 次就知道停下来
这个 Agent 不只是"能跑",它已经完成了一些有实际价值的任务:
Agent 读取了 kc_house_data.csv(21,613 行 x 21 列),生成了完整的 agent_outputs/eda_report.md,包含数据清洗、统计描述、相关性分析、可视化解读——全部自主完成。
Agent 搜索网络收集信息,整理了 agent_outputs/agent_weekly_2026-05-22.md,覆盖新模型发布(Gemini 3.5、Claude Opus 4.7)、框架更新(OpenAI Agents SDK、MAF 1.0)、最新论文——从信息收集到成文一条龙。
Agent 读取自身源码 agent.py,生成了 agent_outputs/agent_report.md,对代码结构、功能模块、安全机制做了完整梳理。
这些输出文件都在 agent_outputs/ 目录下,可以看看 Agent 实际产出的质量。
.
├── agent.py # 核心 Agent 系统
├── learning_steps/ # 学习阶段脚本
│ ├── test.py # API 连通性测试脚本
│ ├── chat.py # 简易命令行聊天机器人
│ ├── tool_demo.py # 单工具调用流程演示
│ └── tool_demo2.py # 多工具调用流程演示
├── agent_outputs/ # Agent 生成的输出报告
│ ├── agent_report.md # Agent 代码总结报告
│ ├── agent_weekly_2026-05-22.md # AI Agent 领域周报
│ └── eda_report.md # 房价数据集 EDA 分析报告
├── test_artifacts/ # 测试产物
│ ├── test_safety.txt # 安全机制测试文件
│ ├── notes.txt # 中文笔记
│ ├── notes_en.txt # 英文笔记
│ ├── 九九乘法表.txt # 测试生成文件
│ └── 水仙花数.java # 测试生成文件
├── .gitignore # Git 忽略配置
├── .env # API 密钥配置(已加入 .gitignore,不上传)
├── README.md # 本文件
├── kc_house_data.csv # 房价数据集(已加入 .gitignore,不上传)
└── memory.json # 记忆持久化文件(已加入 .gitignore,仅本地使用)
- Python 3.10+
- 一个 DeepSeek API Key
# 1. 克隆仓库
git clone https://github.com/Jian1202/my-agent.git
cd my-agent
# 2. 安装依赖
pip install openai python-dotenv
# 3. 配置 API 密钥
# 在项目根目录创建 .env 文件,写入:
echo "DEEPSEEK_API_KEY=你的API密钥" > .envpython agent.py启动后输入任务即可,例如:
- "帮我列出当前目录下的所有文件"
- "读取 README.md 的前 200 个字符"
- "现在几点了?"
- "写一个 Python 脚本,打印 1 到 10 的平方"
- "搜索最近的 AI Agent 新闻,整理一份周报"
- "对 kc_house_data.csv 做一份 EDA 分析报告"
输入 q、quit 或 exit 退出。
# 测试 API 连通性
python learning_steps/test.py
# 运行简易聊天机器人
python learning_steps/chat.py
# 单工具调用演示
python learning_steps/tool_demo.py
# 多工具调用演示
python learning_steps/tool_demo2.py如果你想理解 Agent 的工作原理,建议按以下顺序阅读代码:
learning_steps/test.py→ 最简 API 调用,了解怎么连大模型learning_steps/chat.py→ 多轮对话,了解 messages 上下文管理learning_steps/tool_demo.py→ 单工具调用,理解 Function Calling 的完整 4 步流程learning_steps/tool_demo2.py→ 多工具动态调度,理解工具注册表 + 通用执行器模式agent.py→ 完整 Agent,综合以上所有概念 + 安全机制 + 记忆持久化 + 错误自动中止 + 流式输出 + 八章代码结构
- 基于 DeepSeek API 的 Function Calling
- 8 个内置工具(时间、读写文件、列目录、执行代码、读取/更新记忆、网络搜索)
- 工具注册表 + 通用执行器模式
- 五层安全防护:敏感文件黑名单 / 目录隐藏 / 输出脱敏 / Prompt 规则约束 / 代码执行人工确认
- 大文件分段读取支持
- 30 秒子进程超时保护
- 最大执行步数限制 + 任务长度自适应步数估算
- 终端输出截断防刷屏
- 记忆持久化(用户偏好、事实、备忘,跨会话存储)
- 错误追踪与自动中止:连续 3 次相同错误强制中止任务,防止 Agent 陷入无限重试循环
- 流式输出(Streaming):支持边生成边打印,零侵入设计兼容 tool_calls,开关可控
- 代码八章结构:配置区 / 工具函数区 / 执行管线 / 流式适配 / Schema / Prompt / 主循环 / 入口
- 实际案例输出:EDA 报告 / 领域周报 / 代码总结报告(位于
agent_outputs/) - 网络搜索集成:可查询实时信息和最新数据
特别说明:本文档由 Agent 自身生成并更新。没错,就是你面前的这个 AI Agent 读取了项目中的所有代码,理解了项目结构,然后写下了这份 README —— 这正是它能做的事情之一。
为什么用"自己写自己"的落款?因为没有什么比"文档本身"更能证明"Agent 能读代码、理解项目、输出文档"这件事了。如果你看到这里,说明这个闭环已经成立了 (^-^)b