难度:★★★(需要第 04、09、10 课基础)
教材:查资料/server.py(1580 行,纯标准库)的新增功能
预计时间:讲解 70 分钟 + 练习 50 分钟
学完你能:把 AI 的自由文本"驯服"成结构化的数据,并把它存起来、关联起来、展示出来
大模型输出的是一段自然语言,比如:
好的,我来帮你整理。首先……根据维基百科(https://zh.wikipedia.org/wiki/...),……
这段文字人看着没问题,但程序没法直接用:标题在哪?标签是什么?可信来源是哪几条?
所以要把"自由文本"变成"结构化数据"(有固定字段的 dict/JSON)。
这一课就是这条流水线的完整实现:
多轮对话文本
↓ AI 总结(约束它输出指定字段的 JSON)
结构化卡片 { title, summary, tags, body }
↓ 正则提取信源 + 白名单校验
sources: [{url, source, verified}]
↓ 存进 SQLite,并维护卡片之间的关系
可检索、可关联、可展示
cd 查资料
启动服务.bat
用手机(或浏览器模拟手机)打开 /mobile:
history)🔗 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)
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 课讲的"结构化输出"是同一招,但这里更规范:
temperature=0.3:摘要类任务要稳定,不需要创意_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 是"交办要求",容错解析是"检查作业"。
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",
}
三个必学动作:
tags 说好是数组,模型可能给你 "天文, 物理" ——字符串就按分隔符切开[:30] 硬截断保底body 为空时用原始对话 transcript,保证卡片不是空的验收标准:不管模型怎么发挥,输出的结构一定合法。
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 课的容错思路完全一致——这是这个项目最稳定的设计风格。
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]
rstrip(".,;:)]})】") 去掉句尾标点——否则 https://a.com。 会被当成 URL 的一部分any(...) 去重,最后只留 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})
关键设计:不在白名单不是"丢弃",而是"标记为未验证"。
程序不替用户判断真假,只负责标注可信度,把判断权留给用户。
(这种"标记而非删除"的思路,在内容平台上很常见。)
卡片之间的关联存在 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
邻接表(每个 id → 关联 id 集合)详情页则要关联卡片的标题(第 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。
_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)})
emit() 小函数:把"写一条 SSE"封装起来(对比第 09 课每次手拼字符串)wfile.flush():流式必须每次 flush,否则内容会卡在缓冲区meta(模式)→ delta(逐段文本)→ sources(信源)→ done(完整文本)finally 里发 close 并设 close_connection = True:主动收尾,防止连接悬挂| 方案 | 做法 | 优点 / 缺点 |
|---|---|---|
| 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 取标题。
_loads_json_loose 三层清洗:去代码围栏、取首尾花括号、再解析warning 不隐瞒UNION 两个方向emit 封装、每次 flush、边界事件(meta/sources/done/close)| 术语 | 人话解释 |
|---|---|
| 结构化输出 | 让模型返回有固定字段的数据,而不是一段话 |
| 容错解析 | 对不规范输入做清洗后再解析 |
| 降级(fallback) | 首选方案失败时退到简单方案 |
| 白名单 | 只允许清单内的东西通过 |
| N+1 查询 | 列表 N 条数据查了 N+1 次库,性能陷阱 |
| 邻接表 | 用字典存"谁和谁相连",图数据的常见结构 |
| 无向关系 | A-B 和 B-A 是同一条关系 |
| flush | 把缓冲区内容立刻发出去 |
| X-Accel-Buffering | 关闭反向代理缓冲,保证真流式 |
第 16 课:安全设计:白名单、令牌、签名链接。
教材换成一个很特别的项目——phone-remote(用手机遥控电脑)。
它把"电脑的敏感操作暴露给手机"这件事做得非常克制:
四道安全闸、动作白名单、签名文件链接、AI 也只能在笼子里活动。
这课教你怎么"安全地开放权限",是求职面试的加分话题。