adaptive-log-parser

第5章 · Coding Agent 与代码生成 · 配套项目 chapter5/adaptive-log-parser

项目说明

自适应的日志解析系统（实验 5-7）

《深入理解 AI Agent》第 5 章「代码作为系统适配器」配套实验：一个能自我进化的 Agent 日志解析系统。系统初始只支持基础日志格式；遇到无法解析的新格式时，不是报错， 而是自动把失败样本 + 报错交给 Agent，让它生成能正确解析的代码，自动测试通过后热更新 注册进解析系统。全流程自动化，无需人工介入。

自愈闭环

  一行日志
     │
     ▼
 [解析引擎] 依次尝试已注册的解析器
     │
     ├── 有解析器认识 → 输出结构化字段 ✅
     │
     └── 全部失败（检测到新格式）❌
             │  失败样本 + 报错
             ▼
        [代码生成 Agent]  ← OpenAI(gpt-5.6-luna)
             │  生成 def parse(line)->dict|None
             ▼
        [自动测试]  数据结构断言（tester.py）
             │
             ├── 不通过 → 把失败报告反馈给 Agent 重试（最多 3 次）
             │
             └── 通过 → [热加载注册] + 持久化到 parsers/*.py
                          │
                          ▼
                   系统现在能正确解析该新格式 ✅（下次重启直接复用，不再问 Agent）

对应代码： - engine.py：解析引擎 + 解析器注册表 + 热加载（importlib）。内置 builtin_json_parser。 - agent.py：代码生成 Agent，调用 OpenAI 生成解析函数，支持带失败反馈迭代修复。 - tester.py：自动测试，对生成的 parse 函数做数据结构断言。 - demo.py：串起整条闭环并逐步打印。 - parsers/：Agent 学会的解析器持久化到这里，供下次直接复用。

演示的三种递进格式

1. 基础 JSON 行（系统原生支持）：{"timestamp": "...", "level": "INFO", "message": "..."}
2. 新格式 A —— 自定义竖线分隔：2026-07-17T10:23:01Z|INFO|agent.planner|step=3|Generated plan...
3. 新格式 B —— 嵌套括号：[2026-07-17 10:24:55] (ERROR) <tool=web_search> {latency_ms=812 status=timeout} :: ...

格式 A、B 初次解析都会失败，触发 Agent 生成解析器 → 自动测试通过 → 热更新后能正确解析。

运行

pip install -r requirements.txt
cp env.example .env      # 填入 OPENAI_API_KEY（默认模型 gpt-5.6-luna）；未配置时设 OPENROUTER_API_KEY 自动改走 OpenRouter

python demo.py                       # 完整演示（两种新格式，两次真实 Agent 调用，需 API Key）
python demo.py --offline             # 离线演示：用预置解析器跑完整机制，无需 API Key
python demo.py --quick               # 快速模式：只演示 1 种新格式，省一次 API 调用
python demo.py --log-file logs.txt   # 步骤 3 改用外部日志文件（每行一条）验证复用
python demo.py --output out.jsonl    # 把解析出的结构化结果写成 JSONL
python demo.py --help                # 查看全部参数

命令行参数：

参数	说明
--offline	用预置（canned）解析器代码代替调用 OpenAI，无需 API Key，确定性地演示整条机制（失败检测→生成→测试→热重载→持久化）。
--quick	只演示 1 种新格式（竖线分隔），跳过嵌套括号格式，省一次 Agent/API 调用。
--model MODEL	覆盖代码生成模型；默认读 MODEL 环境变量再回落 gpt-5.6-luna。--offline 下仅作展示。
--log-file PATH	外部日志文件（每行一条）。给定后步骤 3 改用学到的解析系统解析该文件，替代内置混合样本，验证解析器可复用到真实日志流。
--output PATH	把步骤 3 解析出的结构化结果以 JSONL（每行一条 JSON）写入该文件。

demo.py 默认真实调用 OpenAI，依次演示：(a) 新格式初次解析失败被检测到； (b) Agent 生成解析代码并通过自动测试；(c) 热更新后系统正确解析该新格式并打印结构化结果； 最后新建一个引擎，直接从 parsers/ 加载已学会的解析器，验证持久化复用（不再调用 Agent）。

没有 API Key 时用 --offline：离线模式换用 agent.py 里的 OfflineCodeGenAgent，它按必需字段 查表返回预写好的解析器源码（并非真让 LLM 现写），但失败检测→自动测试→热加载注册→持久化这些 运行时机制与在线模式完全一致，可完整跑通并验证闭环。

预期输出示例（真实运行片段）

以下摘自一次真实运行（python demo.py，模型 gpt-5.6-luna）：

步骤 1：遇到新格式 A —— 自定义竖线分隔格式
(a) 先让系统解析，预期【失败】：
  ❌ 解析失败：2026-07-17T10:23:01Z|INFO|agent.planner|step=3|Generated plan with 5 actions
触发自愈闭环：
  🔎 检测到无法解析的新格式，触发自愈。报错：没有任何已注册解析器能解析该行：...
  --- 第 1/3 次：Agent 生成解析代码 ---
    | import re
    | _PATTERN = re.compile(
    |     r"^\s*"
    |     r"(?P<timestamp>\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z)"
    |     r"\s*\|\s*(?P<level>[A-Za-z]+)\s*\|\s*(?P<module>[^|]+?)"
    |     r"\s*\|\s*step\s*=\s*(?P<step>\d+)\s*\|\s*(?P<message>\S(?:.*\S)?)\s*$"
    | )
    | def parse(line: str) -> dict | None:
    |     match = _PATTERN.match(line)
    |     if not match:
    |         return None
    |     fields = match.groupdict()
    |     fields["module"] = fields["module"].strip()
    |     fields["message"] = fields["message"].strip()
    |     fields["step"] = int(fields["step"])
    |     return fields
  🧪 自动测试（数据结构断言）：
    [样本1] 通过，解析出字段：['level', 'message', 'module', 'step', 'timestamp']
  ✅ 自动测试通过，已热更新注册解析器 'pipe_parser' 并持久化到 parsers/pipe_parser.py
(c) 热更新后重新解析同样的日志，预期【成功】：
  ✅ [pipe_parser] {'_parser': 'pipe_parser', 'timestamp': '2026-07-17T10:23:01Z',
     'level': 'INFO', 'module': 'agent.planner', 'step': 3, 'message': 'Generated plan with 5 actions'}
...
演示结束
新格式 A（竖线分隔）自愈结果：成功
新格式 B（嵌套括号）自愈结果：成功
持久化复用（混合格式全部解析）：成功

LLM 生成的代码每次可能略有不同（如变量名、正则写法），但只要通过自动测试即视为成功。 若用 python demo.py --offline，预置解析器让输出确定性复现上述闭环（无需 API Key）。

如何适配 / 扩展

• 换模型 / 供应商：本项目统一走 OpenAI 兼容协议，改环境变量即可，无需改代码。
• MODEL：换模型，例如 MODEL=gpt-5.6；也可在命令行用 python demo.py --model gpt-5.6 临时覆盖。
• OPENAI_BASE_URL：换成任意 OpenAI 兼容端点（如自建网关、Moonshot/火山方舟等）， 再把 OPENAI_API_KEY 换成对应服务的 key、MODEL 换成该服务的模型名即可。
• 三者的读取逻辑集中在 agent.py 的 CodeGenAgent.__init__。
• 换输入日志格式：在 demo.py 里按现有 PIPE_LOGS / BRACKET_LOGS 的写法，加一组 你自己的样本（XXX_LOGS）和必需字段列表（XXX_REQUIRED），再调一次 self_heal(engine, agent, "your_parser", XXX_LOGS, XXX_REQUIRED) 即可让系统自学。 required_keys 决定自动测试的验收标准（哪些字段必须被解析出且非空）。
• 接入真实日志流：把 engine.parse_line(line) 接到你的日志读取循环上；捕获 ParseError 即触发自愈闭环。已学会的解析器持久化在 parsers/*.py，重启后由 engine.load_persisted() 自动加载复用。

局限与说明

• 可视化验证降级：书中原方案是把生成的可视化代码放进虚拟浏览器渲染，再用 Vision LLM 检查渲染效果。本机没有 playwright/浏览器环境，因此把这一步降级为对 生成的解析函数做数据结构断言（用样本数据断言解析出的结构化字段正确）。核心闭环 （检测失败 → 生成解析代码 → 自动测试 → 热加载注册新解析器 → 持久化复用）是真实实现的。
• 安全性：Agent 生成的代码通过 importlib 直接执行，仅适用于可信实验环境；生产中应 加沙箱、AST 白名单、资源限制等隔离手段。系统提示已约束只用标准库、无副作用。
• 确定性：LLM 生成代码存在不确定性，故设置了「测试不通过→带反馈重试」的迭代修复 （最多 3 次）；仍可能失败，属正常现象，重跑即可。

源代码

agent.py

"""
agent.py —— 代码生成 Agent（自愈闭环的“大脑”）

职责：拿到无法解析的失败样本 + 报错，调用 OpenAI，生成一个能正确解析该格式的
Python 解析函数 `def parse(line: str) -> dict | None`。支持把上一轮自动测试的
失败报告作为反馈再次生成（迭代修复）。
"""

from __future__ import annotations

import os
import re
from typing import List, Optional

from openai import OpenAI

# .env 加载（可选依赖）
try:
    from dotenv import load_dotenv

    load_dotenv()
except Exception:
    pass


SYSTEM_PROMPT = """你是一个"日志解析器代码生成器"。用户会给你一批**同一种未知格式**的日志样本，
以及现有系统解析失败的报错。你的任务：编写一个 Python 函数，把这种格式的每一行解析成结构化字段。

严格要求：
1. 只输出一个 Python 代码块（```python ... ```），不要任何解释文字。
2. 代码块里必须定义一个函数：def parse(line: str) -> dict | None
   - 输入是一行日志（字符串）。
   - 如果这行符合你要解析的格式，返回一个 dict，键为字段名（英文小写下划线），值为解析出的内容。
   - 如果这行**不符合**这种格式，必须返回 None（不要抛异常，把机会让给其它解析器）。
3. 只能使用 Python 标准库（re、json、datetime 等），不要 import 第三方库。
4. 不要有任何 print、input、文件读写、网络访问等副作用。
5. 必须解析出用户指定的**所有必需字段**（required_keys），字段值不能为空。
6. 尽量健壮：用正则/分隔符解析，容忍字段顺序内的空格。
"""


def _build_user_prompt(
    samples: List[str],
    required_keys: List[str],
    error_report: str,
    feedback: Optional[str],
) -> str:
    sample_block = "\n".join(samples)
    parts = [
        "现有系统无法解析下面这种格式的日志，请生成解析函数。",
        "",
        "【失败样本（同一种新格式）】",
        sample_block,
        "",
        f"【系统报错】\n{error_report}",
        "",
        f"【必需解析出的字段 required_keys】\n{required_keys}",
    ]
    if feedback:
        parts += [
            "",
            "【上一版代码没通过自动测试，请修复后重新生成】",
            feedback,
        ]
    return "\n".join(parts)


def _extract_code(text: str) -> str:
    """从模型回复中抽取 Python 代码块；没有围栏时退回整段文本。"""
    m = re.search(r"```(?:python)?\s*(.*?)```", text, re.DOTALL)
    return (m.group(1) if m else text).strip()


OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"


def _map_to_openrouter_model(model: str) -> str:
    """把直连模型名映射为 OpenRouter 上的 id（非可映射 id 统一兜底到当前廉价旗舰）。"""
    if not model or "/" in model:
        return model or "openai/gpt-5.6-luna"
    m = model.lower()
    if m.startswith(("gpt-", "o1", "o3", "o4")):
        return "openai/" + model
    if m.startswith("claude"):
        if "haiku" in m:
            return "anthropic/claude-haiku-4.5"
        if "sonnet" in m:
            return "anthropic/claude-sonnet-4.6"
        return "anthropic/claude-opus-4.8"
    if m.startswith("gemini"):
        return "google/" + model
    return "openai/gpt-5.6-luna"


class CodeGenAgent:
    def __init__(self, model: Optional[str] = None):
        model = model or os.getenv("MODEL", "gpt-5.6-luna")
        api_key = os.getenv("OPENAI_API_KEY")
        base_url = os.getenv("OPENAI_BASE_URL")
        orkey = os.getenv("OPENROUTER_API_KEY")
        # 通用 OpenRouter 兜底：无直连 key，或默认 gpt-5.x（直连需组织实名认证）时改走 OpenRouter。
        prefer_or = bool(orkey) and (model or "").lower().startswith("gpt-5")
        if prefer_or or (not api_key and orkey):
            api_key, base_url, model = orkey, OPENROUTER_BASE_URL, _map_to_openrouter_model(model)
        if not api_key:
            raise SystemExit("未找到 OPENAI_API_KEY（或 OPENROUTER_API_KEY 兜底），请在环境变量或 .env 中设置。")
        # timeout / max_retries：让偶发的网络/SSL 抖动自动重试，不至于整轮崩溃
        client_kwargs = {"api_key": api_key, "timeout": 60.0, "max_retries": 3}
        if base_url:
            client_kwargs["base_url"] = base_url
        self.client = OpenAI(**client_kwargs)
        self.model = model

    def generate_parser_code(
        self,
        samples: List[str],
        required_keys: List[str],
        error_report: str,
        feedback: Optional[str] = None,
    ) -> str:
        """调用 LLM 生成解析器代码，返回纯 Python 源码字符串。"""
        user_prompt = _build_user_prompt(samples, required_keys, error_report, feedback)
        # 推理模型（gpt-5 / o 系列等）不接受 temperature=0。
        _reasoning = any(k in (self.model or "").lower()
                         for k in ("gpt-5", "o1", "o3", "o4", "thinking", "reasoner", "kimi-k3"))
        resp = self.client.chat.completions.create(
            model=self.model,
            temperature=1 if _reasoning else 0,
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": user_prompt},
            ],
        )
        return _extract_code(resp.choices[0].message.content or "")


# ---------------------------------------------------------------------------
# 离线（无 API）代码生成 Agent
# ---------------------------------------------------------------------------
# 与 CodeGenAgent 接口完全一致，但不调用 OpenAI，而是根据必需字段返回**预置**的
# 解析器源码。它的用途是：在没有 API Key 的环境里，仍能确定性地演示与验证整条
# 机制——失败检测 → （预置）生成代码 → 自动测试 → 热加载注册 → 持久化复用。
# 注意：这里的“生成”是查表返回预写好的代码，并非真正让 LLM 现写；只有换用
# CodeGenAgent 才是真正的代码生成。
_CANNED_PARSERS = {
    # 竖线分隔格式：时间戳|级别|模块|step=N|消息
    frozenset(["timestamp", "level", "module", "step", "message"]): '''import re


def parse(line: str) -> dict | None:
    pattern = (
        r"^(?P<timestamp>\\S+)\\|(?P<level>\\S+)\\|(?P<module>\\S+)"
        r"\\|step=(?P<step>\\d+)\\|(?P<message>.+)$"
    )
    match = re.match(pattern, line.strip())
    if match:
        return match.groupdict()
    return None
''',
    # 嵌套括号格式：[时间] (级别) <tool=名字> {k=v k=v} :: 消息
    frozenset(["timestamp", "level", "tool", "message"]): '''import re


def parse(line: str) -> dict | None:
    pattern = (
        r"\\[(?P<timestamp>.*?)\\] \\((?P<level>.*?)\\) <tool=(?P<tool>.*?)> "
        r"\\{latency_ms=(?P<latency_ms>\\d+) status=(?P<status>\\w+)\\} :: (?P<message>.*)"
    )
    match = re.match(pattern, line.strip())
    if match:
        return match.groupdict()
    return None
''',
}


class OfflineCodeGenAgent:
    """离线桩：查表返回预置解析器代码，接口与 CodeGenAgent 一致（无需 API Key）。"""

    def __init__(self, model: Optional[str] = None):
        self.model = model or "offline-canned"

    def generate_parser_code(
        self,
        samples: List[str],
        required_keys: List[str],
        error_report: str,
        feedback: Optional[str] = None,
    ) -> str:
        key = frozenset(required_keys)
        code = _CANNED_PARSERS.get(key)
        if code is not None:
            return code
        # 未预置该格式：返回一个永远返回 None 的桩，让自动测试如实失败，
        # 从而演示“测试未通过 → 放弃该格式”的分支（离线模式无法真正现写代码）。
        return (
            "def parse(line: str) -> dict | None:\n"
            "    # 离线模式未预置该格式的解析器\n"
            "    return None\n"
        )

demo.py

"""
demo.py —— 自适应日志解析系统：自愈闭环演示

演示整条自愈流程（全流程自动化）：
  初始系统只认基础 JSON 日志 →
  遇到没见过的新格式 → 解析【失败】被检测到 →
  把失败样本 + 报错交给 Agent → Agent【生成解析代码】→
  【自动测试】（数据结构断言）→ 通过后【热加载注册 + 持久化】→
  系统【正确解析】了新格式。

运行：
  python demo.py            # 完整演示（两种新格式，两次 Agent 调用，需 API Key）
  python demo.py --offline  # 离线演示：用预置解析器跑完整机制，无需 API Key
  python demo.py --quick    # 快速模式：只演示 1 种新格式，省一次 API 调用
  python demo.py --help     # 查看全部参数

命令行参数见文件底部的 build_arg_parser()。
"""

from __future__ import annotations

import argparse
import json
import os
import textwrap
from typing import List, Tuple

from engine import LogParserEngine, ParseError, builtin_json_parser
from agent import CodeGenAgent, OfflineCodeGenAgent
from tester import run_tests

HERE = os.path.dirname(os.path.abspath(__file__))
PARSERS_DIR = os.path.join(HERE, "parsers")

MAX_ATTEMPTS = 3  # Agent 生成→测试的最大迭代修复次数


# ---------------------------------------------------------------------------
# 演示用的三种递进日志格式
# ---------------------------------------------------------------------------
# 格式 1：基础 JSON 行 —— 初始系统就支持
JSON_LOGS = [
    '{"timestamp": "2026-07-17T10:22:31Z", "level": "INFO", "message": "Agent started task planning"}',
    '{"timestamp": "2026-07-17T10:22:33Z", "level": "DEBUG", "message": "Loaded 12 tools into context"}',
]

# 格式 2：自定义竖线分隔格式 —— Agent 没见过
#   时间戳|级别|模块|step=N|消息
PIPE_LOGS = [
    "2026-07-17T10:23:01Z|INFO|agent.planner|step=3|Generated plan with 5 actions",
    "2026-07-17T10:23:04Z|WARNING|agent.executor|step=4|Tool call retried once",
    "2026-07-17T10:23:07Z|ERROR|agent.executor|step=5|Tool web_search returned empty result",
]
PIPE_REQUIRED = ["timestamp", "level", "module", "step", "message"]

# 格式 3：嵌套括号格式 —— Agent 也没见过
#   [时间] (级别) <tool=名字> {k=v k=v} :: 消息
BRACKET_LOGS = [
    "[2026-07-17 10:24:55] (ERROR) <tool=web_search> {latency_ms=812 status=timeout} :: upstream request failed",
    "[2026-07-17 10:25:01] (INFO) <tool=code_run> {latency_ms=134 status=ok} :: executed snippet successfully",
    "[2026-07-17 10:25:09] (WARN) <tool=file_read> {latency_ms=45 status=partial} :: file truncated at 1MB",
]
BRACKET_REQUIRED = ["timestamp", "level", "tool", "message"]


# ---------------------------------------------------------------------------
# 小工具
# ---------------------------------------------------------------------------
def hr(title: str = "") -> None:
    print("\n" + "=" * 78)
    if title:
        print(title)
        print("=" * 78)


def try_parse_all(
    engine: LogParserEngine, logs: List[str]
) -> Tuple[bool, List[dict]]:
    """尝试解析一批日志，打印结果；返回 (是否全部成功, 成功解析出的结构化记录列表)。"""
    all_ok = True
    records: List[dict] = []
    for line in logs:
        try:
            result = engine.parse_line(line)
            records.append(result)
            print(f"  ✅ [{result['_parser']}] {result}")
        except ParseError:
            all_ok = False
            print(f"  ❌ 解析失败：{line}")
    return all_ok, records


def read_log_file(path: str) -> List[str]:
    """从外部日志文件读取日志（每行一条，忽略空行）。"""
    with open(path, "r", encoding="utf-8") as f:
        return [line.rstrip("\n") for line in f if line.strip()]


def write_output(path: str, records: List[dict]) -> None:
    """把解析出的结构化记录写成 JSONL（每行一条 JSON）。"""
    with open(path, "w", encoding="utf-8") as f:
        for rec in records:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")


# ---------------------------------------------------------------------------
# 自愈闭环：检测失败 → 生成 → 测试 → 热更新
# ---------------------------------------------------------------------------
def self_heal(
    engine: LogParserEngine,
    agent: "CodeGenAgent | OfflineCodeGenAgent",
    parser_name: str,
    samples: List[str],
    required_keys: List[str],
) -> bool:
    """针对一种新格式跑完整的自愈闭环，成功注册返回 True。"""
    # (a) 触发原因：拿一条样本让系统解析，确认确实失败
    failing_line = samples[0]
    try:
        engine.parse_line(failing_line)
        print("  （该格式已能解析，无需自愈）")
        return True
    except ParseError as exc:
        error_report = str(exc)
        print(f"  🔎 检测到无法解析的新格式，触发自愈。报错：{error_report}")

    target_path = os.path.join(PARSERS_DIR, f"{parser_name}.py")
    feedback = None

    for attempt in range(1, MAX_ATTEMPTS + 1):
        print(f"\n  --- 第 {attempt}/{MAX_ATTEMPTS} 次：Agent 生成解析代码 ---")
        code = agent.generate_parser_code(
            samples=samples,
            required_keys=required_keys,
            error_report=error_report,
            feedback=feedback,
        )
        print(textwrap.indent(code, "    | "))

        # 写入候选文件（parsers/），再热加载
        with open(target_path, "w", encoding="utf-8") as f:
            f.write(code)

        # 热加载生成的 parse 函数
        try:
            fn = LogParserEngine.load_parser_from_file(target_path)
        except Exception as exc:
            feedback = f"代码无法导入/执行：{type(exc).__name__}: {exc}"
            print(f"  ⚠️ 热加载失败：{feedback}")
            continue

        # (b) 自动测试：数据结构断言
        print("  🧪 自动测试（数据结构断言）：")
        test = run_tests(fn, samples, required_keys)
        print(textwrap.indent(test["report"], "    "))

        if test["passed"]:
            # (c) 通过 → 热更新注册进引擎，文件已持久化到 parsers/
            engine.register(parser_name, fn)
            print(f"  ✅ 自动测试通过，已热更新注册解析器 '{parser_name}' 并持久化到 parsers/{parser_name}.py")
            return True

        feedback = "自动测试未通过，失败详情如下：\n" + test["report"]
        print("  ↻ 测试未通过，把失败报告反馈给 Agent 重试。")

    # 全部尝试失败：删除无效文件
    if os.path.exists(target_path):
        os.remove(target_path)
    print(f"  ❌ {MAX_ATTEMPTS} 次尝试后仍未通过，放弃该格式。")
    return False


# ---------------------------------------------------------------------------
# 主流程
# ---------------------------------------------------------------------------
def main(args: argparse.Namespace) -> None:
    hr("自适应日志解析系统 —— 自愈闭环演示（实验 5-7）")
    print("初始系统只内置一个基础解析器：JSON 行解析器。")
    if args.quick:
        print("（--quick 快速模式：仅演示 1 种新格式，省一次 Agent/API 调用）")
    if args.offline:
        print("（--offline 离线模式：用预置解析器代替 OpenAI，无需 API Key，机制完全一致）")

    os.makedirs(PARSERS_DIR, exist_ok=True)  # 确保持久化目录存在（新克隆时可能只有 .gitkeep）

    engine = LogParserEngine()
    engine.register("builtin_json", builtin_json_parser)
    print(f"当前已注册解析器：{engine.parser_names}")

    # model=None 时回落到 MODEL 环境变量/默认 gpt-5.6-luna；离线模式不触碰 API
    agent = OfflineCodeGenAgent(args.model) if args.offline else CodeGenAgent(model=args.model)
    print(f"代码生成 Agent 使用模型：{agent.model}")

    # 步骤 0：基础 JSON 格式，系统本来就能解析
    hr("步骤 0：解析基础 JSON 日志（系统原生支持）")
    try_parse_all(engine, JSON_LOGS)

    # 步骤 1：自定义竖线分隔格式（Agent 没见过）
    hr("步骤 1：遇到新格式 A —— 自定义竖线分隔格式")
    print("原始日志样本：")
    for l in PIPE_LOGS:
        print(f"  {l}")
    print("\n(a) 先让系统解析，预期【失败】：")
    try_parse_all(engine, PIPE_LOGS)
    print("\n触发自愈闭环：")
    ok1 = self_heal(engine, agent, "pipe_parser", PIPE_LOGS, PIPE_REQUIRED)
    if ok1:
        print("\n(c) 热更新后重新解析同样的日志，预期【成功】：")
        try_parse_all(engine, PIPE_LOGS)

    # 步骤 2：嵌套括号格式（Agent 也没见过）—— 快速模式下跳过，省一次 API 调用
    ok2 = None
    if args.quick:
        hr("步骤 2：（--quick 模式已跳过新格式 B 的演示）")
    else:
        hr("步骤 2：遇到新格式 B —— 嵌套括号格式")
        print("原始日志样本：")
        for l in BRACKET_LOGS:
            print(f"  {l}")
        print("\n(a) 先让系统解析，预期【失败】：")
        try_parse_all(engine, BRACKET_LOGS)
        print("\n触发自愈闭环：")
        ok2 = self_heal(engine, agent, "bracket_parser", BRACKET_LOGS, BRACKET_REQUIRED)
        if ok2:
            print("\n(c) 热更新后重新解析同样的日志，预期【成功】：")
            try_parse_all(engine, BRACKET_LOGS)

    # 步骤 3：验证持久化复用 —— 新引擎直接加载 parsers/，无需再问 Agent
    hr("步骤 3：验证持久化复用（重启系统，直接加载已学会的解析器）")
    engine2 = LogParserEngine()
    engine2.register("builtin_json", builtin_json_parser)
    loaded = engine2.load_persisted(PARSERS_DIR)
    print(f"新引擎从 parsers/ 热加载了：{loaded}")
    if args.log_file:
        print(f"用学到的解析系统解析外部日志文件（不再调用 Agent）：{args.log_file}")
        mixed = read_log_file(args.log_file)
    else:
        print("直接解析之前的新格式（不再调用 Agent）：")
        mixed = [JSON_LOGS[0], PIPE_LOGS[0]]
        if not args.quick:
            mixed.append(BRACKET_LOGS[0])  # 快速模式没生成 bracket_parser，混合样本里也不放它
    all_ok, records = try_parse_all(engine2, mixed)

    if args.output:
        write_output(args.output, records)
        print(f"已将 {len(records)} 条结构化解析结果写入（JSONL）：{args.output}")

    hr("演示结束")
    print(f"新格式 A（竖线分隔）自愈结果：{'成功' if ok1 else '失败'}")
    if ok2 is None:
        print("新格式 B（嵌套括号）：--quick 模式已跳过")
    else:
        print(f"新格式 B（嵌套括号）自愈结果：{'成功' if ok2 else '失败'}")
    print(f"持久化复用（混合格式全部解析）：{'成功' if all_ok else '失败'}")
    print(f"已学会并持久化的解析器目录：{PARSERS_DIR}")


def build_arg_parser() -> argparse.ArgumentParser:
    """构造命令行参数解析器（提供 --help / --quick / --model）。"""
    parser = argparse.ArgumentParser(
        description="自适应日志解析系统：自愈闭环演示（检测失败 → Agent 生成解析代码 → "
        "自动测试 → 热加载注册 → 持久化复用）。默认走 OpenAI，需 OPENAI_API_KEY；"
        "加 --offline 用预置解析器演示同一套机制，无需 API Key。",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument(
        "--offline",
        action="store_true",
        help="离线模式：用预置（canned）解析器代码代替调用 OpenAI，无需 API Key，"
        "确定性地演示“失败检测→生成→测试→热重载→持久化”整条机制。",
    )
    parser.add_argument(
        "--quick",
        action="store_true",
        help="快速模式：只演示 1 种新格式（竖线分隔），跳过嵌套括号格式，省一次 Agent/API 调用。",
    )
    parser.add_argument(
        "--model",
        default=None,
        help="覆盖代码生成使用的模型；默认读取环境变量 MODEL，再回落到 gpt-5.6-luna。"
        "（--offline 下此项仅作展示，不影响预置解析器。）",
    )
    parser.add_argument(
        "--log-file",
        default=None,
        metavar="PATH",
        help="外部日志文件路径（每行一条日志）。给定后，步骤 3 改用学到的解析系统解析"
        "该文件，替代内置混合样本；用于验证学到的解析器可复用到真实日志流。",
    )
    parser.add_argument(
        "--output",
        default=None,
        metavar="PATH",
        help="把步骤 3 解析出的结构化结果以 JSONL（每行一条 JSON）写入该文件。",
    )
    return parser


if __name__ == "__main__":
    main(build_arg_parser().parse_args())

engine.py

"""
engine.py —— 自适应日志解析引擎（自愈闭环的“运行时”）

设计要点：
- 引擎维护一个**解析器注册表**（有序列表）。每来一行日志，依次尝试每个解析器，
  谁能解析（返回非空 dict）就用谁的结果；全部失败则抛出 ParseError —— 这就是
  “前端检测到无法解析的新格式”的信号，触发后续的自愈流程。
- 每个解析器就是一个纯函数 `parse(line: str) -> dict | None`：
    * 能解析 → 返回结构化字段（dict）
    * 不认识这行 → 返回 None（把机会让给别的解析器，避免“抢答”）
- 生成的解析器可以持久化成 parsers/*.py 模块，下次启动直接热加载复用，无需再问 Agent。

注意：这里对“可视化”做了降级——书中用虚拟浏览器 + Vision LLM 验证渲染效果，
本项目改为对解析函数做**数据结构断言**（见 tester.py），核心自愈闭环是真实实现的。
"""

from __future__ import annotations

import importlib.util
import json
import os
from typing import Callable, Dict, List, Optional, Tuple

# 一个解析器 = (名字, 解析函数)
ParserFn = Callable[[str], Optional[Dict]]


class ParseError(Exception):
    """所有已注册解析器都无法解析该行时抛出，携带原始样本供 Agent 分析。"""

    def __init__(self, line: str):
        self.line = line
        super().__init__(f"没有任何已注册解析器能解析该行：{line!r}")


def builtin_json_parser(line: str) -> Optional[Dict]:
    """内置的基础解析器：只认标准 JSON 行（JSON Lines）。

    形如：{"timestamp": "...", "level": "INFO", "message": "..."}
    不是 JSON，或不含基本字段，则返回 None（不是我的格式）。
    """
    line = line.strip()
    if not (line.startswith("{") and line.endswith("}")):
        return None
    try:
        obj = json.loads(line)
    except json.JSONDecodeError:
        return None
    if not isinstance(obj, dict):
        return None
    # 至少要有一个基本字段，才认为是“合法的 JSON 日志”
    if not any(k in obj for k in ("timestamp", "level", "message")):
        return None
    return obj


class LogParserEngine:
    """日志解析系统：持有一组解析器，并支持热加载注册新解析器。"""

    def __init__(self) -> None:
        self._parsers: List[Tuple[str, ParserFn]] = []

    # -- 注册 / 查询 --------------------------------------------------------
    def register(self, name: str, fn: ParserFn) -> None:
        """注册（或替换同名）解析器。新解析器优先级更高，放到列表末尾后再尝试。"""
        # 若同名已存在则先移除，实现“热更新替换”
        self._parsers = [(n, f) for (n, f) in self._parsers if n != name]
        self._parsers.append((name, fn))

    @property
    def parser_names(self) -> List[str]:
        return [n for n, _ in self._parsers]

    # -- 解析 ---------------------------------------------------------------
    def parse_line(self, line: str) -> Dict:
        """尝试用每个解析器解析一行；成功则在结果里标注 _parser。全部失败抛 ParseError。"""
        for name, fn in self._parsers:
            try:
                result = fn(line)
            except Exception:
                # 某个解析器对这行报错，不代表别的不行，继续尝试
                continue
            if result:
                return {"_parser": name, **result}
        raise ParseError(line)

    # -- 热加载：从 .py 文件加载 parse 函数 ----------------------------------
    @staticmethod
    def load_parser_from_file(path: str) -> ParserFn:
        """把一个 parsers/*.py 模块动态导入，取出其中的 parse 函数。"""
        module_name = "genparser_" + os.path.splitext(os.path.basename(path))[0]
        spec = importlib.util.spec_from_file_location(module_name, path)
        if spec is None or spec.loader is None:
            raise ImportError(f"无法加载模块：{path}")
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)  # 执行模块，定义 parse
        fn = getattr(module, "parse", None)
        if not callable(fn):
            raise ImportError(f"{path} 中未找到可调用的 parse(line) 函数")
        return fn

    def load_persisted(self, parsers_dir: str) -> List[str]:
        """启动时把 parsers/ 目录下已持久化的解析器全部热加载注册（复用历史成果）。"""
        loaded: List[str] = []
        if not os.path.isdir(parsers_dir):
            return loaded
        for fname in sorted(os.listdir(parsers_dir)):
            if not fname.endswith(".py") or fname.startswith("_"):
                continue
            path = os.path.join(parsers_dir, fname)
            fn = self.load_parser_from_file(path)
            name = os.path.splitext(fname)[0]
            self.register(name, fn)
            loaded.append(name)
        return loaded

tester.py

"""
tester.py —— 自动测试（对生成的解析器做数据结构断言）

书中原方案：把生成的可视化代码放进虚拟浏览器渲染，再用 Vision LLM 检查图像。
本机没有 playwright/浏览器，因此**降级**为对解析函数做单元测试：
用一批样本日志喂给生成的 parse 函数，断言它能解析出预期的结构化字段。
这保证了“生成的代码确实能正确解析新格式”，是自愈闭环里真正的质量闸门。
"""

from __future__ import annotations

from typing import Callable, Dict, List, Optional

ParserFn = Callable[[str], Optional[Dict]]


def run_tests(
    parse_fn: ParserFn,
    samples: List[str],
    required_keys: List[str],
) -> Dict:
    """对 parse_fn 跑一组断言，返回 {passed: bool, report: str, results: [...]}。

    通过条件（对每一条样本都要满足）：
      1. parse_fn(line) 不抛异常；
      2. 返回值是非空 dict；
      3. required_keys 中的每个字段都存在，且值不为空（非 None、非空字符串）。
    """
    lines: List[str] = []
    results: List[Optional[Dict]] = []
    all_passed = True

    for i, sample in enumerate(samples, 1):
        try:
            out = parse_fn(sample)
        except Exception as exc:  # 生成的代码在样本上直接崩了
            all_passed = False
            results.append(None)
            lines.append(f"[样本{i}] 解析抛出异常：{type(exc).__name__}: {exc}")
            continue

        if not isinstance(out, dict) or not out:
            all_passed = False
            results.append(out)
            lines.append(f"[样本{i}] 未返回非空 dict，实际返回：{out!r}")
            continue

        missing = [k for k in required_keys if k not in out or out[k] in (None, "")]
        if missing:
            all_passed = False
            lines.append(
                f"[样本{i}] 缺少/为空的必需字段：{missing}；实际解析出：{out}"
            )
        else:
            lines.append(f"[样本{i}] 通过，解析出字段：{sorted(out.keys())}")
        results.append(out)

    report = "\n".join(lines)
    return {"passed": all_passed, "report": report, "results": results}