TOPIC | 高级 AI 智能体开发实战课件 | 从零构建「小林」

课程定位：进阶实战课。适合已掌握 Python 基础、了解 LLM API 基本调用，希望深入 Agent 工程化开发的开发者。 实战项目：一个能聊天、记得住事、会操作电脑、能自己写脚本解决问题的桌面智能体。 项目代码规模：约 2000 行 Python，5 个模块。

01 | 课程总览 | 我们要造什么

一个叫「小林」的智能体，具备以下能力：

学习目标

完成本课程后，你将掌握：

1. LLM 应用工程化：不仅会调 API，更懂如何管理上下文、处理流式响应、控制成本
2. Function Calling 深度实践：工具定义、流式参数拼接、多轮工具调用循环
3. Agent 安全设计：白名单 / 黑名单策略、运行时沙箱、静态 + 动态双层防护
4. ReAct 范式落地：如何让模型"边做边想"，而非盲目执行
5. 可观测性设计：Agent 内部过程如何暴露给人看

技术栈

语言        Python 3.13LLM         DeepSeek（OpenAI 兼容接口）SDK         openai语音        火山引擎豆包 ASR/TTS（WebSocket 二进制协议）音频        sounddevice协议        WebSocket / Function Calling / MCP 兼容设计

核心设计理念

Agent = LLM + 工具 + 循环 + 记忆 + 安全边界

02 | 项目架构设计 | 五个分层各司其职

文件结构

E:\Workbuddy\小林\├── xiaolin.py            主程序：对话循环、记忆、工具调度、ReAct├── tools_powershell.py   PowerShell 工具（安全策略 + MCP 兼容定义）├── tools_python.py       Python 脚本工具（白名单沙箱 + 运行时守卫）├── asr.py                语音识别（豆包流式 ASR）├── tts.py                语音合成（豆包 TTS，台湾腔女声）├── kevin.json            长期记忆存储├── scripts/              脚本存放目录│   ├── _guard_runner.py  运行时安全守卫（自动生成）│   └── *.py              智能体自己写的脚本├── 启动.bat              一键启动└── .workbuddy/           工作记忆与日志

分层架构

┌─────────────────────────────────────────┐│         交互层（main / 命令行）           ││  键盘输入 · 退出指令 · 开场白              │└─────────────────┬───────────────────────┘                  │┌─────────────────▼───────────────────────┐│      推理层（chat：ReAct 循环）           ││  思考 → 行动 → 观察 → 循环                ││  上下文裁剪 · 流式输出 · 轮次控制          │└─────────────────┬───────────────────────┘                  │┌─────────────────▼───────────────────────┐│      调度层（dispatch_tool）              ││  按工具名路由到具体实现                    │└─────┬───────────┬───────────┬───────────┘      │           │           │┌─────▼─────┐ ┌───▼──────┐ ┌─▼──────────┐│  记忆工具  │ │PowerShell│ │ Python 脚本 ││  2 个      │ │  1 个    │ │   3 个      │└───────────┘ └──────────┘ └────────────┘      │           │           │┌─────▼───────────▼───────────▼───────────┐│      安全层（静态检查 + 运行时守卫）       │└─────────────────────────────────────────┘

设计原则

STEP 01 | 流式对话与上下文管理

为什么要流式输出

非流式：等待模型全部生成完毕才返回，用户看不到过程，长回复感知延迟高。 流式：逐字返回，用户立刻看到内容，感知延迟大幅降低。

stream = client.chat.completions.create(    model=MODEL,    messages=messages,    stream=True,          # 关键参数)reply = ""for chunk in stream:ifnot chunk.choices:      # 心跳包保护continue    piece = chunk.choices[0].delta.contentif piece:print(piece, end="", flush=True)   # 边收边打印        reply += piece                     # 同时拼接完整回复

三个易错点：

上下文窗口管理

问题：对话越长，消息越多，最终超出模型上下文限制，且成本线性上升。

本项目方案：滑动窗口，上限 60 条。

MAX_CONTEXT_MESSAGES = 60deftrim_context():"""裁剪上下文，保证总条数不超过上限"""    limit = MAX_CONTEXT_MESSAGES - 1# 减去 system 提示whilelen(messages) - 1 > limit:        messages.pop(1)                   # 从最早的一条对话消息开始删

设计要点：

• 第 0 条 System 提示永远保留（否则人格丢失）
• 删除时从索引 1 开始（不能删 system）

进阶方案对比：

System Prompt 设计

好的 System Prompt 应包含四要素：

SYSTEM_PROMPT = """你是小林，用户的朋友兼工作搭子。          # 1. 身份定位性格与说话方式：                            # 2. 风格约束- 像一个真实的人，语气自然、轻松、有温度- 回答要短。日常聊天一两句就够，别写成小作文- 不要用"好的，我明白了"这类客套话开头结尾能力：                                      # 3. 能力边界- 平时可以陪用户聊天、吐槽、闲扯- 需要干活时直接上手给结果，别先问一堆问题关于长期记忆（重要）：                       # 4. 工具使用指导- 【什么时候存】用户提到自己的信息时调用 save_user_info- 【什么时候读】用户问"你记得我吗"时调用 load_user_info- 不要跟用户汇报"我已保存"这类机器话"""

教学要点：

关键洞察：工具定义（schema）只告诉模型"有这个工具"，System Prompt 才告诉模型"什么时候该用它"。两者缺一不可。

STEP 02 | Function Calling 长期记忆

工作原理

用户输入 → 模型判断需要工具 → 返回 tool_calls（含工具名+参数）    ↓你的代码执行工具 → 把结果作为 tool 消息回传    ↓模型基于结果生成最终回复

工具定义（JSON Schema）

TOOLS = [    {"type": "function","function": {"name": "save_user_info","description": "保存或更新关于用户的长期记忆。当用户透露了姓名、职业、""城市、喜好等以后还会用到的信息时调用。同类信息覆盖旧值。","parameters": {"type": "object","properties": {"key":   {"type": "string", "description": "记忆类别，如：所在城市"},"value": {"type": "string", "description": "记忆内容，如：杭州"},                },"required": ["key", "value"],            },        },    },]

description 写法要点：

• 说清是什么：这个工具做什么
• 说清何时用：什么场景下应该调用（这是模型决策的关键依据）
• 说清约束：同类信息覆盖旧值

流式 tool_calls 的分片拼接（重点难点）

这是本项目最容易踩坑的地方。

流式模式下，工具调用的参数是分片返回的：

chunk 1: tool_calls[0] = {index: 0, id: "call_abc", function: {name: "save_user_info", arguments: '{"key":'}}chunk 2: tool_calls[0] = {index: 0, function: {arguments: '"所在城市",'}}chunk 3: tool_calls[0] = {index: 0, function: {arguments: '"value":"杭州"}'}}

规律：

• id 和 name只在第一个分片出现
• arguments 被切成多段，需要按 index 累积拼接

正确实现：

tool_calls = {}   #Dindex -> {id, name, arguments}for chunk in stream:ifnot chunk.choices:continue    delta = chunk.choices[0].deltaif delta.tool_calls:for tc in delta.tool_calls:            idx = tc.index                      # 用 index 作为分组键if idx notin tool_calls:           # 首次见到，初始化                tool_calls[idx] = {"id": "", "name": "", "arguments": ""}if tc.id:                           # id 只在首片                tool_calls[idx]["id"] = tc.idif tc.function and tc.function.name:                tool_calls[idx]["name"] = tc.function.nameif tc.function and tc.function.arguments:                tool_calls[idx]["arguments"] += tc.function.arguments   # 拼接

消息顺序协议（必踩之坑）

OpenAI 兼容接口对消息顺序有强制要求：

✅ 正确顺序：assistant（带 tool_calls）    ↓tool（工具结果，tool_call_id 必须对应）❌ 错误顺序：userassistant（带 tool_calls）user          ← 中间插入了其他消息，接口报错tool

实现：

# 1. 先把带 tool_calls 的 assistant 消息存入上下文assistant_msg = {"role": "assistant","content": reply orNone,        # 有工具调用时 content 通常为 None"tool_calls": [        {"id": tc["id"],"type": "function","function": {"name": tc["name"], "arguments": tc["arguments"]},        }for tc in tool_calls.values()    ],}messages.append(assistant_msg)# 2. 再依次追加 tool 结果消息for tc in tool_calls.values():    args = json.loads(tc["arguments"]) if tc["arguments"] else {}   # 解析失败兜底    result = dispatch_tool(tc["name"], args)    messages.append({"role": "tool","tool_call_id": tc["id"],     # 必须与 assistant 中的 id 对应"content": result,    })

记忆的读写实现

MEMORY_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "kevin.json")def_read_memory() -> dict:"""读取全部记忆；文件不存在或损坏时返回空字典"""ifnot os.path.exists(MEMORY_FILE):return {}try:withopen(MEMORY_FILE, "r", encoding="utf-8") as f:            data = json.load(f)return data ifisinstance(data, dict) else {}except (json.JSONDecodeError, OSError):return {}          # 文件坏了也不能让程序崩def_write_memory(data: dict) -> None:"""写回记忆（UTF-8 中文直存，不用转义）"""withopen(MEMORY_FILE, "w", encoding="utf-8") as f:        json.dump(data, f, ensure_ascii=False, indent=2)

两个工程细节：

• ensure_ascii=False：中文直接存储，文件可读
• 读取时容错：文件损坏不影响主流程

查询兜底策略

需求：「查询记忆时，查不到就返回全部」。

deftool_load_user_info(query: str = "") -> str:    data = _read_memory()ifnot data:return"暂无任何长期记忆。"ifnot query:                                  # 无关键词 → 全部return"全部记忆：\n" + "\n".join(f"- {k}：{v}"for k, v in data.items())    hits = {k: v for k, v in data.items() if query in k or query in v}   # key 或 value 命中ifnot hits:                                   # 查不到 → 返回全部（按要求）return"未找到匹配项，以下是全部记忆：\n" + "\n".join(f"- {k}：{v}"for k, v in data.items())return"匹配到的记忆：\n" + "\n".join(f"- {k}：{v}"for k, v in hits.items())

设计思考：查不到时返回全部，虽然可能浪费 token，但能提高模型回答的准确率（宁可多给信息）。这是产品决策而非技术决策，需要按场景权衡。

STEP 03 | PowerShell 工具与安全策略

给 Agent 装上"手"的风险

让 LLM 执行系统命令，能力很强，但风险极高。一个幻觉就可能删掉重要文件。

核心原则：默认拒绝危险操作，而非默认允许。

三层安全策略

# 一级：危险命令黑名单（正则匹配）DANGEROUS_PATTERNS = [    (r"\bFormat-Volume\b", "格式化磁盘分区"),    (r"\bClear-Disk\b", "清除磁盘所有分区"),    (r"\bRemove-Item\b.*-Recurse.*[A-Za-z]:\\?(\s|$|\")", "递归删除整盘"),    (r"\bStop-Computer\b", "关机"),    (r"\bshutdown\b", "关机/重启"),    (r"\bRemove-Item\b.*[\\/]System32", "删除系统目录"),    (r"\bRemove-Item\b.*HKLM:", "删除注册表项"),    (r"\breg\s+delete\b", "删除注册表项"),    (r"\bvssadmin\s+delete\b", "删除卷影副本"),# ... 共 20+ 条规则]# 二级：受保护路径（删除类动词 + 关键路径组合时拒绝）PROTECTED_PATHS = [r"C:\\Windows", r"C:\\Program Files", r"C:\\Program Files \(x86\)",r"\\System32", r"\\SysWOW64",]defcheck_command_safety(command: str) -> tuple:"""返回 (是否放行, 拒绝原因)"""ifnot command ornot command.strip():returnFalse, "命令为空"# 一级拦截for pattern, desc in DANGEROUS_PATTERNS:if re.search(pattern, command, re.IGNORECASE):returnFalse, f"拒绝执行：检测到危险操作【{desc}】"# 二级拦截：删除动词 + 受保护路径    delete_verbs = r"(Remove-Item|del\b|rmdir\b|rd\b|rm\b|Clear-Content)"if re.search(delete_verbs, command, re.IGNORECASE):for path in PROTECTED_PATHS:if re.search(path, command, re.IGNORECASE):returnFalse, f"拒绝执行：涉及系统关键路径（{path}）"returnTrue, ""

三级：执行层防护

result = subprocess.run(    ["powershell", "-NoProfile", "-NonInteractive", "-Command", command],    capture_output=True,    timeout=COMMAND_TIMEOUT,      # 超时保护    encoding="gbk",               # 中文环境防乱码    errors="replace",)

黑名单 vs 白名单

本项目的选择：

• PowerShell 工具用黑名单（命令组合无限，白名单不现实）
• Python 脚本用白名单（模块可枚举，适合严格管控）

教学要点：这不是二选一，而是按场景选择。系统命令空间太大不宜白名单；代码模块数量有限，适合白名单。

MCP 兼容设计

工具定义独立存放，结构与 MCP 规范对齐：

MCP_TOOL_DEFINITION = {"name": "run_powershell","description": "在本机执行 PowerShell 或 CMD 命令...危险操作会被安全策略拦截。","inputSchema": {                          # 注意：MCP 用 inputSchema"type": "object","properties": {"command": {"type": "string", "description": "要执行的命令"},"purpose": {"type": "string", "description": "命令用途，用于安全审计"},        },"required": ["command"],    },}defget_openai_tool_schema() -> dict:"""转换为 OpenAI Function Calling 格式"""return {"type": "function","function": {"name": MCP_TOOL_DEFINITION["name"],"description": MCP_TOOL_DEFINITION["description"],"parameters": MCP_TOOL_DEFINITION["inputSchema"],   # 字段名映射        },    }

为什么这样设计：

• 一份定义，两种消费：本地 Function Calling 和未来 MCP Server 都能用
• 换协议只改转换函数，不改工具实现

STEP 04 | Python 脚本能力与白名单沙箱

让 Agent 自己写代码的威力与风险

威力：用户说"帮我分析这份数据"，Agent 可以写脚本处理，能力无上限。 风险：Agent 可能写出 shutil.rmtree('C:\\') 这样的代码。

设计选择：分离式 vs 一体式

选分离式的理由：符合"自己写脚本，自己运行"的语义，且脚本成为可积累的资产。

三个工具的职责

# 1. 写脚本：安全检查 → 语法自检 → 落盘defwrite_python_script(name: str, code: str) -> str# 2. 运行脚本：读取 → 再自检 → 守卫执行defrun_python_script(name: str, args: str = "") -> str# 3. 列脚本：帮助模型确认现状deflist_python_scripts() -> str

白名单的模块清单

ALLOWED_MODULES = {# 系统与路径"sys", "os", "os.path", "pathlib", "io", "tempfile",# 数据格式"json", "csv", "configparser", "xml", "html",# 文本处理"re", "string", "textwrap", "unicodedata", "difflib",# 数学与统计"math", "cmath", "statistics", "random", "decimal", "fractions",# 容器与算法"collections", "itertools", "functools", "operator", "heapq", "bisect",# 时间"time", "datetime", "calendar",# 编码与散列"hashlib", "hmac", "base64", "uuid",# 类型与结构"typing", "dataclasses", "enum", "abc", "copy", "pprint",# 其他"warnings", "contextlib", "glob", "fnmatch", "shlex",}   # 共 51 个

明确禁用的模块（带原因，便于给出清晰拒绝信息）：

BLOCKED_MODULES = {"subprocess":     "调用外部程序","shutil":         "文件系统操作（含递归删除）","ctypes":         "调用系统底层 API","socket":         "网络通信","winreg":         "操作注册表","pickle":         "反序列化（可执行任意代码）","importlib":      "动态导入（可绕过白名单）","builtins":       "内置函数（可通过它绕过限制）",# ... 共 31 个}

静态检查为什么不够（核心难点）

用正则检查 import 是完全不够的。

# 绕过示例：字符串拼接x = 'subpro' + 'cess'm = __import__(x)          # 正则匹配不到 "subprocess"

结论：静态检查只是第一道门槛，必须配合运行时防护。

运行时守卫（真正的安全边界）

架构：不直接执行用户脚本，而是通过守卫运行器 _guard_runner.py 中转。

defrun_python_script(name, args=""):# ...# 通过守卫运行器执行，在运行期注入受限 builtins    runner_path = os.path.join(SCRIPT_DIR, "_guard_runner.py")    cmd = [PYTHON_EXE, runner_path, path]

守卫的三个动作：

# 1. 劫持 import，仅放行白名单_real_import = builtins.__import__def_guarded_import(name, globals=None, locals=None, fromlist=(), level=0):    top = name.split(".")[0]if top notin _ALLOWED:raise SecurityError(f"运行时拦截：模块 '{name}' 不在白名单内")if top == "os":return _get_os_proxy()          # 关键：os 返回受限代理return _real_import(name, globals, locals, fromlist, level)# 2. os 代理对象，屏蔽危险属性class_OsProxy:def__getattr__(self, item):if item in _OS_BLOCKED:         # remove/unlink/rmdir/system/popen...raise SecurityError(f"运行时拦截：os.{item}() 已被禁用。")import os as _real_osreturngetattr(_real_os, item)# 3. 替换 builtins，危险函数调用时抛异常_restricted_builtins = {}for _k, _v invars(builtins).items():if _k in _BLOCKED_BUILTINS:         # eval/exec/compile/__import__/open...        _restricted_builtins[_k] = _blocked_func(_k)else:        _restricted_builtins[_k] = _v

执行时注入受限命名空间：

namespace = {"__name__": "__main__","__file__": target,"__builtins__": _restricted_builtins,   # 受限内置"os": _get_os_proxy(),                  # 受限 os 代理"sys": sys,}exec(compile(source, target, "exec"), namespace)

实战踩坑：首次实现时的严重漏洞

第一版实现的问题：

# 错误写法：只在命名空间里放了代理namespace["os"] = _make_os_proxy()

但用户脚本里写 import os 时，走的是 _guarded_import，os 在白名单里，于是返回了真实 os 模块——代理对象根本没被使用。

实测暴露：

[绕过4：os.system]  -> hack              ← 命令真的执行了！[绕过5：os.remove]  -> FileNotFoundError  ← 删除函数真的被调用了！

修复：在 _guarded_import 中拦截 os 并返回代理对象：

def_guarded_import(name, ...):    top = name.split(".")[0]if top notin _ALLOWED:raise SecurityError(...)if top == "os":                          # 关键修复return _get_os_proxy()return _real_import(name, ...)

修复后验证：

[绕过4：os.system]  -> [安全拦截] os.system() 已被禁用[绕过5：os.remove]  -> [安全拦截] os.remove() 已被禁用

教学价值：这个案例说明"看起来对"的安全设计可能完全不生效。安全机制必须用攻击视角实测验证，不能靠推理确认。

九种绕过手法实测表

已知残留风险（诚实说明）

# 理论上仍可逃逸的路径().__class__.__base__.__subclasses__()   # 可取到 type 类，遍历对象图

为什么难防：这是 Python 语言层面的固有难题。RestrictedPython 等成熟方案靠 AST 重写实现，而非运行时拦截。

彻底隔离方案：容器化 / 子进程降权（Windows Job Object + 低权限令牌）。

副作用说明：open 被禁用后脚本无法直接读写文件，但 io.StringIO、json.dumps、csv 内存操作均正常，典型数据处理场景不受影响。

STEP 05 | ReAct 循环与行动前自检

什么是 ReAct

ReAct = Reason（推理）+ Act（行动）

让模型像人一样做事：先想清楚要干嘛 → 动手 → 看结果 → 不对就调整 → 直到搞定。

对比：

实现结构

MAX_REACT_STEPS = 6# 防死循环defchat(user_input: str):    messages.append({"role": "user", "content": user_input})    trim_context()# ReAct 主循环：每轮 = 一次「行动 + 观察」for step inrange(1, MAX_REACT_STEPS + 1):# ---------- Reason + Act ----------        stream = client.chat.completions.create(            model=MODEL, messages=messages, tools=TOOLS, stream=True,        )# ... 流式接收文本与 tool_calls ...# 没有工具调用 → 推理结束，输出最终回复ifnot tool_calls:            messages.append({"role": "assistant", "content": reply})return# 有工具调用 → 记录 assistant 消息        messages.append(assistant_msg)# ---------- Act + Observe ----------print(f"\n  [第 {step} 轮行动]", end="")for tc in tool_calls.values():            args = json.loads(tc["arguments"]) if tc["arguments"] else {}print(f" → {tc['name']}", end="", flush=True)     # 可观测            result = dispatch_tool(tc["name"], args)            messages.append({"role": "tool", "tool_call_id": tc["id"], "content": result})        trim_context()# 回到循环开头，让模型基于观察结果继续推理# 达到上限：优雅收尾，不卡死    final = "这事儿我绕了几圈没弄利索，先停一下。要不你换个说法？"    messages.append({"role": "assistant", "content": final})

轮次上限的必要性

没有上限会怎样：模型可能陷入"调用 → 失败 → 重试 → 失败"的循环，消耗大量 token 且永不返回。

上限设多少：本项目设 6。经验值 —— 太小时复杂任务做不完，太大时异常情况浪费成本。可按业务复杂度调整。

行动前自检（"先检查，后操作"）

用 compile() 做纯语法校验（不执行代码），检查不过就不落盘：

defcheck_script_syntax(code: str) -> tuple:"""行动前自检：编译校验脚本语法是否正确"""try:compile(code, "<script>", "exec")     # 只编译，不执行returnTrue, ""except SyntaxError as e:returnFalse, f"第 {e.lineno} 行语法错误：{e.msg}"# 带行号，便于修复

两个环节都自检：

# 写脚本时：安全检查 → 语法自检 → 才落盘defwrite_python_script(name, code):    allowed, reason = check_script_safety(code)ifnot allowed:returnf"[已拒绝] {reason}"    ok, err = check_script_syntax(code)       # 语法不过 → 不落盘ifnot ok:returnf"[自检未通过] 脚本语法有误，未保存。{err}"# ... 写入文件 ...# 运行脚本时：再自检一次（防脚本被外部改坏）defrun_python_script(name, args=""):# ... 读取文件 ...    allowed, reason = check_script_safety(code)ifnot allowed:returnf"[已拒绝] {reason}"    ok, err = check_script_syntax(code)ifnot ok:returnf"[自检未通过] 脚本语法有误，已中止运行。{err}"

实测效果：

print('未闭合的括号'  -> 第 1 行：'(' was never closeddef f(: ...           -> 第 1 行：invalid syntaxif True ...           -> 第 1 行：expected ':'x = = 1               -> 第 1 行：invalid syntax

且验证：语法错误的脚本不会产生任何文件。

自我修正能力

ReAct 的核心价值：失败后能换个方法重试，而不是一次失败就放弃。

实测案例：让小林读 kevin.json 并统计字段数。

小林：先找一下这个文件在哪。  [第 1 轮行动] → run_powershell  [第 2 轮行动] → write_python_script (被拦截)写脚本那套被安全策略挡了（不让用 open 读文件），我换 PowerShell 直接读。  [第 3 轮行动] → run_powershell  [第 4 轮行动] → run_powershellkevin.json 是 4 个字段：姓名、所在城市、职业、饮食偏好值分别是 徐恺 / 杭州 / 产品经理 / 不吃辣

观察到的关键行为：

1. 脚本方案被拦截 → 没有原样重试，而是读懂了拦截原因
2. 主动改用 PowerShell 完成同等任务
3. 向用户解释"为什么换方法"

教学要点：ReAct 的自我修正依赖两个条件 —— ① 工具返回可理解的错误信息（而非只抛异常）；② System Prompt 明确要求"被拦截时换合规方式重试，别原样再试"。两者缺一不可。

可观测性设计

Agent 内部过程如果不可见，出问题就无从排查。本项目在三个位置打印状态：

# 1. 每轮行动的开始print(f"\n  [第 {step} 轮行动]", end="")# 2. 每次工具调用（含工具名）print(f" → {tc['name']}", end="", flush=True)# 3. 工具被拦截时特别标注is_blocked = result.startswith("[已拒绝]") or result.startswith("[自检未通过]")if is_blocked:print(" (被拦截)", end="", flush=True)

实际输出效果：

小林：  [第 1 轮行动] → write_python_script  [第 2 轮行动] → run_python_script算出来是 42925。

工程建议：生产环境可把 print 换成 logging，输出到文件，便于事后审计。本项目为教学演示用 print 更直观。

STEP 06 | 语音交互（ASR/TTS）

本章说明：本模块代码已完整实现，但因缺少火山引擎控制台的 App ID，鉴权未通过，功能未实际跑通。本章重点讲解协议实现，而非效果演示。

为什么语音接入要懂二进制协议

多数云服务提供 SDK，但 SDK 往往只支持移动端。服务端接入需要直接实现 WebSocket 二进制协议。

火山引擎的协议结构：

┌────────────┬──────────────┬─────────────┬─────────┐│  Header    │ Sequence(可选)│ PayloadSize │ Payload ││  4 字节     │   4 字节      │   4 字节     │  变长    │└────────────┴──────────────┴─────────────┴─────────┘

Header 4 字节的位布局（关键）：

Byte 0: [协议版本(4bit)][头大小(4bit)]Byte 1: [消息类型(4bit)][消息标志(4bit)]Byte 2: [序列化方式(4bit)][压缩方式(4bit)]Byte 3: 保留字节（0x00）

实现：

PROTOCOL_VERSION = 0b0001DEFAULT_HEADER_SIZE = 0b0001def_build_header(message_type, flags, serialization, compression) -> bytes:"""构造 4 字节协议头"""    header = bytearray(4)    header[0] = (PROTOCOL_VERSION << 4) | DEFAULT_HEADER_SIZE    header[1] = (message_type << 4) | flags    header[2] = (serialization << 4) | compression    header[3] = 0x00returnbytes(header)

消息类型常量

# 消息类型FULL_CLIENT_REQUEST   = 0b0001# 客户端：带参数的完整请求（首包）AUDIO_ONLY_REQUEST    = 0b0010# 客户端：纯音频数据包FULL_SERVER_RESPONSE  = 0b1001# 服务端：识别结果SERVER_ERROR_RESPONSE = 0b1111# 服务端：错误# 消息标志NO_SEQUENCE       = 0b0000# 不带 sequencePOS_SEQUENCE      = 0b0001# 带正数 sequenceNEG_SEQUENCE      = 0b0010# 最后一包（无 sequence）NEG_WITH_SEQUENCE = 0b0011# 最后一包（带负数 sequence）# 序列化与压缩JSON_SERIALIZATION = 0b0001GZIP_COMPRESSION   = 0b0001

ASR：流式音频上传流程

asyncdef_recognize_async(self) -> str:    headers = {"X-Api-Key": ASR_API_KEY,"X-Api-Resource-Id": "volc.bigasr.sauc.duration","X-Api-Connect-Id": str(uuid.uuid4()),    }asyncwith websockets.connect(ASR_URL, additional_headers=headers,                                  max_size=None, ping_interval=None) as ws:# 1. 发送首包：full client request（含音频参数）        request_body = {"user": {"uid": "xiaolin_user"},"audio": {"format": "pcm", "codec": "raw", "rate": 16000,"bits": 16, "channel": 1},"request": {"model_name": "bigmodel", "enable_itn": True,"enable_punc": True, "enable_ddc": True},        }        payload = gzip.compress(json.dumps(request_body).encode("utf-8"))        packet = (_build_header(FULL_CLIENT_REQUEST, POS_SEQUENCE,                                JSON_SERIALIZATION, GZIP_COMPRESSION)                  + (1).to_bytes(4, "big")            # sequence = 1                  + len(payload).to_bytes(4, "big")   # payload 大小                  + payload)await ws.send(packet)# 2. 并发：一边发音频，一边收结果        sender = asyncio.create_task(self._send_audio(ws))        receiver = asyncio.create_task(self._recv_result(ws))await sendertry:await asyncio.wait_for(receiver, timeout=10)except asyncio.TimeoutError:            receiver.cancel()returnself._final_text

音频参数要点：

分包建议：官方推荐双向流式每包 200ms（本项目 CHUNK_MS = 200）。

录音实现

def_audio_callback(self, indata, frames, time_info, status):"""sounddevice 录音回调（独立线程执行）"""ifself._recording:# indata 是 float32 数组，需转 int16 原始 PCM        pcm = (indata * 32767).astype(np.int16).tobytes()self._audio_queue.put(pcm)defstart_recording(self):self._recording = Trueself._stream = sd.InputStream(        samplerate=SAMPLE_RATE,        channels=CHANNELS,        dtype="float32",        blocksize=int(SAMPLE_RATE * CHUNK_MS / 1000),   # 每块 200ms        callback=self._audio_callback,    )self._stream.start()

结果解析（含错误处理）

asyncdef_recv_result(self, ws):whileTrue:        msg = await ws.recv()ifnotisinstance(msg, (bytes, bytearray)) orlen(msg) < 4:continue        header_size = msg[0] & 0x0F        message_type = msg[1] >> 4        flags = msg[1] & 0x0F        compression = msg[2] & 0x0F        offset = header_size * 4if (flags & 0x01) == 0x01:      # 有 sequence            offset += 4        payload_size = int.from_bytes(msg[offset:offset + 4], "big")        offset += 4        payload = msg[offset:offset + payload_size]if message_type == SERVER_ERROR_RESPONSE:self._error = payload.decode("utf-8", errors="ignore")breakif message_type == FULL_SERVER_RESPONSE:if compression == GZIP_COMPRESSION and payload:                payload = gzip.decompress(payload)            data = json.loads(payload.decode("utf-8"))            result = data.get("result")# result 可能是 dict 或 list，做兼容            text = result.get("text", "") ifisinstance(result, dict) else \                   (result[0].get("text", "") if result else"")if text:self._final_text = textif flags in (NEG_SEQUENCE, NEG_WITH_SEQUENCE):   # 最后一包break

TTS：台湾腔女声

音色选择（从官方音色表查得）：

# 台湾腔女声，官方描述："甜美活泼的女声，有明显的台湾口音"VOICE_TYPE = "zh_female_xiaohe_jupiter_bigtts"

请求体结构（与 ASR 不同，TTS 用 req_params）：

request_body = {"user": {"uid": "xiaolin_user"},"req_params": {"text": text,"speaker": VOICE_TYPE,"audio_params": {"format": "mp3", "sample_rate": 24000},    },}

实战踩坑：鉴权失败的排查方法

现象：所有请求返回 401。

排查步骤（这是可复用的方法论）：

第一步：打印完整响应头，找出服务端接受的鉴权字段白名单：

except websockets.exceptions.InvalidStatus as e:    resp = e.responseprint(f"状态码: {resp.status_code}")for k, v in resp.headers.items():print(f"{k}: {v}")

发现响应头里有：

access-control-allow-headers: DNT,X-Mx-ReqToken,...,X-Api-App-Key,X-Api-Access-Key,X-Api-Request-Id,X-Api-Resource-Id

关键推论：白名单里没有X-Api-Key，说明该网关走的是旧版双 Key 鉴权（App ID + Access Token），而非新版单 Key。

第二步：读取响应体，拿到官方错误文案：

print("响应体:", e.response.body.decode("utf-8", errors="ignore"))# 输出：{"error":"load grant: requested grant not found in SaaS storage"}

第三步：解读错误语义：

• Invalid X-Api-Key → 网关不认这个字段（鉴权模式错了）
• load grant: requested grant not found in SaaS storage → 认得凭据值，但查不到服务授权记录

最终结论：不是代码问题，是服务未开通 / 凭据不完整。需要用户去控制台确认并补齐 App ID。

方法论总结：云服务鉴权失败时，不要盲目改代码。正确顺序是 —— ① 看响应头判断协议期望；② 看响应体拿官方错误文案；③ 解读语义定位是"凭据错"还是"环境错"。

03 | MCP 协议适配设计

MCP 是什么

MCP（Model Context Protocol） 是标准化的工具接口协议，让不同 Agent 能复用同一套工具实现。

核心价值：工具提供方和消费方解耦。

本项目设计：一份定义，两种消费

# 1. MCP 格式定义（对齐 MCP tools/list 规范）MCP_TOOL_DEFINITIONS = [    {"name": "write_python_script","description": "...","inputSchema": {          # ← MCP 用 inputSchema"type": "object","properties": {...},"required": [...],        },    },]# 2. 转换为 OpenAI 格式defget_openai_tool_schemas() -> list:return [        {"type": "function","function": {"name": d["name"],"description": d["description"],"parameters": d["inputSchema"],   # ← 字段名映射            },        }for d in MCP_TOOL_DEFINITIONS    ]

两种格式的差异：

统一调度入口

defhandle_tool_call(name: str, args: dict) -> str:"""工具调用入口，供调度器调用"""if name == "write_python_script":return write_python_script(args.get("name", ""), args.get("code", ""))if name == "run_python_script":return run_python_script(args.get("name", ""), args.get("args", ""))if name == "list_python_scripts":return list_python_scripts()returnf"未知的 Python 工具：{name}"

主程序调度器：

defdispatch_tool(name: str, args: dict) -> str:if name == "save_user_info":return tool_save_user_info(args.get("key", ""), args.get("value", ""))if name == "load_user_info":return tool_load_user_info(args.get("query", ""))if name == "run_powershell":return tools_powershell.handle_tool_call(args)if name in ("write_python_script", "run_python_script", "list_python_scripts"):return tools_python.handle_tool_call(name, args)returnf"未知工具：{name}"

演进路径

当前架构：单机 Agent，本地 Function Calling    ↓ 改造点：把 handle_tool_call 暴露为 MCP Server 端点MCP Server：工具实现不变，加一层协议适配    ↓多 Agent 复用：任何支持 MCP 的 Agent 都能调用这套工具

04 | 工程经验与踩坑总结

流式处理的三个坑

消息顺序协议

必须遵守：带 tool_calls 的 assistant 消息，后面紧跟对应的 tool 结果消息。

违反后果：接口直接报错，且错误信息不直观，排查耗时。

安全机制必须实测

本项目最深刻的教训：

第一版白名单沙箱"看起来完全正确"，但 os.system 实际能执行。原因是 import os 走白名单放行，拿到的是真实模块，代理对象从未生效。

教训：

• 安全设计不能靠推理确认，必须用攻击视角实测
• 要主动构造绕过用例（字符串拼接、__import__ 直调、from import 等）
• "测试通过"不等于"安全"——测试用例本身要包括恶意用例

Windows 环境的两个坑

坑一：.bat 文件中的中文

# 错误：批处理本体写中文，被 CMD 按 GBK 解析成乱码'ho.' 不是内部或外部命令# 正确：批处理本体保持纯 ASCII，只加 chcp 65001 保证运行期中文输出@echo offchcp 65001 >nultitle Xiaolin"C:\path\to\python.exe" xiaolin.py

坑二：中文编码

PowerShell 输出用 encoding="gbk" 解码；Python 脚本文件统一 UTF-8；JSON 存储用 ensure_ascii=False。

容错设计的三个位置

# 1. 文件读取容错：记忆文件损坏不能让程序崩try:    data = json.load(f)except (json.JSONDecodeError, OSError):return {}# 2. 工具参数解析容错：模型偶尔给出非法 JSONtry:    args = json.loads(tc["arguments"]) if tc["arguments"] else {}except json.JSONDecodeError:    args = {}# 3. 工具执行容错：异常不能中断整个会话try:    chat(user_input)except Exception as e:print(f"[出错了] {e}")

增量开发的价值

本项目的开发节奏：一次只加一个功能，验证通过再进下一个。

1. 流式对话 + 上下文     → 验证通过2. Function Calling 记忆 → 验证通过3. PowerShell 工具       → 验证通过4. Python 脚本工具       → 验证通过5. 白名单安全升级         → 验证通过6. ReAct 循环            → 验证通过7. 语音模块              → 协议实现完成，鉴权阻塞

好处：

• 问题定位范围小（新加的代码就是嫌疑范围）
• 不会因"重构"破坏已验证的功能
• 每个功能都有独立的实测记录

05 | 动手练习 | 六个渐进式任务

练习 1：基础 —— 换一个人设（★）

任务：把"小林"改成一个"专业严谨的技术顾问"。

要求：

• 修改 System Prompt，调整语气风格
• 但保持"工具使用指导"部分不变（这是功能性的）
• 测试：让它解释一个技术概念，对比修改前后的输出差异

思考：为什么"工具使用指导"不能跟着语气一起改？

练习 2：进阶 —— 上下文按 token 裁剪（★★）

任务：把当前的"按条数裁剪"改成"按 token 数裁剪"。

要求：

• 设定 token 上限（如 4000）
• 超出时从最早的消息开始删
• 保持 System 提示不删
• 简单的 token 估算即可（如 len(text) / 1.5 估算中文 token 数）

思考：估算不精确会带来什么问题？

练习 3：进阶 —— 增加"删除记忆"工具（★★）

任务：新增 delete_user_info 工具，让用户可以删掉某条记忆。

要求：

• 参数为 key
• 在 System Prompt 中说明"何时该调用"
• 测试：先存一条记忆，再让小林删掉，最后验证 kevin.json

思考：为什么工具定义和 Prompt 指导必须配套改？

练习 4：挑战 —— 攻击自己的沙箱（★★★）

任务：尝试绕过第六章实现的 Python 白名单沙箱。

可尝试的方向：

# 方向 1：拼接模块名x = 'sub' + 'process'# 方向 2：对象图遍历().__class__.__base__.__subclasses__()# 方向 3：通过白名单模块找危险能力import osos.path.__class__.__mro__       # 能否找到有用的属性？# 方向 4：通过 glob/glob 模块的其他函数import globglob.__builtins__

要求：

• 记录每个尝试的结果（成功 / 被拦截 / 报错）
• 对成功的逃逸，分析原因
• 给出你的修补方案

思考：存在"完美的沙箱"吗？为什么？

练习 5：挑战 —— 让 ReAct 循环更聪明（★★★）

任务：当前 ReAct 循环达上限就直接放弃了，改进它。

可选改进方向（任选其一）：

1. 失败记忆：记录本次会话中失败的操作，避免重复尝试
2. 渐进式提示：接近上限时（如第 5 轮）主动提示模型"再不给出答案就要结束了"
3. 成本控制：统计 token 消耗，超过阈值时提前终止

要求：说清改进方案的取舍。

练习 6：扩展 —— 挂载真正的 MCP Server（★★★★）

任务：把现有工具暴露为标准 MCP Server。

要求：

• 复用 MCP_TOOL_DEFINITIONS 定义
• 实现 MCP 的 tools/list 和 tools/call 两个端点
• 用支持 MCP 的客户端连接测试

参考资料：MCP 官方规范文档

06 | 附录 A：完整工具清单

附录 B：关键配置项速查

# xiaolin.pyAPI_BASE = "https://api.deepseek.com"MODEL = "deepseek-flash"MAX_CONTEXT_MESSAGES = 60# 上下文条数上限MAX_REACT_STEPS = 6# ReAct 轮次上限MEMORY_FILE = ".../kevin.json"# 长期记忆路径# tools_powershell.pyCOMMAND_TIMEOUT = 30# 命令超时（秒）MAX_OUTPUT_CHARS = 4000# 输出截断长度# tools_python.pyRUN_TIMEOUT = 60# 脚本超时（秒）ALLOWED_MODULES = {...}          # 51 个白名单模块BLOCKED_MODULES = {...}          # 31 个禁用模块BLOCKED_BUILTINS = {...}         # 12 个禁用内置函数# asr.pySAMPLE_RATE = 16000# 采样率CHUNK_MS = 200# 分包时长ASR_URL = "wss://openspeech.bytedance.com/api/v3/sauc/bigmodel_async"# tts.pyVOICE_TYPE = "zh_female_xiaohe_jupiter_bigtts"# 台湾腔女声ENCODING = "mp3"

附录 C：实测验证清单

本项目每个功能都做了实测，验证记录如下：

附录 D：项目完整文件清单

E:\Workbuddy\小林\├── xiaolin.py                      436 行   主程序├── tools_powershell.py             239 行   PowerShell 工具├── tools_python.py                 754 行   Python 脚本工具├── asr.py                          333 行   语音识别├── tts.py                          235 行   语音合成├── kevin.json                             长期记忆├── 启动.bat                               一键启动├── 高级AI智能体开发实战课件.md              本课件├── scripts/│   ├── _guard_runner.py                   运行时守卫（自动生成）│   ├── char_count.py                      示例：字符统计│   ├── list_dir.py                        示例：目录列表│   ├── sum_even_1_to_100.py               示例：偶数和│   └── sum_of_squares.py                  示例：平方和└── .workbuddy/    └── memory/                            工作记忆与日志

代码总计：约 2000 行 Python

OUTRO | 结语 | 五条核心方法论

本课程的核心方法论

1. Agent = LLM + 工具 + 循环 + 记忆 + 安全边界 —— 五个要素缺一不可
2. 工具定义 + Prompt 指导必须配套 —— schema 说"有什么"，Prompt 说"何时用"
3. 安全机制必须用攻击视角实测 —— "看起来对"不等于"真的对"
4. 增量开发，逐步验证 —— 一次一个功能，问题定位范围小
5. 可观测性优先 —— Agent 内部过程必须可见，否则无法排查

一个反复出现的教训

本项目在实现 Python 沙箱时，第一版代码"逻辑完全正确"，但实测发现 os.system 仍可执行。

原因是 import os 走了白名单放行分支，返回了真实模块，精心构造的代理对象从未被使用。

这个 bug 如果只靠代码审查，几乎不可能发现。是"用攻击手法实测"暴露了它。

继续深入的方向

课件生成日期：2026-10-01