把整份日志直接贴给模型,通常只能得到「数据库连接池耗尽」「下游超时」这类结论,但对应哪一次请求、哪一行第一次报错,往往说不清。可行的顺序是:先用时间戳切出一个能人工核对的窗口,再按 request_id / trace_id 把散落的多行归并成一次调用,最后在时间线表上标出首次异常。切片是为了让模型看到因果顺序,归并是为了让异常在链路中有位置;这两步都在本地脚本里完成,模型只读已经收窄的内容。
当日志跨多个服务、单次请求夹杂多行堆栈时,不要一次性投喂。建议先按首次报错时间切出窗口(一般取报错前若干分钟到重试结束),再用脚本按调用标识分组并按时间排序,产出带行号的时间线表;把切片和时间线表一起交给模型,并要求它回答时引用行号。边界在于:切片起止点选错会让结论整体偏移,需要用同一脚本重跑并核对分组数量与首次异常行号一致后,再据此下判断。
在日志里按时间戳切出异常前后窗口
切片的起止时间不要凭感觉定,先在不切分的情况下找锚点。用 grep -n 把可疑行连同行号捞出来,确定「首次报错」的候选行,再以这一行为基准向前覆盖到该请求的上游入口时间、向后覆盖到重试或熔断结束,这样切出来的片段才保留完整因果。
# 第一步:找锚点,不要一次看完整个文件
grep -n -E 'ERROR|Exception|timeout|status=5[0-9][0-9]' app.log | head -n 40
# 第二步:时间戳同为 ISO8601 且宽度一致时,可直接按字符串比较取窗口
awk '$0 >= "2026-01-01T09:10:00" && $0 <= "2026-01-01T09:17:00"' app.log > slice.log
wc -l slice.log
两个容易踩的点:一是日志行首不是完整时间戳(前面有容器名前缀、或时间格式带中文)时,awk 的字符串比较会失效,需要先用正则抽取时间再比较;二是切片必须保留原始行号,否则后面让模型引用位置时对不上文件,可以在切片时用 awk '{print NR": "$0}' 补一列。切片后的行数检查是必要的一步:几百行以内一般可以直接读,如果切片仍有上万行,多半是窗口里混进了并发请求,此时应加 grep 按调用标识或服务名再过滤一次,而不是把大文件交给模型。
用脚本把多行记录按调用标识归并
日志里同一请求的记录通常被打散在多个服务、多个线程之间,还可能夹杂没有标识的堆栈续行。下面的骨架只做三件事:读文件、按标识分组、按时间排序输出,输入是上一步的切片文件,输出是每行一条 JSON 的文件(JSONL),便于后续逐条引用行号。
import re, sys, json
from collections import defaultdict
# 这两条正则按实际日志格式调整,标识字段名不同的改括号内的别名
TS = re.compile(r'(\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:[.,]\d{3})?)')
RID = re.compile(r'(?:request_id|trace_id|req_id)[=: ]+([\w-]+)')
def load(path):
groups, cur, last_ts = defaultdict(list), None, ''
with open(path, encoding='utf-8', errors='replace') as f:
for no, line in enumerate(f, 1):
m_rid, m_ts = RID.search(line), TS.search(line)
if m_rid:
cur = m_rid.group(1)
if cur is None:
continue # 尚未出现任何标识的行先丢弃
if m_ts:
last_ts = m_ts.group(1).replace(',', '.')
ts = last_ts # 无时间戳的堆栈续行沿用上一条时间
groups[cur].append({'rid': cur, 'ts': ts, 'line_no': no,
'raw': line.rstrip('\n')})
return groups
def dump(groups, out):
with open(out, 'w', encoding='utf-8') as f:
for rid, rows in groups.items():
rows.sort(key=lambda r: (r['ts'], r['line_no']))
for r in rows:
f.write(json.dumps(r, ensure_ascii=False) + '\n')
if __name__ == '__main__':
dump(load(sys.argv[1]), sys.argv[2])
运行方式为 python3 group_log.py slice.log timeline.jsonl。输出字段固定为 rid / ts / line_no / raw:rid 是归并键,ts 是归一化后的时间,line_no 回指切片文件中的行,raw 是原始行。需要结合环境确认的两处:正则里的标识字段名是否与日志一致;同一条请求在不同服务中是否复用了同一个标识,若上游用 trace_id、下游只打印 span_id,分组会被拆散,此时应先补齐标识或按 trace_id 前缀归并。
输出时间线表并标出首次异常
归并完成后,把 JSONL 压成一张人能读、模型也能引用位置的表。字段不必多,但要保证每一列都能从日志原文取到,取不到的写明来源。
| 字段 | 取值示例 | 说明 |
|---|---|---|
| 时间 ts | 09:12:33.418 | 取自日志时间戳,统一到同一时区、同一精度 |
| 调用标识 rid | req-8f3a… | 归并键,同一请求内保持一致 |
| 阶段 / 服务 | gateway → order → db | 按日志中的服务名或方法名填写 |
| 状态 status | ok / error / 500 | 没有显式状态码时按关键字判定,并在备注中写明依据 |
| 耗时 cost_ms | 812 | 日志自带则直接取;没有则用同请求相邻两条 ts 差值估算,并标注「估算」 |
| 异常标记 flag | first_error / after_error / 空 | 见下方判定规则 |
| 行号 line_no | 41 | 指向切片文件,供模型引用 |
首次异常的判定建议放在脚本里做,不要交给模型猜:先按时间排序,取第一条满足「状态码 ≥ 500」或「原文命中 error、timeout、exception、panic」的记录,标为 first_error;同一请求中排在它之后的失败记录一律标 after_error,这样能区分「根因步骤」和「被拖垮的后续步骤」。如果同一时刻有多条失败,保留行号最小的那条作为首次异常,并在表里把另一条列在下一行,方便人工复核。
把时间线表和对应切片一起交给模型时,提示词要限定回答范围:
下面是一次请求的日志切片(带行号)与时间线表。只依据给定内容回答:
1) 首次异常出现在哪一行、哪个阶段;
2) 异常发生前最后一个成功的步骤是什么;
3) 要把调用链补全,还需要补采哪几条日志。
回答中每个判断都要给出 line_no;给定内容不足以判断时直接写「无法判断」,不要补全推测。
用同一脚本重跑确认可复现
单次切片得出的首次异常点,可能只是切片边界造成的错觉,所以要用同一脚本重跑并比对输出。脚本本身是确定性的,两次运行结果应当逐字节一致:
python3 group_log.py slice.log run1.jsonl
python3 group_log.py slice.log run2.jsonl
diff run1.jsonl run2.jsonl && echo '两次输出一致'
比对时先看分组数量和每组条数,再看 first_error 的行号。更稳妥的一步是换一个稍宽的窗口(前后各多留一点时间)再跑一次:如果首次异常仍落在同一阶段、同一行附近,说明结论不依赖切片起点;如果阶段变了,首先要怀疑窗口本身截断了上游调用。
两次结果不一致时,按以下顺序排查:日志在运行期间被持续写入,导致行号和内容漂移;正则误匹配到 span_id 等子标识,把一次请求拆成了多组;同一秒内多条记录且毫秒缺失,排序实际依赖行号;日志中混用了 UTC 与本地时间,排序被时区打乱;多行堆栈被切在窗口边界之外,续行挂到了错误的标识上。逐项排除后再看首次异常的位置,比直接相信一次输出更可靠。