← 代码课堂

第 15 课:结构化输出实战 —— AI 总结 + 信源提取 + 关联图谱

难度:★★★(需要第 04、09、10 课基础)
教材:查资料/server.py(1580 行,纯标准库)的新增功能
预计时间:讲解 70 分钟 + 练习 50 分钟
学完你能:把 AI 的自由文本"驯服"成结构化的数据,并把它存起来、关联起来、展示出来


课前须知:AI 说的话,程序怎么用?

大模型输出的是一段自然语言,比如:

好的,我来帮你整理。首先……根据维基百科(https://zh.wikipedia.org/wiki/...),……

这段文字人看着没问题,但程序没法直接用:标题在哪?标签是什么?可信来源是哪几条?
所以要把"自由文本"变成"结构化数据"(有固定字段的 dict/JSON)。
这一课就是这条流水线的完整实现:

多轮对话文本
   ↓  AI 总结(约束它输出指定字段的 JSON)
结构化卡片 { title, summary, tags, body }
   ↓  正则提取信源 + 白名单校验
sources: [{url, source, verified}]
   ↓  存进 SQLite,并维护卡片之间的关系
可检索、可关联、可展示

第一步:体验这条流水线

cd 查资料
启动服务.bat

用手机(或浏览器模拟手机)打开 /mobile:

  1. 多轮对话:连着问两三个相关问题,它会记住上下文(history)
  2. AI 总结成卡片:点归档,选择"AI 总结成卡片"——AI 把整段对话压成一张卡片
  3. 看信源:卡片里的链接带 ✓(白名单可信)或 ⚠(未验证)徽标
  4. 看关联:列表页卡片上有 🔗 N 徽标,详情页有"关联卡片"按钮可跳转

第二步:模块地图

server.py(新增/相关部分)
├── 结构化输出
│   ├── _loads_json_loose()   第 715-722 行  容错解析模型输出的 JSON
│   ├── ai_summarize()        第 725-765 行  调模型 + 字段规范化
│   └── local_summarize()     第 768-777 行  离线兜底
├── 信源
│   ├── extract_sources()     第 594-603 行  正则抓链接 + 去重限量
│   └── validate_sources()    第 577-591 行  域名白名单 + verified 标记
├── 关联
│   ├── links_of()            第 538-544 行  单张卡片的关联
│   ├── attach_links()        第 547-560 行  批量附加关联数(避免 N+1)
│   └── card_full()           第 563-574 行  详情 + 关联卡片标题
└── 接口
    ├── _summarize()          第 1413-1429 行 /api/summarize
    └── _chat()               第 1431-1475 行 流式对话(meta/delta/sources/done)

第三步:逐块精讲

块 1:ai_summarize() —— 用 prompt 约定字段(第 725-765 行)

payload = {
    "model": AI_MODEL,
    "stream": False,
    "temperature": 0.3,          # 低随机性:摘要要稳,不要发挥
    "messages": [
        {
            "role": "system",
            "content": (
                "你是知识整理助手。请把用户提供的多轮问答对话提炼成一张知识卡片,"
                "只输出一个 JSON 对象,不要输出任何解释或代码块标记。字段:"
                "title(简洁标题,20 字内)、summary(100 字内核心概述)、"
                "tags(不超过 3 个分类标签的字符串数组)、"
                "body(整理后的知识正文,Markdown 格式,可用小标题和列表)。"
            ),
        },
        {"role": "user", "content": "请整理以下对话:\n\n" + transcript},
    ],
}

和第 10 课讲的"结构化输出"是同一招,但这里更规范:

块 2:_loads_json_loose() —— 永远别相信模型会乖乖输出(第 715-722 行)

def _loads_json_loose(content):
    """从模型输出中稳健地提取 JSON 对象(容忍代码块围栏/前后缀文字)。"""
    text = (content or "").strip()
    text = re.sub(r"^```[a-zA-Z]*\s*|```\s*$", "", text).strip()   # 去掉 ```json ... ```
    start, end = text.find("{"), text.rfind("}")                   # 抠出最外层花括号
    if start != -1 and end != -1 and end > start:
        text = text[start : end + 1]
    return json.loads(text)

模型实际输出可能是:

````
好的,这是整理结果:

{"title": "...", ...}

希望有帮助!
````

直接 json.loads 会失败。这个函数做了三层清洗:去代码围栏 → 取首尾花括号 → 再解析。

核心心态:把模型当"能力很强但不太守规矩的实习生"——
prompt 是"交办要求",容错解析是"检查作业"。

块 3:字段规范化 —— 拿到数据后还要"消毒"(第 754-764 行)

data = _loads_json_loose(obj["choices"][0]["message"]["content"])
tags = data.get("tags") or []
if isinstance(tags, str):                       # 模型可能返回字符串而不是数组
    tags = re.split(r"[,,、\s]+", tags)
tags = [str(t).strip() for t in tags if str(t).strip()][:3]   # 去空、限 3 个
return {
    "title": str(data.get("title") or "").strip()[:30],       # 截断兜底
    "summary": str(data.get("summary") or "").strip()[:120],
    "tags": tags,
    "body": str(data.get("body") or "").strip() or transcript,  # 正文空就用原文
    "mode": "online",
}

三个必学动作:

  1. 类型宽容:tags 说好是数组,模型可能给你 "天文, 物理" ——字符串就按分隔符切开
  2. 长度兜底:即使 prompt 说了 20 字,也可能超;[:30] 硬截断保底
  3. 空值兜底:body 为空时用原始对话 transcript,保证卡片不是空的

验收标准:不管模型怎么发挥,输出的结构一定合法。

块 4:降级链 —— 永远有结果(第 768-777、1420-1428 行)

def local_summarize(question, transcript):
    """离线兜底:用本地规则生成卡片字段。"""
    return {
        "title": (question[:30] if question else make_title(question, transcript)),
        "summary": make_summary(transcript),
        "tags": suggest_tags(question, transcript),
        "body": transcript,
        "mode": "offline-demo",
    }
if AI_API_KEY:
    try:
        result = ai_summarize(transcript)
    except Exception as e:
        result = local_summarize(question, transcript)   # AI 失败 → 本地规则
        result["warning"] = str(e)                        # 但告诉用户发生了什么
else:
    result = local_summarize(question, transcript)
result["sources"] = extract_sources(transcript)

三级降级:AI 成功 → AI 失败用本地规则 → 没有 Key 就直接本地。
而且失败时附一个 warning 字段,不隐瞒、不崩溃。
这和第 10、14 课的容错思路完全一致——这是这个项目最稳定的设计风格。

块 5:extract_sources() —— 从文本里"捞"信源(第 594-603 行)

def extract_sources(text):
    found = []
    # 1) Markdown 链接 [标题](https://...)
    for m in re.finditer(r"\[([^\]]+)\]\((https?://[^)\s]+)\)", text or ""):
        found.append({"title": m.group(1).strip(), "url": m.group(2).strip()})
    # 2) 裸链接 https://...(排除中文标点结尾)
    for m in re.finditer(r"https?://[^\s)\]<>\"',。;]+", text or ""):
        url = m.group(0).rstrip(".,;:)]})】")
        if not any(f["url"] == url for f in found):
            found.append({"title": "", "url": url})
    return validate_sources(found)[:5]

再看校验(第 577-591 行):

domain = (urllib.parse.urlparse(url).netloc or "").lower()
domain = domain[4:] if domain.startswith("www.") else domain
if domain in AUTHORITY_SOURCES:          # 白名单:维基/百度百科/科普中国…
    validated.append({"...": "...", "source": AUTHORITY_SOURCES[domain], "verified": True})
else:
    validated.append({..., "source": domain or "未知", "verified": False})

关键设计:不在白名单不是"丢弃",而是"标记为未验证"。
程序不替用户判断真假,只负责标注可信度,把判断权留给用户。
(这种"标记而非删除"的思路,在内容平台上很常见。)

块 6:关联图谱 —— 避开 N+1 查询(第 547-574 行)

卡片之间的关联存在 links(a, b) 表里(无向关系)。
列表页要给每张卡显示关联数,最容易犯的错是每张卡查一次数据库(N+1 问题)。

看正确做法:

def attach_links(cards):
    """给卡片列表批量附加关联卡片 id(供列表页显示关联数)。"""
    if not cards:
        return cards
    conn = db()
    rows = conn.execute("SELECT a,b FROM links").fetchall()   # 一次拿全部关系
    conn.close()
    m = {}
    for r in rows:                       # 在内存里建无向邻接表
        m.setdefault(r["a"], set()).add(r["b"])
        m.setdefault(r["b"], set()).add(r["a"])
    for c in cards:
        c["links"] = sorted(x for x in m.get(c["id"], set()) if x != c["id"])
    return cards

详情页则要关联卡片的标题(第 563-574 行):

ids = [x for x in links_of(cid) if x != cid]
card["links"] = ids
card["link_cards"] = []
for x in ids:
    c = card_by_id(x)
    if c:
        card["link_cards"].append({"id": x, "title": c["title"]})

前端就能渲染成可点击的"关联卡片"按钮(web.js 第 387-392 行、mobile.js 第 424-455 行)。

知识点:links_of 里的 SQL 用了 UNION 两个方向
(SELECT b ... WHERE a=? 并上 SELECT a ... WHERE b=?),
因为关系是无向的——A 关联 B 也意味着 B 关联 A。

块 7:_chat() 的流式协议(第 1431-1475 行)

self.send_response(200)
self.send_header("Content-Type", "text/event-stream; charset=utf-8")
self.send_header("Cache-Control", "no-cache")
self.send_header("X-Accel-Buffering", "no")     # 关掉反向代理缓冲(真流式)
self.send_header("Connection", "close")
self.end_headers()

def emit(event, payload):
    self.wfile.write(f"event: {event}\ndata: {json.dumps(payload, ensure_ascii=False)}\n\n".encode("utf-8"))
    self.wfile.flush()                          # 每发一条就 flush,别攒着

emit("meta", {"mode": "online" if AI_API_KEY else "offline-demo"})
...
for piece in gen:
    acc.append(piece)
    emit("delta", {"text": piece})
emit("sources", {"sources": ...})
emit("done", {"full": "".join(acc)})

第四步:对照开源,别人怎么让 AI 输出结构化数据

三种主流方案

方案做法优点 / 缺点
Prompt 约束 + 容错解析(本项目)在 prompt 里给字段和格式,代码清洗解析通用、零依赖;需要容错
JSON Mode / 结构化输出 API接口参数强制返回 JSON更可靠;依赖模型支持
Function Calling / Tool Use把字段定义成"函数参数",模型填参最可靠;概念多一点

本项目选第一种,因为它只用标准库(urllib 直接请求),
不想引入 SDK——和第 02 课"零依赖服务器"的取舍一脉相承。

对照:白名单校验

AUTHORITY_SOURCES 是一张"可信域名表",思路和
第 12 课的 .gitignore、第 16 课要讲的"动作白名单"一样:
用穷举的允许清单代替"聪明地判断",安全、可审计、易维护。

对照:关联数据的两种查法

单条查(N+1)批量查(本项目)
查询次数N+1 次1 次
代码简单稍复杂
性能数据一多就慢稳定

凡是"列表页要显示关联信息"的场景,都优先用批量 + 内存聚合。


动手练习(做完才算过关)

练习 1(热身):看协议
用 F12 的 Network 打开 /mobile,发一个问题,
观察 /api/chat 的 EventStream:找出 meta/delta/sources/done/close 各出现几次。

练习 2(必做):加一个字段
给 ai_summarize 的 system prompt 增加一个字段 confidence(0-1 的置信度),
在返回结果里带上它(记得用 _loads_json_loose 解析、做类型兜底)。
再去 web/js/mobile.js 的总结弹窗里把它显示出来。

练习 3(必做):容错实验
把 _loads_json_loose 临时改成直接 json.loads(text),
然后手动构造一段带 ```json 围栏 的文本喂给它(可以写个小脚本调用),
观察报错,再改回稳健版。这就是"为什么要有容错"的现场证明。

练习 4(必做):扩充白名单
在 AUTHORITY_SOURCES 里加一个你认可的来源域名(比如 developer.mozilla.org → "MDN"),
然后用这个链接测一次 extract_sources,看 verified 是否变成 True。

练习 5(思考题):
validate_sources 对不在白名单的链接只标记 verified: False,不丢弃。
如果改成"直接丢弃",会有什么坏处?

看答案

白名单之外也有很多正确内容。

练习 6(选做,进阶):
模仿 attach_links 的思路,写一个函数 top_linked_cards(n=3):
找出"被关联最多"的 n 张卡片并返回 [{id, title, link_count}]。
提示:先用一条 SQL 把所有关系读出来建邻接表,再按集合大小排序,
最后用 card_by_id 取标题。


本课小结

术语表

术语人话解释
结构化输出让模型返回有固定字段的数据,而不是一段话
容错解析对不规范输入做清洗后再解析
降级(fallback)首选方案失败时退到简单方案
白名单只允许清单内的东西通过
N+1 查询列表 N 条数据查了 N+1 次库,性能陷阱
邻接表用字典存"谁和谁相连",图数据的常见结构
无向关系A-B 和 B-A 是同一条关系
flush把缓冲区内容立刻发出去
X-Accel-Buffering关闭反向代理缓冲,保证真流式

下节预告

第 16 课:安全设计:白名单、令牌、签名链接。
教材换成一个很特别的项目——phone-remote(用手机遥控电脑)。
它把"电脑的敏感操作暴露给手机"这件事做得非常克制:
四道安全闸、动作白名单、签名文件链接、AI 也只能在笼子里活动。
这课教你怎么"安全地开放权限",是求职面试的加分话题。