Agent 30 课程开发指南 - 第24课

Agent 30 课程开发指南从零开始构建一个生产级 AI 助手框架。本指南将带你从向 LLM 问好一步步走到一个完整的多提供者、多通道 AI 智能体具备工具调用、记忆、安全防护和 Web 界面。每节课程都建立在上一节课的基础之上。每节课都包含可运行的代码和测试。本教程的主要思路来自于Nanobot (https://github.com/HKUDS/nanobot)Learn-Claude-Code (https://github.com/shareAI-lab/learn-claude-code/)本课程设计由AI辅助下完成因为课程自身也在不停修正请参考 https://github.com/junfhu/UltrabotStepByStep如果您觉得对您有帮助请帮助点亮一颗星。课程 24智能分块 — 平台感知的消息拆分目标构建一个分块器将较长的机器人回复拆分为各平台安全的片段同时不破坏代码块和句子完整性。你将学到为什么每个聊天平台的消息长度上限各不相同两种拆分策略基于长度 和 基于段落如何在拆分过程中检测并保护 Markdown 代码围栏将分块功能接入出站通道路径新建文件ultrabot/chunking/__init__.py— 公共导出ultrabot/chunking/chunker.py—ChunkMode、chunk_text()、平台限制表步骤 1定义平台限制与分块模式每个消息平台在达到特定字符数后会截断或拒绝消息。我们维护一个查找表使分块器在消息流经 Telegram、Discord、Slack 或其他通道时能自动适配。# ultrabot/chunking/chunker.py按通道对出站消息进行分块。from__future__importannotationsfromenumimportEnumclassChunkMode(str,Enum):拆分策略。LENGTHlength# 按字符限制拆分优先在空白处断开PARAGRAPHparagraph# 按空行边界拆分# ── 平台上限字符数 ──────────────────────────────────# 每个通道驱动可以覆盖这些值但以下是合理的默认值。CHANNEL_CHUNK_LIMITS:dict[str,int]{telegram:4096,discord:2000,slack:4000,feishu:30000,qq:4500,wecom:2048,weixin:2048,webui:0,# 0 无限制Web UI 会完整流式传输响应}DEFAULT_CHUNK_LIMIT4000DEFAULT_CHUNK_MODEChunkMode.LENGTHdefget_chunk_limit(channel:str,override:int|NoneNone)-int:返回 *channel* 的分块限制。0 表示无限制。ifoverrideisnotNoneandoverride0:returnoverridereturnCHANNEL_CHUNK_LIMITS.get(channel,DEFAULT_CHUNK_LIMIT)关键设计决策0表示无限制 — Web UI 直接流式传输到浏览器因此不需要拆分。override参数允许按通道配置覆盖默认值。步骤 2主入口chunk_text()调度器检查快速退出条件空文本、在限制范围内然后委托给相应的策略。defchunk_text(text:str,limit:int,mode:ChunkModeChunkMode.LENGTH,)-list[str]:将 *text* 拆分为遵守 *limit* 的分块。 - limit 0 → 将完整文本作为一个分块返回不拆分。 - LENGTH 模式 → 优先在换行/空白处断开感知代码围栏。 - PARAGRAPH 模式 → 在空行处拆分对过大的段落回退到 LENGTH 模式。 ifnottext:return[]iflimit0:return[text]iflen(text)limit:return[text]ifmodeChunkMode.PARAGRAPH:return_chunk_by_paragraph(text,limit)return_chunk_by_length(text,limit)步骤 3基于长度的拆分与代码围栏保护棘手的部分我们绝不能在 代码块内部拆分。如果拆分点落在未闭合的围栏内我们会将分块扩展到包含闭合围栏。def_chunk_by_length(text:str,limit:int)-list[str]:按 *limit* 拆分优先在换行/空白边界处断开。 Markdown 围栏感知不会在 代码块内部拆分。 chunks:list[str][]remainingtextwhileremaining:iflen(remaining)limit:chunks.append(remaining)breakcandidateremaining[:limit]# ── 代码围栏保护 ───────────────────────────# 统计开启/关闭围栏的数量。奇数表示我们在代码块内部。fence_countcandidate.count()iffence_count%21:# 找到最后一个开启围栏之后的关闭围栏fence_endremaining.find(,candidate.rfind()3)iffence_end!-1andfence_end3len(remaining):split_atfence_end3# 对齐到关闭围栏之后的下一个换行nlremaining.find(\n,split_at)ifnl!-1andnlsplit_at10:split_atnl1chunks.append(remaining[:split_at])remainingremaining[split_at:]continue# ── 寻找最佳断开点 ───────────────────────# 优先级双换行 单换行 空格best-1forsepin[\n\n,\n, ]:poscandidate.rfind(sep)ifposlimit//4:# 不要断得太早bestposlen(sep)breakifbest0:chunks.append(remaining[:best].rstrip())remainingremaining[best:].lstrip()else:# 没有合适的断开点 — 硬拆分chunks.append(remaining[:limit])remainingremaining[limit:]return[cforcinchunksifc.strip()]步骤 4基于段落的拆分对于像 Telegram 这样能渲染 Markdown 的平台按段落边界拆分能产生最干净的视觉效果。def_chunk_by_paragraph(text:str,limit:int)-list[str]:按段落边界空行拆分。 对于过大的段落回退到基于长度的拆分。 paragraphstext.split(\n\n)chunks:list[str][]currentforparainparagraphs:parapara.strip()ifnotpara:continue# 单个段落超过限制 → 回退到基于长度的拆分iflen(para)limit:ifcurrent:chunks.append(current.rstrip())currentchunks.extend(_chunk_by_length(para,limit))continue# 尝试追加到当前分块candidatef{current}\n\n{para}ifcurrentelseparaiflen(candidate)limit:currentcandidateelse:ifcurrent:chunks.append(current.rstrip())currentparaifcurrent:chunks.append(current.rstrip())return[cforcinchunksifc.strip()]步骤 5包初始化# ultrabot/chunking/__init__.py按通道对出站消息进行分块。fromultrabot.chunking.chunkerimport(CHANNEL_CHUNK_LIMITS,DEFAULT_CHUNK_LIMIT,DEFAULT_CHUNK_MODE,ChunkMode,chunk_text,get_chunk_limit,)__all__[CHANNEL_CHUNK_LIMITS,DEFAULT_CHUNK_LIMIT,DEFAULT_CHUNK_MODE,ChunkMode,chunk_text,get_chunk_limit,]测试# tests/test_chunking.py智能分块系统的测试。importpytestfromultrabot.chunking.chunkerimport(ChunkMode,chunk_text,get_chunk_limit,CHANNEL_CHUNK_LIMITS,)classTestGetChunkLimit:deftest_known_channel(self):assertget_chunk_limit(telegram)4096assertget_chunk_limit(discord)2000deftest_unknown_channel_returns_default(self):assertget_chunk_limit(matrix)4000deftest_override_wins(self):assertget_chunk_limit(telegram,override1000)1000deftest_zero_override_uses_channel_default(self):assertget_chunk_limit(discord,override0)2000deftest_webui_unlimited(self):assertget_chunk_limit(webui)0classTestChunkText:deftest_empty_text(self):assertchunk_text(,100)[]deftest_within_limit_returns_single(self):assertchunk_text(hello,100)[hello]deftest_unlimited_returns_single(self):bigx*10_000assertchunk_text(big,0)[big]deftest_splits_at_whitespace(self):textword *100# 500 字符chunkschunk_text(text.strip(),120)assertlen(chunks)2forchunkinchunks:assertlen(chunk)140# rstrip 后有一些余量deftest_code_fence_protection(self):代码块绝不应该在中间被拆分。textBefore\npython\nx 1\n*50\nAfterchunkschunk_text(text,100)# 找到包含代码围栏开始的分块forchunkinchunks:ifpythoninchunk:# 必须同时包含闭合围栏assertinchunk[chunk.index(python)3:]breakdeftest_paragraph_mode_splits_at_blank_lines(self):textPara one.\n\nPara two.\n\nPara three.chunkschunk_text(text,20,modeChunkMode.PARAGRAPH)assertlen(chunks)2deftest_paragraph_mode_oversized_falls_back(self):textShort.\n\nx*200# 第二个段落很大chunkschunk_text(text,50,modeChunkMode.PARAGRAPH)assertlen(chunks)2assertchunks[0]Short.检查点python-mpytest tests/test_chunking.py-v预期结果所有测试通过。验证代码围栏保持完整fromultrabot.chunkingimportchunk_text textHere:\n\nline\n*500\nDone.chunkschunk_text(text,200)forcinchunks:countc.count()assertcount%20orcount0,f分块中代码围栏被破坏print(f✓{len(chunks)}个分块所有围栏完好)本课成果一个平台感知的消息拆分器支持两种策略长度和段落、代码围栏保护以及按通道的限制表。通道在发送前调用chunk_text(response, get_chunk_limit(telegram))用户将永远不会看到被破坏的代码块。本课使用的 Python 知识from __future__ import annotations这是一个特殊的导入语句让 Python 把所有类型注解当作字符串处理延迟求值使新式类型语法在较早版本的 Python 中可用。from__future__importannotationsdefchunk_text(text:str,limit:int)-list[str]:...为什么在本课中使用代码中使用了list[str]、dict[str, int]、int | None等内置泛型类型注解加上这一行确保兼容 Python 3.9。Enum枚举类型与str, Enum多重继承Enum用于定义一组命名常量。同时继承str和Enum后枚举值可以直接当字符串比较和使用。fromenumimportEnumclassChunkMode(str,Enum):LENGTHlengthPARAGRAPHparagraphprint(ChunkMode.LENGTHlength)# Trueprint(ChunkMode.LENGTH.value)# length为什么在本课中使用ChunkMode定义了两种拆分策略LENGTH和PARAGRAPH用枚举可以防止传入无效的模式字符串又因为继承了str可以方便地序列化和比较。dict[str, int]类型注解的字典Python 3.9 允许直接在内置类型上使用泛型语法dict[str, int]来标注字典的键值类型。CHANNEL_CHUNK_LIMITS:dict[str,int]{telegram:4096,discord:2000,slack:4000,}为什么在本课中使用平台限制查找表是一个从通道名字符串到字符数限制整数的映射用dict[str, int]清晰表达了数据结构。list[str]类型注解的列表与字典类似list[str]标注一个元素全为字符串的列表。defchunk_text(text:str,limit:int)-list[str]:chunks:list[str][]...returnchunks为什么在本课中使用chunk_text()返回拆分后的文本列表用list[str]清晰标注返回值类型帮助 IDE 和类型检查器提供更好的提示。字符串方法split()、strip()、count()、find()、rfind()Python 字符串提供了丰富的内置方法用于拆分、清理和搜索textHello\n\nWorld\n\nPython# split() — 按分隔符拆分paragraphstext.split(\n\n)# [Hello, World, Python]# strip() / rstrip() / lstrip() — 去除首尾空白 hello .strip()# hellohello .rstrip()# hello# count() — 统计子串出现次数codemore.count()# 3# find() / rfind() — 查找子串位置rfind 从右向左查text.find(World)# 7从左找text.rfind(World)# 7从右找为什么在本课中使用分块器需要在合适的位置断开文本——split(\n\n)按段落拆分rfind(\n)找到最后一个换行处断开count()统计代码围栏数量判断是否在代码块内部。while循环while循环在条件为真时反复执行适合不知道具体迭代次数的场景。remainingvery long text...chunks[]whileremaining:iflen(remaining)limit:chunks.append(remaining)breakchunkremaining[:limit]chunks.append(chunk)remainingremaining[limit:]为什么在本课中使用基于长度的拆分算法需要不断从剩余文本中切出符合限制的分块直到没有剩余文本——这正是while循环的典型应用。for循环与break/continuefor遍历可迭代对象。break立即退出循环continue跳过本次迭代进入下一轮。forsepin[\n\n,\n, ]:poscandidate.rfind(sep)ifposlimit//4:bestposlen(sep)break# 找到最佳断开点退出循环为什么在本课中使用寻找最佳断开点时按优先级依次尝试双换行、单换行、空格——一旦找到合适的位置就用break退出不再尝试更低优先级的分隔符。列表推导与条件过滤列表推导可以在一行内从可迭代对象生成新列表if子句可以过滤不符合条件的元素。# 过滤掉空白分块chunks[cforcinchunksifc.strip()]为什么在本课中使用拆分后可能产生空的分块全是空白字符用列表推导加条件过滤一步清理干净。字符串切片Python 的切片语法s[start:end]可以从字符串中取出子串。支持省略start从头开始或end到末尾。textHello, World!print(text[:5])# Hello — 前 5 个字符print(text[7:])# World! — 第 7 个字符到末尾print(text[-6:])# World! — 倒数 6 个字符到末尾为什么在本课中使用分块的核心操作就是切片——remaining[:limit]取出一个分块remaining[best:]保留剩余文本。函数默认参数函数定义时可以为参数设置默认值调用时如果不传该参数就使用默认值。defchunk_text(text:str,limit:int,mode:ChunkModeChunkMode.LENGTH,# 默认使用长度模式)-list[str]:...为什么在本课中使用chunk_text()的mode参数默认为LENGTH大多数调用者不需要关心拆分模式简化了接口。__all__模块导出控制__all__是一个字符串列表定义了使用from module import *时导出哪些名称。它就像模块的公开 API 清单。# ultrabot/chunking/__init__.py__all__[ChunkMode,chunk_text,get_chunk_limit,CHANNEL_CHUNK_LIMITS,]为什么在本课中使用明确声明 chunking 包的公开接口隐藏内部实现函数如_chunk_by_length、_chunk_by_paragraph让使用者只看到需要用的部分。策略模式函数调度根据条件选择不同的处理函数执行——这是策略模式的简单实现。在 Python 中用if/elif调度即可无需复杂的类继承。defchunk_text(text:str,limit:int,mode:ChunkModeChunkMode.LENGTH)-list[str]:ifmodeChunkMode.PARAGRAPH:return_chunk_by_paragraph(text,limit)# 段落策略return_chunk_by_length(text,limit)# 长度策略为什么在本课中使用分块器提供两种策略——基于长度和基于段落。chunk_text()根据mode参数调度到不同的内部函数符合开放-封闭原则添加新策略只需增加新函数和一个elif分支。f-string格式化字符串f-stringf...{expr}...可以在字符串中直接嵌入变量或表达式简洁高效。channeltelegramlimit4096print(f通道{channel}的消息限制是{limit}字符)为什么在本课中使用测试代码中用 f-string 格式化输出信息如f✓ {len(chunks)} 个分块使调试信息更清晰可读。pytest测试框架pytest是 Python 最流行的测试框架支持类组织测试、丰富的断言、参数化等功能。importpytestclassTestChunkText:deftest_empty_text(self):assertchunk_text(,100)[]deftest_within_limit_returns_single(self):assertchunk_text(hello,100)[hello]为什么在本课中使用分块逻辑有很多边界情况空文本、在限制内、代码围栏、段落拆分需要全面的测试来确保每种情况都正确处理。