如果你长期跟日志分析、批量文本清洗,或者给大模型构造输入数据这些事打交道,大概率碰过这样一个别扭场景:命令行里 grep、sed、awk 这些老伙计个个能打,可一旦数据不是“一行一条记录”,而是“一段一个含义”,它们就集体失灵。我最近打磨的 context-mode,核心就是补这个缺口——它把数据处理从“按行处理”切成“按上下文块处理”,让工具能够感知周围的段落、嵌套结构和边界条件。
我自己最早碰到这个问题,是在处理一批跨多行的应用日志时。日志里单条报错会跟着一个五六行的堆栈,而堆栈和正常输出之间只有空行分隔。用 grep 把错误行筛出来后,堆栈全丢了;用 awk 输出匹配行的前后 N 行,又会出现错误连报时堆栈被切得七零八落。那段时间我天天在跟上下文搏斗,最后实在忍不了,才动手写了 context-mode 这套模式化的处理逻辑。
这篇文章不打算讲某个特定软件怎么用,而是分享我在“context-mode”这个设计理念上的完整思路:它到底要解决什么、上下文边界怎么定义、最小可用的引擎怎么实现、真实任务里怎么用、工程落地时有哪些坑。无论你是写命令行脚本、搞日志处理,还是做数据管道拼接,这套东西都能直接抄走。
1. context-mode 解决的真正问题:工具“看不见上下文”
1.1 逐行处理工具的天然短板
绝大多数文本处理工具,默认的输入模型是“一行一个记录”。这个模型在处理结构化数据时非常好用,比如 CSV、访问日志、键值对——因为每行承载的信息是完整的,行与行之间没有太多依赖关系。
但现实世界的文本大量存在“跨行上下文”:
- 异常日志里,一行是错误描述,接下来的好几行堆栈是正文,它们是一个整体,拆开就失去意义。
- Markdown 文档里,标题下面跟着说明文字和代码块,三个东西合起来才是完整章节。
- 配置文件里,一个服务块下面有若干缩进的配置项,缩进层级本身就代表归属关系。
- 给大模型构造 prompt 时,角色说明、指令、输入内容、输出要求,往往需要拼接成一个上下文块再一起发送。
逐行工具面对这些场景有三个硬伤。
第一,它们只告诉你“命中在哪一行”,不告诉你“这个行属于哪个块”。grep 命中了 ERROR 行,但堆栈行没命中,输出结果就缺失了关键信息。
第二,awk 虽然能做到跨行状态记录,比如$0 ~ /^ERROR/ {flag=1},但处理起来逻辑完全是脚本化的,遇到嵌套结构、可变边界,代码很快就变得难以维护。
第三,sed 的范围操作只能表达“从地址 A 到地址 B”,无法表达“根据空行自适应切分”或“缩进层级变化触发块切换”这一类动态边界。
1.2 粗粒度处理带来的真实损失
拿我调日志的真实场景举例。原始数据长这样:
2024-06-01 10:00:01 INFO 启动任务 2024-06-01 10:00:03 INFO 读取配置文件 2024-06-01 10:00:05 ERROR 请求上游接口失败 com.example.AppClient: Connection refused at com.example.RemoteInvoke.run(RemoteInvoke.java:88) at com.example.TaskProcessor.execute(TaskProcessor.java:45) 2024-06-01 10:00:06 WARN 重试策略生效,等待 3 秒 2024-06-01 10:00:09 INFO 重试请求用 grep ERROR 过滤,得到的是孤零零的一句“请求上游接口失败”,没有堆栈、没有调用链,排查时等于没查。用 grep -A 5 处理,确实能带上后续 5 行,但一来堆栈行数不确定,二来在 ERROR 连续出现的场景下,第二个错误会把第一个错误的堆栈截断。我调这些日志时经常要反复人工拼接,效率极低。
其实问题本质不在 grep 或 awk 本身,而在于它们没有“块”这个抽象层。数据处理管线如果只看得到行,看不到块,那么任何需要跨行理解的操作都会流向“人工介入”。我提出 context-mode,就是想在“行”和“全文”之间补一层“块”的能力。
1.3 context-mode 是什么:借鉴系统状态的设计思路
context-mode 这个名字,其实借鉴了软件系统里“模式切换”的概念。内核有内核态和用户态,程序运行时会根据当前处理的对象切换自己的工作方式;context-mode 也干了类似的事——它让数据处理引擎在“普通行模式”和“上下文块模式”之间自动切换。
在普通行模式下,每一行独立处理;一旦识别出某个上下文块的起点,引擎就切换进入块模式,把后续行按就近归属的方式划分进块里,直到遇到块的结束边界。整个过程是流动的、流式的,而不是一次性把整个文件加载进来再切分。
这样做的好处是:处理逻辑和结构感知分离了。你要做的是定义边界规则,剩下的分块、归属、嵌套都由引擎完成。后续不管你要按块统计、按块摘要、按块删除,还是按块重新组装输出,都变得非常自然。
2. 先给“上下文块”定义清楚三种边界
要写出可靠的 context-mode 引擎,第一件事不是写代码,而是把“什么时候算一个新块开始、什么时候算当前块结束”这件事梳理清楚。我把边界归纳成三类:空白边界、标记边界、结构边界。
2.1 空白边界:最简单的块切分
第一类边界是空行。空的逻辑非常清晰:两个非空行之间出现了空行,说明这两行之间的上下文被工具中断了,应该视作两个不同的块。
拿日志举例,错误信息后面跟一个空行,再跟下一条正常日志,那么这个空行就是一个天然的边界。拿文本举例,Markdown 里段落之间就是用空行分隔的,空行天然承担了“上下文结束”的语义。
实现上,空白边界只需要判断“上一行是否为空”。上一行非空、当前行为空,这意味着一个块结束;上一行为空、当前行非空,这意味着一个新块开始。这个规则简单可靠,几乎不需要额外配置。
提示:千万别只在“当前行为空”时切断,否则块内最后的空行和块本身会纠缠不清。标准做法是“看到空行,结束当前块;看到非空新行且当前块已结束,开启新块”。这样块内容永远是净数据,不含尾随空行。
2.2 标记边界:让行产生上下文含义
第二类边界是标记行。标记行不是靠位置判断的,而是靠内容或正则应激触发的。比如日志里以2024-06-01开头的时间行,文件中以#开头的标题行,或者配置里以[section]开头的分节行。
标记边界在设计时要考虑的问题比空行多:一个标记行是“新块开始”还是“当前块结束”?
我见过的许多实现都喜欢搞双向判断——既判断起点又判断终点。那样逻辑复杂,而且容易在块边界重叠时出错。我的做法是单边判断:只判断块的起点,块的结束完全交给空白边界和结构边界完成。为什么?因为在大多数文本格式里,标题后不一定紧跟空行,但标题本身已经强有力地宣告“新内容来了”,此时即便没有空行,旧块也应当终止。
实现上,边界规则是个列表,每条规则包含一个名称、一个正则表达式(或者自定义判断函数),返回 True 表示“当前行是块的起点”。引擎在逐行处理时,只要命中任意起点规则,就立刻把当前块关闭,开启新块。
2.3 结构边界:感受缩进,读懂嵌套
第三类边界是结构边界,这是我整个 context-mode 里最值钱的部分。空行和标记能切的都是平铺结构,但大量真实文本是有层级和归属的:缩进的配置、嵌套的代码块、复合的日志结构。
结构边界以缩进为信号。具体做法是保留一个上下文栈:当前行缩进量大于上一行,说明进入嵌套层级,栈中压入一个新块;当前行缩进量小于等于上一行,说明结束一个或几个子块,从栈中弹块。
举个例子,有这么一段:
模块A: 设置项1: 值1 设置项2: 值2 子模块: 设置项3: 值3 模块B: 设置项4: 值4处理时,“模块A”开启一个块,“设置项1”进入子块,“子模块”进入更深的子块,“模块B”处缩进回到顶层,触发连续弹出两个子块。最终我们得到一个嵌套结构,而不仅仅是平的切片。
结构边界的优势在于,它可以处理“父子关系”而不是“邻居关系”。这在做序列化、配置分析、代码结构抽取时,价值直接翻倍。
2.4 边界规则如何排优先级
三类边界同时存在时,必须定义优先级。我实际用下来的顺序是:标记边界优先、结构边界次之、空白边界最后。
逻辑也很直接:标记行意味着“新块开始”的强信号,即使上一行缩进层级很深,它也能直接终结父块;结构边界负责表达层级变化,它要在标记边界之后判断,否则缩进的细微变化会被当作了不起的层级事件,导致误切;空白边界是最弱的信号,只在本行和上一行都没有触发标记和结构变化时起到兜底作用。
有三条很容易踩的坑。第一条,空行会自动把嵌套块切断,但这在一些场景下并不正确。比如代码块内的空行,应该保留块上下文而不是切块。解决方法是:结构边界可以配置为“空行不触发结束”,即块结束时才允许空行作废。第二条,标记规则不要写得太宽泛,否则每一行都会成为边界,块失去意义。第三条,结构边界必须容忍缩进写成 Tab 或空格混排,不能简单地用字符串长度判断,而是要先把 Tab 展开成固定宽度。
3. 实现一个最小可用的 context-mode 引擎
理清边界模型之后,实现反倒很直接。我下面给出一个完整可跑的 Python 引擎,供你参考改造。这里的重点不是代码本身,而是“流式 + 栈 + 规则”三层结构。
3.1 引擎的输入输出模型
引擎读入的不是整个文件,而是一个迭代器,每次拿到一行;输出也不是一个静态数组,而是“块完成事件”。每当识别出一个上下文块结束时,引擎立刻回调外部处理器。
这个设计有两点优势。对内,处理大文件时内存占用只跟当前未结束块的深度相关,不会因为文件几十 GB 而爆炸。对外,使用者可以在拿到一个完整块时立刻做处理,不需要等所有块切割完,管道的时效性更强。
3.2 边界分类器实现
先定义规则基类和内置规则:
# context_mode.py from dataclasses import dataclass, field from typing import Callable, Iterable, List, Optional, Tuple @dataclass class BoundaryRule: name: str is_start: Callable[[str, Optional[str]], bool] """判断当前行是否为块的起点,参数为(当前行, 上一行)""" def blank_boundary(current_line: str, prev_line: Optional[str]) -> bool: """空白边界:上一行非空,当前行为空,说明块正在结束; 当前行非空、上一行为空,则开启新块。""" if prev_line is None: return True # 第一行天然是新块的起点 return prev_line.strip() != "" and current_line.strip() == ""这里的空白边界逻辑我稍微调整了一下:把“空行出现”和“新非空行出现”合并成同一个起点判定,即只要这一行是空行且上一行非空,就把它当作本块的结束信号;当前行非空且上一行为空时再作为块起点。这样写起来最简单,也符合单边判断的原则。
标记边界用正则匹配:
import re def make_marker_rule(name: str, pattern: str) -> BoundaryRule: regex = re.compile(pattern) def is_start(current_line: str, prev_line: Optional[str]) -> bool: return bool(regex.match(current_line)) return BoundaryRule(name=name, is_start=is_start)结构边界需要感知缩进层级,这里我实现成一个带状态的规则:
class IndentBoundaryRule(BoundaryRule): """基于缩进的边界,自动维护一棵树。""" def __init__(self, tab_width: int = 4): super().__init__("indent", self._is_start) self.tab_width = tab_width self.markers: List[int] = [] # 栈,记录每层的缩进大小 def _expand_tabs(self, line: str) -> str: # 先把 Tab 展开成空格,保证缩进可比较 return line.expandtabs(self.tab_width) def _indent_len(self, line: str) -> int: stripped = line.lstrip(" ") return len(line) - len(stripped) def _is_start(self, current_line: str, prev_line: Optional[str]) -> bool: if prev_line is None: self.markers = [] return True cur_indent = self._indent_len(self._expand_tabs(current_line)) if self.markers: last = self.markers[-1] if cur_indent > last: self.markers.append(cur_indent) return True elif cur_indent < last: # 缩进降低:先把当前缩进从栈中弹掉,再判断是否影响父块 while self.markers and cur_indent < self.markers[-1]: self.markers.pop() return False else: self.markers = [cur_indent] return True return False这里有一个细节要说明:缩进降低时,我让它返回 False,只从栈里弹掉过深的层级。因为“缩进降低”这个动作本身更适合由栈弹出逻辑处理,而不是新开一个块。如果你也按我最初的想法,一看到缩进变化就开启新块,那配置文件的父子关系就会完全乱掉。
3.3 上下文栈与块输出
有了边界分类器,接下来就是引擎本体。引擎内部维护一个栈,栈里每个元素是一个“上下文块”。
@dataclass class ContextBlock: title: str rule: Optional[str] = None lines: List[str] = field(default_factory=list) parent: Optional["ContextBlock"] = None def render(self) -> str: return "\n".join(self.lines) def summary(self) -> str: return f"{self.title}({len(self.lines)} 行)" class ContextModeEngine: def __init__(self, rules: List[BoundaryRule]): self.rules = rules self.stack: List[ContextBlock] = [] def feed_line(self, line: str) -> Optional[str]: line = line.rstrip("\n") if not line.strip() and self.stack: # 空行触发块结束 return self._close_block() for rule in self.rules: prev_line = self.stack[-1].lines[-1] if self.stack and self.stack[-1].lines else None if rule.is_start(line, prev_line): block = ContextBlock(title=line, rule=rule.name, parent=self.stack[-1] if self.stack else None) self.stack.append(block) block.lines.append(line) return None if self.stack: self.stack[-1].lines.append(line) return None block = ContextBlock(title=line) self.stack.append(block) block.lines.append(line) return None def _close_block(self) -> str: if not self.stack: return "" closed = self.stack.pop() if self.stack: self.stack[-1].lines.append("") self.stack[-1].lines.extend(closed.lines) return closed.render() def finish(self) -> List[str]: results = [] while self.stack: item = self.stack.pop() results.append(item.render()) return results这个简化版引擎有几个取舍要说明。第一,它把空行的处理提前到规则匹配之前,默认“空行终结块”。如果你不需要这种终结,可以把feed_line里的空行判断去掉。第二,空行终结时,我没有把空行内容保留在闭块的lines里,而是把它拼接到父块的尾部,保证块与块之间仍然有可阅读的分隔。第三,这个引擎对嵌套块的处理是“合并进父块”,并不是像树那样保留完整子块结构。如果要做完整的树输出,需要增加子块列表;但多数下游任务用“渲染后的文本”就够了,不需要真正的树。
3.4 一次完整运行的效果
拿前面那段日志做输入:
logs = """ 2024-06-01 10:00:01 INFO 启动任务 2024-06-01 10:00:05 ERROR 请求上游接口失败 com.example.AppClient: Connection refused at com.example.RemoteInvoke.run(RemoteInvoke.java:88) at com.example.TaskProcessor.execute(TaskProcessor.java:45) 2024-06-01 10:00:06 WARN 重试策略生效 """.strip().splitlines() engine = ContextModeEngine([ make_marker_rule("timestamp", r"2024-\d{2}-\d{2}"), IndentBoundaryRule(tab_width=4), ]) for line in logs: engine.feed_line(line) for block in engine.finish(): print("---") print(block)运行后,错误信息和它的缩进堆栈会被归成一个块,而 WARN 那行由于时间戳匹配会开启新块。这个结果正是我在开头场景里需要的。
4. 三个真实任务:context-mode 怎么变成生产力
引擎写出来之后,我陆续把它插进了一批原本靠“先 grep、再手工拼接”的工作流里。这里选三个有代表性的任务说明它到底能带来什么。
4.1 多行日志的异常提取与压缩
第一个任务是我最初的目标场景:从大量日志里把“异常块”完整地提取出来,并按块做摘要。
传统做法是grep ERROR然后肉眼追堆栈,遇到跨行堆栈时基本崩溃。用 context-mode 后,我按“时间戳做标记边界”切块,再把包含 ERROR 的块独立导出,效果立竿见影:
python - <<'EOF' from context_mode import ContextModeEngine, make_marker_rule engine = ContextModeEngine([make_marker_rule("timestamp", r"\d{4}-\d{2}-\d{2}")]) with open("app.log", "r") as f: for line in f: engine.feed_line(line) for block in engine.finish(): if "ERROR" in block: print(block) EOF导出的块里不仅包含 ERROR 行,还包含完整堆栈,因为堆栈的缩进并没有触发新的时间戳规则,所以它会被继续挂在同一个块下。
基于块再做统计也更顺手:我可以统计每个块的行数分布,判断是不是有某条错误带有特别长的堆栈;也可以把堆栈里的类名抽出来去重,看是哪几个调用点最常引发超时。而这一切,在“看不到块”的原始工具眼里都是不可操作的。
4.2 文档按代码块拆解
第二个任务是把一堆 Markdown 文档按代码块切出来,做批量翻译和质量校验。
这里单纯按空行切分不够,因为代码块里的空行太多,一回车就断块。正确的切分方式是:以 Markdown 标题行^#作为标记边界开启新块,同时不让空行终结块——块结束发生在下一个标题出现时,或者文件结尾时。
改造起来只需要把空白边界关掉,空行不再终结,只让^#{1,6}这种标题规则终结块。我看过不少人在这个场景里硬用正则循环匹配,结果只能处理“不含代码块”的文档。一旦代码块出现,标题匹配就会命中代码里的井号。context-mode 的解法是:代码块本身也是一个标记边界,所有^三个反引号的行命中后,进入一种“原样保留”的特殊状态,直到下一个反引号行关闭。这其实就是“模式”两个字的价值所在——你可以给引擎定义很多种模式,在代码块模式里标题规则暂时失效。
这个场景我特别想强调:边界规则不是只能开启新块,也可以切换工作模式。我后续在 context-mode 里给规则加了一个属性,叫mode。普通规则属于“block 模式”,只有切块能力;“literal 模式”则用来处理不需要被后续规则干扰的原始区域。这样整个引擎的能力一下就从“分块”扩展到“解析”了。
4.3 给大模型构造带上下文的输入
第三个任务我用了不到二十行代码就解决了:从一批代码仓库文件里,按函数块切分源码,然后为每个函数生成“文件头注释 + 函数定义 + 函数体”的连续上下文,再交给大模型做代码审查。
具体实现是用大括号缩进作为结构边界规则。当看到def开头时标记边界触发;进入函数体后,每当缩进增加一层,压栈;遇到return结束前,弹栈;下一个def触发新块。每个函数块的文字经过简单拼接,正好是模型需要的上下文格式。
如果不用 context-mode,你也许能用 tree-sitter 这类解析器来做,但那是为语法分析设计的,配置起来很重;而 context-mode 用“行 + 缩进”的轻量模型就足够解决绝大多数“按结构取片段”的需求。适用的底层不追求完美解析,只求可靠切块。
4.4 关于效率的一句话体会
这三个任务让我得出一个结论:很多你以为需要“智能”才能干的事,其实只是缺少一个结构化的行分块能力。context-mode 不聪明,它不会判断“这一行是否属于上一行”,它只负责把上下文建模成规则,然后机械地执行。但恰恰是这种机械执行,让流程变得可复用、可调试、可放到构建管线里反复跑。对我来说,这种确定性比任何模糊的聪明都重要。
5. 工程化落地必须处理的四个细节
从原型走到能放心跑批处理的工具,中间有几个坑是必然要踩的。我整理成四条,免得你再走一遍。
5.1 大文件的流式边界:千万别做一次性切块
很多人在初版实现里习惯readlines()把整个文件读进去,再对整个列表做切分。文件小的时候没感觉,一旦日志一天几百 MB,内存直接被打满。context-mode 应该天然流式:一次只取一行,喂给引擎;结果通过回调或者攒够一定数量后批量输出。
我自己的做法是给引擎增加了“最大块数”参数。每积累 N 个闭合块就 flush 一次,既控制了内存,也方便在管道下实时输出结果。闭块之后,块对象立刻失去了引用,可以被回收,栈里只保留尚未闭合的祖先块。
5.2 边界误判:宁可漏切,不要错切
边界规则写宽了,会把正常内容也切成碎块;写窄了,块会过度膨胀。我的经验是:在规则匹配前先定义“敏感度”。比如“空行终结”在日志里可以开,但在代码里往往要关;“标记边界”用正则时,优先用^而不是任意位置的匹配,避免日志内容里的误触发。
我处理过一个具体问题:日志内容里有时会包含一行以时间开头的文本,但那其实是一段报文的字符串,不是新日志。后来我在标记规则里加了一个排除前缀:只有行首时间满足“且后跟至少一个空格和 LOG 级别词”才触发块边界。这种收敛让误报率几乎降到了零。
5.3 编码、行尾与缩进符号:统一成规范再解析
第三个问题是数据源不规范。同一个文件里,空行可能是\r\n,缩进可能是 Tab、四个空格、两个空格混用。你如果用“行以\n结尾”来做切分,在 CRLF 文件上会残留\r;用“字符串长度算缩进”又会把 Tab 当成一个字符,导致层级误判。
我的处理是:读文件时统一用io打开并指定newline='',手动把\r\n和\n都当作行结束;喂给引擎前先rstrip('\r');缩进判断前做expandtabs(4)。这套统一化流程放入preprocess步骤后,再没出过因为符号差异导致的块乱掉。
5.4 把规则做成配置,而不是硬编码进代码
最后一点可能是最影响长期维护的建议。context-mode 的规则不应该写死在代码里,而应该通过 JSON 或 YAML 配置。这样换一个项目,改配置文件就能复用同一套引擎,不用改代码重新部署。
我的规则配置文件长这样:
rules: - name: timestamp type: marker pattern: "^\\d{4}-\\d{2}-\\d{2}" mode: block - name: heading type: marker pattern: "^#{1,6} " mode: block - name: indentation type: indent tab_width: 4 mode: block - name: literal_code type: literal start_pattern: "^```" end_pattern: "^```"加载时,把 YAML 翻译成BoundaryRule列表,引擎完全不需要感知规则细节。团队其他人拿到配置后,也能很快看懂这个处理流程怎么定义边界。对我来说,这比把逻辑埋进 script 里要可维护得多。
注意:综上这几个细节,我不建议一上来就追求完整功能。第一版只需要实现“流式读入 + 一种边界 + 一种输出”,先跑通一个真实场景;有了真实反馈,再逐步加入标记边界、结构边界和多规则优先级。规则越多,相互之间的干扰越难预测,把核心链路跑稳比功能丰富重要得多。
context-mode 不是什么新潮的东西,它本质上只是给“行”和“全文”之间补了一个“块”的抽象层。但这层抽象一旦存在,很多原本要写一坨脚本逻辑的活,就变成了几行规则和一个循环。而我在实际项目里最常想起它的时刻,不是处理那三个任务时,而是低头面对一段结构混乱、没有规律可言的文本时——我知道,任何不依靠规则就能切分的东西,最终都只能靠人去猜;真正能用规则描述的东西,交给 context-mode 这样的工具才最踏实。