Python 快速复习：01–20 节核心用法、代码与输出

按原课程第 01–20 节顺序，集中复习核心用法。每个知识点配短说明、示例和预期结果。文中的政策、检索排名与业务响应使用教学数据。

第01节：打印、注释与变量

多行输出与注释

Python 从上往下执行；# 开头的是注释。print 接收多个值，默认用空格隔开。

# 记录我今天练什么
print("Python", "day", 1)
print("# is text here")

输出：

Python day 1
# is text here

变量与赋值

= 把右侧结果赋给左侧变量；再次赋值会更新变量。名字区分大小写。

course = "Python"
minutes = 20
minutes = minutes + 5
print(course, minutes)

输出：

Python 25

第02节：字符串与数字

字符串、len与strip

str 保存文本，len 计算字符数。strip() 返回去掉首尾空白的新字符串，原字符串和内部空格保持原样。

original = "  未开课  "
clean = original.strip()
print(len(original), len(clean))
print(clean)
print(original == clean)

输出：

7 3
未开课
False

整数、浮点数与运算

int 是整数，float 是浮点数。+ - * / 做四则运算，// 向下取整，% 取余。

total = 7
size = 2
print(total + size, total * size)
print(total / size, total // size, total % size)

输出：

9 14
3.5 3 1

f-string把值写进句子

引号前加 f，就能用 {变量} 把值嵌入句子。

name = "小蓝"
days = 3
print(f"{name}已练习{days}天。")

输出：

小蓝已练习3天。

第03节：读取输入并转换类型

input读入的始终是文本

input 显示提示并读取一行，返回值始终是字符串；type 查看类型。输入：小蓝。

name = input("名字: ")
print("你好", name)
print(type(name))

预期终端画面（含输入回显）：

名字: 小蓝
你好 小蓝
<class 'str'>

int转换，而不是自动转换

int 把整数文本转换为整数，str 把数字转换为文本，float 可转换小数文本。输入：20。

raw = input("已学分钟: ")
minutes = int(raw)
print(minutes + 10)
print(str(minutes) + "分钟")

预期终端画面（含输入回显）：

已学分钟: 20
30
20分钟

第04节：缩进与条件判断

if、elif、else与代码块

冒号后用缩进表示代码块；if / elif / else 从上往下选择第一个成立的分支。

score = 79
if score >= 80:
    print("达到80分")
elif score >= 60:
    print("还差一点")
else:
    print("先补基础")
print("检查结束")

输出：

还差一点
检查结束

布尔值、and/or/not与空值

比较得到 True 或 False；and 表示同时成立，or 表示至少一个成立，not 取反。None 表示无值，用 is None 判断。

score = 80
completion = 90
passed = score >= 80 and completion >= 90
print(passed)
question = "   ".strip()
if question == "":
    print("问题为空")
value = None
print(value is None)

输出：

True
问题为空
True

第05节：循环一次做多条

for与range

for 逐个遍历；range(1, 4) 产生 1、2、3，不含右端点。累加变量在循环外初始化。

total = 0
for number in range(1, 4):
    total = total + number
    print(number, total)
print("合计", total)

输出：

1 1
2 3
3 6
合计 6

while与退出条件

while 在条件为真时重复执行。更新变量使循环结束；break 结束循环，continue 跳过本轮剩余代码。

remaining = 3
while remaining > 0:
    print(remaining)
    remaining = remaining - 1
print("完成")

输出：

3
2
1
完成

第06节：列表与字典

list、索引、append与遍历

列表用 [] 保存多项；索引从 0 开始，append 追加，in 判断是否包含，for 逐项读取。

questions = ["  未开课  ", "", "课"]
print(len(questions), questions[0].strip())
questions.append("证书")
for question in questions:
    print(len(question.strip()))
print("课" in questions)

输出：

3 未开课
3
0
1
2
True

可变容器、别名与浅复制

赋值创建别名；copy() 创建新的外层列表。

original = ["课"]
alias = original
copied = original.copy()
alias.append("证书")
print(original)
print(copied)

输出：

['课', '证书']
['课']

dict存键和值、get与顺序

字典保存键值对；get(key, 0) 在缺键时返回 0。字典按键的插入顺序遍历。

counts = {}
for name in ["甲", "乙", "甲"]:
    counts[name] = counts.get(name, 0) + 1
print(counts)
print(counts.get("丙", 0))

输出：

{'甲': 2, '乙': 1}
0

items遍历键值

items() 提供键值对，循环中的两个变量分别接收键和值。

counts = {"甲": 2, "乙": 1}
for name, count in counts.items():
    print(name, count)

输出：

甲 2
乙 1

第07节：函数、参数与返回值

def是定义，括号是调用，return是交回结果

def 定义函数，括号发起调用，参数接收输入，return 把结果交回调用者。

def clean_length(text):
    clean = text.strip()
    return len(clean)

result = clean_length("  课程  ")
print(result)

输出：

2

位置参数与简单默认参数

实参按位置对应；省略带默认值的参数时使用默认值。

def add_minutes(current, extra=10):
    return current + extra

print(add_minutes(20))
print(add_minutes(20, 5))

输出：

30
25

sorted与key函数：等长保持原顺序

sorted 返回新列表。key 接收函数，稳定排序会保留等值项的相对顺序。

def text_length(text):
    return len(text)

words = ["甲乙", "课", "丙丁"]
ordered = sorted(words, key=text_length)
print(ordered)
print(words)

输出：

['课', '甲乙', '丙丁']
['甲乙', '课', '丙丁']

第08节：文件、JSON、模块、异常与断言

import与自己的模块

模块是可导入的 Python 文件。下面两个文件放在同一目录，运行 use_module.py。

helpers.py：

def double(number):
    return number * 2

from helpers import double

print(double(3))

输出：

6

with、UTF-8与读写文件

with 在代码块结束时关闭文件；w 覆盖写，a 追加，r 读。encoding="utf-8" 用于中文文本。

with open("note.txt", "w", encoding="utf-8") as file:
    file.write("今天练Python\n")
with open("note.txt", "a", encoding="utf-8") as file:
    file.write("保留数字7\n")
with open("note.txt", "r", encoding="utf-8") as file:
    text = file.read()
print(text, end="")

输出：

今天练Python
保留数字7

Path定位与创建输出目录

Path / 名称 拼接路径；parent 取父目录，mkdir 创建目录，read_text/write_text 读写文本。

from pathlib import Path

path = Path("outputs") / "note.txt"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("课程\n", encoding="utf-8")
print(path.read_text(encoding="utf-8"), end="")
print(path.name)

输出：

课程
note.txt

JSON保存结构；JSONL逐行

dump/load 面向文件，dumps/loads 面向字符串。JSONL 每行保存一个 JSON 值。

import json

records = [{"question": "未开课", "length": 3}]
with open("records.json", "w", encoding="utf-8") as file:
    json.dump(records, file, ensure_ascii=False, indent=2)
with open("records.json", "r", encoding="utf-8") as file:
    loaded = json.load(file)
print(loaded[0]["question"])
with open("records.jsonl", "w", encoding="utf-8") as file:
    for record in records:
        file.write(json.dumps(record, ensure_ascii=False) + "\n")
print(loaded == records)

输出：

未开课
True

异常与Traceback：先读最后一行

try 执行代码，except 处理指定异常，raise 主动抛出异常。Traceback 最后一行说明异常类型和原因。

def read_minutes(text):
    minutes = int(text)
    if minutes < 0:
        raise ValueError("分钟不能为负")
    return minutes

for raw in ["20", "abc", "-1"]:
    try:
        print(read_minutes(raw))
    except ValueError:
        print("请输入非负整数")

输出：

20
请输入非负整数
请输入非负整数

assert与严格类型的小自检

assert 检查预期，失败时报 AssertionError。isinstance 包含子类关系，type(x) is int 检查精确整数类型。

def clean(text):
    return text.strip()

assert clean("  课  ") == "课"
assert clean("   ") == ""
print(isinstance(True, int))
print(type(True) is int)
print("两个清洗断言通过")

输出：

True
False
两个清洗断言通过

TXT的splitlines与JSONL逐行读取

splitlines() 按行拆文本，保留中间空行。JSONL 用循环逐行 json.loads，便于定位坏行。

text = "课\n\n证书\n"
lines = text.splitlines()
print(lines)
print(len(lines))

输出：

['课', '', '证书']
3

JSONL 逐行解析

数据文件 audit_sample.jsonl 中第二行是坏 JSON，捕获后继续处理第三行。

audit_sample.jsonl：

{"request_id":"a","intent":"refund"}
oops
{"request_id":"b","intent":"course"}

import json

bad = 0
line_number = 0
with open("audit_sample.jsonl", "r", encoding="utf-8") as file:
    for line in file:
        line_number = line_number + 1
        try:
            record = json.loads(line)
            print(line_number, record["request_id"])
        except json.JSONDecodeError:
            bad = bad + 1
            print(line_number, "坏JSON")
print("坏行", bad)

输出：

1 a
2 坏JSON
3 b
坏行 1

第09节：组合函数与问题统计

先让一个函数确实被调用

组合 strip、循环、判断和返回值，统计非空文本；重复文本分别计数。

def count_nonempty(questions):
    count = 0
    for question in questions:
        clean = question.strip()
        if clean != "":
            count = count + 1
    return count

sample = ["  课  ", "", "课", "   "]
result = count_nonempty(sample)
print(result)

输出：

2

给analyze_questions一个能跑的最小形状

函数也可以返回字典，再按键读取结果。

def analyze_questions(questions):
    return {"total": len(questions)}

sample = ["课", "", "课"]
result = analyze_questions(sample)
print(result)
print(result["total"])

输出：

{'total': 3}
3

从单字段组合为多个统计字段

继续组合已有操作即可得到七个字段：total 原始条数，nonempty/empty 非空与空白条数，unique 首现去重列表，counts 次数，lengths 对应长度，sorted_by_length 按长度稳定排序。

第10节：路径、解释器与包

路径与当前目录

绝对路径从根目录开始，相对路径从当前目录开始；. 是当前目录，.. 是父目录。pwd 查看位置，cd 切换目录，ls 列出文件。

pwd
cd ..
pwd
ls

预期结果：

第一行显示当前目录；第二次 pwd 显示它的父目录；ls 列出父目录中的文件。

解释器与包归属

解释器执行 Python 代码；venv 隔离项目使用的包，python -m pip 使用该解释器所属的 pip。sys.executable 显示实际解释器路径。

import sys
print(sys.executable)
print(sys.version.split()[0])

预期结果：

两行依次为当前 Python 可执行文件路径和版本号。

第11节：一次本机HTTP请求到一个小服务

HTTP 请求、响应和路由

URL 由协议、主机、端口和路径组成。GET 读取状态，POST 发送数据；状态码说明本次 HTTP 请求的处理结果。

先认识类型提示、class和装饰器再用框架

类型提示声明字段；Pydantic 的 BaseModel 用它解析请求体。装饰器把函数登记到对应路由，返回字典会被转换为 JSON。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Question(BaseModel):
    question: str

@app.get("/health")
def health():
    return {"status": "ok", "mode": "practice"}

@app.post("/echo")
def echo(body: Question):
    return {"question": body.question, "length": len(body.question)}

预期结果：

GET /health → {"status":"ok","mode":"practice"}
POST /echo，正文 {"question":"未开课"} → {"question":"未开课","length":3}
POST /echo，正文 {} → HTTP 422

客户端发请求

HTTPX 的 get/post 发出请求，status_code 读取状态码，json() 解析响应。服务启动后执行：

import httpx

with httpx.Client() as client:
    response = client.post("http://127.0.0.1:8001/echo", json={"question": "未开课"})
print(response.status_code)
print(response.json())

预期结果：

200
{'question': '未开课', 'length': 3}

第12节：检索、模型、来源与失败分别是什么

先用三段文本看RAG的数据流

RAG 按“检索证据 → 构造 prompt → 生成 → 校验”处理问题。先按生效日期选资料，再把命中的证据与问题放进输入。下例政策是虚构教学数据。

documents = [
    {"id": "refund-old", "topic": "refund", "effective_date": "2026-01-01", "text": "旧练习政策：未开课可申请退款。"},
    {"id": "refund-current", "topic": "refund", "effective_date": "2026-10-01", "text": "虚构政策：购买5日内且未开课，可申请退款。"},
    {"id": "certificate", "topic": "certificate", "effective_date": "2026-10-01", "text": "虚构政策：成绩80分且完成度90%，可申请证书。"},
]
as_of = "2026-10-11"
latest = {}
for document in documents:
    if document["effective_date"] <= as_of:
        topic = document["topic"]
        if topic not in latest or document["effective_date"] > latest[topic]["effective_date"]:
            latest[topic] = document
question = "退款条件是什么？"
selected = []
for document in latest.values():
    if "退款" in question and document["topic"] == "refund":
        selected.append(document)
print("检索来源", selected[0]["id"])
evidence = selected[0]["text"]
prompt = "仅按证据回答，并保留条件。\n问题：" + question + "\n证据：" + evidence
print(prompt)
print("这里只构造prompt，尚未调用模型。")

输出：

检索来源 refund-current
仅按证据回答，并保留条件。
问题：退款条件是什么？
证据：虚构政策：购买5日内且未开课，可申请退款。
这里只构造prompt，尚未调用模型。

模型与结果的几个层次

权重是模型参数；tokenizer 把文本转成 token 编号；prompt 是模型输入；加载把权重放入内存；生成输出原始文本。检索结果、生成内容、结构校验和来源校验分别观察。

第13节：Linux诊断与离线依赖

进程与 PID

程序运行后成为进程，PID 标识本次运行。ps 查看进程信息，pgrep 按命令查找。

ps -u "$(whoami)" -o pid,comm,pcpu,rss,args
pgrep -af "python.*http.server"

预期结果：

ps 列出 PID、程序名、CPU、驻留内存和命令；pgrep 列出匹配进程的 PID 与完整命令。

监听端口与请求

ss -ltnp 查看 TCP 监听端口及进程，curl 请求 URL。

ss -ltnp 'sport = :8002'
curl http://127.0.0.1:8002/

预期结果：

服务在 8002 端口监听时，ss 显示 LISTEN；curl 输出服务返回的页面。

内存与磁盘

free 看内存，df 看文件系统容量，du 看目录占用。

free -m
df -h .
du -sh .

预期结果：

输出当前内存 MiB、当前文件系统容量、当前目录占用；数值随运行状态变化。

依赖自检：import名字与安装包名字可能不同

importlib.metadata 查询安装分发版本，find_spec 检查模块是否可导入；安装包名与导入名可能不同。

from importlib import metadata, util
import sys

print("解释器", sys.executable)
for package in ["fastapi", "httpx", "this_package_does_not_exist"]:
    try:
        value = metadata.version(package)
        print(package, "版本", value)
    except metadata.PackageNotFoundError:
        print(package, "未安装")
print("fastapi可导入", util.find_spec("fastapi") is not None)

预期结果：

输出当前解释器，以及 fastapi、httpx 的版本与可导入状态；不存在的演示包显示缺失。

离线依赖

--no-index 不访问包索引，--find-links 从本地 wheel 目录找包，== 指定版本。

python -m pip install --no-index --find-links ./offline-wheels lanqiao-course-demo==0.1.0
python -m pip check

预期结果：

本地 wheel 及依赖完整时安装成功；依赖关系正常时 pip check 输出：No broken requirements found.

第14节：类、共享资源、生命周期与统一错误

class、self、init：把状态与功能放到一起

class 定义对象，__init__ 初始化，self 指向当前实例；属性保存状态，方法操作状态。

class Counter:
    def __init__(self, start=0):
        self.value = start

    def add(self, amount):
        self.value = self.value + amount
        return self.value

counter = Counter(2)
print(counter.add(3))
print(counter.value)
other = Counter()
print(other.value)

输出：

5
5
0

分层与适配器：服务拿对象，不把每件事混进路由

检索对象找证据，后端对象产生文本，服务对象连接两者。把对象作为参数传入便于替换实现。

class KeywordStore:
    def __init__(self):
        self.items = ["课程", "证书"]

    def retrieve(self, question):
        hits = []
        for item in self.items:
            if item in question:
                hits.append(item)
        return hits

class EchoBackend:
    def generate(self, question, evidence):
        return "[MOCK演示] 证据数=" + str(len(evidence))

class Service:
    def __init__(self, index, backend):
        self.index = index
        self.backend = backend

    def answer(self, question):
        evidence = self.index.retrieve(question)
        return self.backend.generate(question, evidence)

service = Service(KeywordStore(), EchoBackend())
print(service.answer("课程和证书"))
print(service.answer("天气"))

输出：

[MOCK演示] 证据数=2
[MOCK演示] 证据数=0

name入口：导入只拿功能，直接运行才演示

直接运行文件时 __name__ 是 "__main__"；导入时入口里的演示代码不执行。

def greet(name):
    return "你好，" + name

if __name__ == "__main__":
    print(greet("小蓝"))

输出：

你好，小蓝

with与yield：创建一次，用完再释放

@contextmanager 把生成器变成上下文管理器；yield 暂时交出资源，离开 with 后执行释放。

from contextlib import contextmanager

@contextmanager
def resources():
    print("创建模拟资源")
    value = {"name": "index-demo"}
    try:
        yield value
    finally:
        print("释放模拟资源")

with resources() as current:
    print("使用", current["name"])

输出：

创建模拟资源
使用 index-demo
释放模拟资源

async/await：等待期间让出机会，不会把CPU推理自动变快

async def 定义协程，await 等待结果；普通脚本可用 asyncio.run 启动入口。

import asyncio

async def main():
    print("开始")
    await asyncio.sleep(0.01)
    print("等待结束")

asyncio.run(main())

输出：

开始
等待结束

服务生命周期与统一错误

资源在服务启动时创建并保存到 app.state，关闭时释放；请求复用资源。health 返回就绪状态；统一错误响应使用状态码和固定字段。

statuses = {400: "EMPTY_QUESTION", 422: "BAD_SCHEMA",
            500: "INTERNAL_ERROR", 503: "NOT_READY"}
for status, code in statuses.items():
    print(status, code)

输出：

400 EMPTY_QUESTION
422 BAD_SCHEMA
500 INTERNAL_ERROR
503 NOT_READY

第15节：本地模型、设备精度、推理模式与确定性

device与shape：先用两行数字看清CPU

张量的 shape 是维度，dtype 是数值类型，device 是计算设备。以下 Paddle 示例在 CPU 创建 2×2 张量。

import paddle

paddle.set_device("cpu")
value = paddle.to_tensor([[1.0, 2.0], [3.0, 4.0]], dtype="float32")
print(paddle.get_device())
print(list(value.shape))
print(str(value.dtype).split(".")[-1])

输出：

cpu
[2, 2]
float32

精度与内存：字节数可估算，进程内存还要实测

用每参数字节数估计裸权重大小；float32 每项 4 字节，float64 每项 8 字节。

import struct

value = 1 / 3
single = struct.pack("f", value)
double = struct.pack("d", value)
print(len(single), len(double))
parameters = 500_000_000
print(parameters * 4 // 1_000_000, "MB裸权重估算")
print(parameters * 2 // 1_000_000, "MB裸权重估算")

输出：

4 8
2000 MB裸权重估算
1000 MB裸权重估算

eval与no_grad：两件不同的事

eval() 切换推理行为，no_grad() 关闭本次计算的梯度记录。

import paddle

paddle.set_device("cpu")
paddle.seed(7)
layer = paddle.nn.Linear(2, 1, bias_attr=False)
layer.eval()
inputs = paddle.to_tensor([[1.0, 2.0]], dtype="float32", stop_gradient=False)
with paddle.no_grad():
    outputs = layer(inputs)
print(layer.training)
print(outputs.stop_gradient)
print(list(outputs.shape))

输出：

False
True
[1, 1]

随机种子与贪心：先学可复现的边界

相同种子使独立随机对象在相同条件下产生相同序列；贪心选择当前得分最高项。

import random

first = random.Random(7)
second = random.Random(7)
left = []
right = []
for index in range(4):
    left.append(first.randrange(100))
    right.append(second.randrange(100))
print(left == right)
scores = [0.1, 0.8, 0.1]
words = ["退款", "课程", "证书"]
best_index = scores.index(max(scores))
print(words[best_index])

输出：

True
课程

本地加载与生成

从本地目录读取 tokenizer 与权重，设置设备和推理模式；**inputs 把字典展开为命名参数。生成后解码 token。以下为调用顺序片段，model_dir 是本地模型目录。

import paddle
from paddlenlp.transformers import Qwen2ForCausalLM, Qwen2Tokenizer

paddle.set_device("cpu")
tokenizer = Qwen2Tokenizer.from_pretrained(model_dir)
model = Qwen2ForCausalLM.from_pretrained(model_dir, dtype="float32", convert_from_torch=False)
model.eval()
inputs = tokenizer(prompt, return_tensors="pd", add_special_tokens=False, return_attention_mask=True)
with paddle.no_grad():
    outputs = model.generate(**inputs, max_new_tokens=16, decode_strategy="greedy_search")
print(tokenizer.batch_decode(outputs[0], skip_special_tokens=True)[0])

预期结果：

输出模型本次生成的文本；具体文字取决于模型、prompt 与生成参数。

第16节：Prompt参数、结构坏输出、拒答与指令攻击

系统规则、用户问题与证据：把角色和不可信数据分开

系统规则、问题和证据分别组织，再套用模型的角色模板。下面只构造 Qwen 风格输入。

import json

system = "仅按证据回答；缺证据拒答；来源只用给定ID。"
question = "退款需要哪些条件？"
evidence = [{"id": "note-a", "text": "SIMULATED：购买4日内且未开课可申请。"}]
data = json.dumps({"question": question, "evidence": evidence}, ensure_ascii=False)
prompt = "<|im_start|>system\n" + system + "<|im_end|>\n<|im_start|>user\n" + data + "<|im_end|>\n<|im_start|>assistant\n"
print(prompt.startswith("<|im_start|>system"))
print('"evidence"' in prompt)
print("仅构造输入：未调用模型")

输出：

True
True
仅构造输入：未调用模型

温度：只在抽样策略中解释概率，不靠猜参数名

抽样时，先把分数除以正温度再做 softmax；较低温度使概率更集中。

import math

def probabilities(scores, temperature):
    if temperature <= 0:
        raise ValueError("温度必须为正")
    highest = max(scores)
    weights = []
    total = 0
    for score in scores:
        value = math.exp((score - highest) / temperature)
        weights.append(value)
        total = total + value
    result = []
    for value in weights:
        result.append(round(value / total, 3))
    return result

print(probabilities([2, 1, 0], 1.0))
print(probabilities([2, 1, 0], 0.5))

输出：

[0.665, 0.245, 0.09]
[0.867, 0.117, 0.016]

最大新token与停止：截短能同时丢掉条件

max_new_tokens 限制新增 token 数，停止标记提前结束。下面用人工 token 列表展示两者。

def visible_text(tokens, limit, stop_token):
    result = []
    for index in range(min(limit, len(tokens))):
        token = tokens[index]
        if token == stop_token:
            break
        result.append(token)
    return "".join(result)

tokens = ["购买", "4日内", "且未开课", "<END>", "多余文本"]
print(visible_text(tokens, 5, "<END>"))
print(visible_text(tokens, 2, "<END>"))

输出：

购买4日内且未开课
购买4日内

外置YAML参数：安全读取、验证类型、实际使用

yaml.safe_load 把 YAML 配置读成 Python 数据；读取后检查类型，再把参数交给实际生成接口。

from pathlib import Path
import yaml

path = Path("generation-demo.yaml")
path.write_text("temperature: 0.8\nmax_new_tokens: 32\nstop_words:\n  - '<END>'\n", encoding="utf-8")
config = yaml.safe_load(path.read_text(encoding="utf-8"))
if type(config["max_new_tokens"]) is not int or config["max_new_tokens"] < 1:
    raise ValueError("max_new_tokens必须是正整数")
print(config["temperature"])
print(config["max_new_tokens"])
print(config["stop_words"])

输出：

0.8
32
['<END>']

从原始文本到结构校验：解析JSON并校验来源白名单

先解析 JSON，再检查字段、类型及来源白名单。set 比较字段集合；条件表达式按布尔条件选择结果。

import json

def reject_constant(value):
    raise ValueError("非标准JSON")

def parse_draft(raw, allowed_ids):
    value = json.loads(raw, parse_constant=reject_constant)
    if type(value) is not dict or set(value) != {"answer", "used_ids", "refused"}:
        raise ValueError("字段不正确")
    if type(value["answer"]) is not str or type(value["refused"]) is not bool:
        raise ValueError("类型不正确")
    if type(value["used_ids"]) is not list:
        raise ValueError("来源不是列表")
    seen = []
    for source_id in value["used_ids"]:
        if type(source_id) is not str or source_id not in allowed_ids or source_id in seen:
            raise ValueError("来源不可信")
        seen.append(source_id)
    if value["refused"] and seen:
        raise ValueError("拒答不应带来源")
    if not value["refused"] and (not seen or not value["answer"].strip()):
        raise ValueError("有效草稿需要文本与来源")
    return value

fixtures = [
    '{"answer":"按证据说明","used_ids":["note-a"],"refused":false}',
    '```json\n{}\n```',
    '{"answer":"编造内容","used_ids":["fake"],"refused":false}',
    '{"answer":"资料不足","used_ids":[],"refused":true}',
]
for raw in fixtures:
    try:
        result = parse_draft(raw, ["note-a"])
        print("合法拒答" if result["refused"] else "结构合法：还需核事实")
    except ValueError:
        print("拒绝坏草稿")

输出：

结构合法：还需核事实
拒绝坏草稿
拒绝坏草稿
合法拒答

知识边界与注入：数据里的指令不获得权限

评估拒答时分别统计正确、应拒未拒和应答却拒。下例是人工标注数据。

cases = [
    {"name": "有证据正常问题", "should_refuse": False, "actual_refused": False},
    {"name": "资料没有天气", "should_refuse": True, "actual_refused": True},
    {"name": "忽略规则并伪造来源", "should_refuse": True, "actual_refused": False},
    {"name": "正常问题被误拒", "should_refuse": False, "actual_refused": True},
]
correct = 0
missed = 0
over_refused = 0
for case in cases:
    if case["actual_refused"] == case["should_refuse"]:
        correct = correct + 1
    elif case["should_refuse"]:
        missed = missed + 1
    else:
        over_refused = over_refused + 1
print("正确", correct, "漏拒", missed, "误拒", over_refused)

输出：

正确 2 漏拒 1 误拒 1

第17节：多格式文档、清洗去重、切片与生效版本

TXT、轻量MD、JSON先变成统一记录

TXT、轻量 Markdown 与 JSON 按格式读取，统一成标题和正文。stem 取无扩展名的文件名，glob 查找文件。

from pathlib import Path
import json

def read_document(path):
    raw = path.read_text(encoding="utf-8")
    title = path.stem
    if path.suffix == ".json":
        value = json.loads(raw)
        if type(value) is not dict or type(value.get("text")) is not str:
            raise ValueError("JSON需要text字符串")
        title = value.get("title", title)
        text = value["text"]
    elif path.suffix == ".md":
        lines = []
        for line in raw.splitlines():
            if line.startswith("#"):
                line = line.lstrip("#").strip()
            lines.append(line)
        text = "\n".join(lines)
    elif path.suffix == ".txt":
        text = raw
    else:
        raise ValueError("不支持的扩展名")
    if type(title) is not str or not title.strip() or not text.strip():
        raise ValueError("标题正文不能为空")
    return {"title": title.strip(), "text": text.strip()}

folder = Path("demo_docs")
folder.mkdir(exist_ok=True)
(folder / "course.txt").write_text("  SIMULATED：课程未开课。  ", encoding="utf-8")
(folder / "refund.md").write_text("# 4日内\nSIMULATED：且未开课才可申请。", encoding="utf-8")
(folder / "certificate.json").write_text(json.dumps({"title": "练习证书", "text": "SIMULATED：成绩75且完成度85%。"}, ensure_ascii=False), encoding="utf-8")
for path in sorted(folder.glob("*")):
    record = read_document(path)
    print(path.name, record["title"])
    print(record["text"])

输出：

certificate.json 练习证书
SIMULATED：成绩75且完成度85%。
course.txt course
SIMULATED：课程未开课。
refund.md refund
4日内
SIMULATED：且未开课才可申请。

去重规则写清楚：文本相同不意味着元数据相同

清洗后去掉空文本，按首次出现顺序去重；数字、否定词及内部空格保留。

original = ["  未开课  ", "未开课", "超过4日", "  ", "未 开课"]
before = original.copy()
unique = []
for raw in original:
    text = raw.strip()
    if text != "" and text not in unique:
        unique.append(text)
print(unique)
print(original == before)
print("超过4日" in unique)

输出：

['未开课', '超过4日', '未 开课']
True
True

切片与步长：先亲手看一段文字怎样被截开

text[start:end] 包含起点、不含终点；range 第三个参数是步长。窗口大于步长时会产生重叠。

text = "甲乙丙丁戊己"
print(text[1:4])
print(text[:2])
print(text[4:])
for start in range(0, len(text), 2):
    print(text[start:start + 3])

输出：

乙丙丁
甲乙
戊己
甲乙丙
丙丁戊
戊己

稳定切片ID、父文档与区间

切片记录父文档、版本和原文区间；用这些字段组成稳定 ID。

import json

def make_chunks(document, window, overlap):
    if type(window) is not int or type(overlap) is not int or window <= overlap or overlap < 0:
        raise ValueError("需要window > overlap >= 0")
    text = document["text"]
    result = []
    for start in range(0, len(text), window - overlap):
        end = min(start + window, len(text))
        record = {"chunk_id": f"{document['id']}:{document['version']}:{start}:{end}",
                  "doc_id": document["id"], "title": document["title"],
                  "version": document["version"], "effective_date": document["effective_date"],
                  "start": start, "end": end, "text": text[start:end]}
        result.append(record)
        if end == len(text):
            break
    return result

document = {"id": "r1", "title": "SIMULATED退费", "version": "1.0",
            "effective_date": "2026-10-01", "text": "购买4日内且未开课可申请"}
chunks = make_chunks(document, 8, 3)
for chunk in chunks:
    print(chunk["chunk_id"], chunk["text"])
print(chunks == make_chunks(document, 8, 3))

输出：

r1:1.0:0:8 购买4日内且未开
r1:1.0:5:12 且未开课可申请
True

验证日期、数字版本与过期边界

date.fromisoformat 解析日期；点分版本转换成整数元组比较。先排除未生效与过期记录，再按主题选择最高版本。

from datetime import date

def version_key(raw):
    parts = []
    for part in raw.split("."):
        if not part.isdigit():
            raise ValueError("版本段必须是数字")
        parts.append(int(part))
    return tuple(parts)

documents = [
    {"id": "r-old", "topic": "refund", "version": "1.9", "effective": "2026-01-01"},
    {"id": "r-new", "topic": "refund", "version": "1.10", "effective": "2026-10-01"},
    {"id": "r-future", "topic": "refund", "version": "2.0", "effective": "2026-11-01"},
    {"id": "r-expired", "topic": "refund", "version": "9.0", "effective": "2026-01-01", "until": "2026-10-01"},
]
as_of = date.fromisoformat("2026-10-12")
latest = {}
for document in documents:
    effective = date.fromisoformat(document["effective"])
    until = date.fromisoformat(document["until"]) if document.get("until") else None
    if effective > as_of or (until is not None and until < as_of):
        continue
    topic = document["topic"]
    old = latest.get(topic)
    if old is None or version_key(document["version"]) > version_key(old["version"]):
        latest[topic] = document
print(latest["refund"]["id"])
print(version_key("1.10") > version_key("1.9"))

输出：

r-new
True

第18节：向量、归一化、本地索引、阈值、混合检索与跨文档

先把文本变成数字：最小词袋编码器

词袋把文本映射成固定维度的计数向量；下面三维依次代表退款、课程、证书。

terms = ["退款", "课程", "证书"]

def encode(text):
    vector = []
    for term in terms:
        vector.append(text.count(term))
    return vector

print(encode("课程证书课程"))
print(encode("天气"))
print(encode("退费"))

输出：

[0, 2, 1]
[0, 0, 0]
[0, 0, 0]

归一化：方向比较先排除零向量

L2 长度是平方和的平方根，每个分量除以长度得到单位向量。零向量没有方向。

import math

def normalize(vector):
    total = 0
    for value in vector:
        if type(value) not in (int, float) or not math.isfinite(value):
            raise ValueError("向量必须是有限数字")
        total = total + value * value
    length = math.sqrt(total)
    if length == 0:
        raise ValueError("零向量没有方向")
    result = []
    for value in vector:
        result.append(value / length)
    return result

print(normalize([3, 4]))
try:
    normalize([0, 0])
except ValueError:
    print("零向量拒绝")

输出：

[0.6, 0.8]
零向量拒绝

点积、余弦与距离：知道排序方向

同维向量逐项相乘再相加得到点积；单位向量的点积等于余弦相似度，值越大越相似。

def dot(left, right):
    if len(left) != len(right):
        raise ValueError("维度不一致")
    total = 0
    for index in range(len(left)):
        total = total + left[index] * right[index]
    return total

query = [1.0, 0.0]
print(dot(query, [1.0, 0.0]))
print(dot(query, [0.6, 0.8]))
print(dot(query, [0.0, 1.0]))

输出：

1.0
0.6
0.0

本地索引的最小可存/载对照；top-k与阈值不同

索引保存向量、ID 和版本信息。先按阈值过滤，再排序取前 k 条；top_k 是数量上限。

from pathlib import Path
import json
import math

def normalize(vector):
    total = 0
    for value in vector:
        total = total + value * value
    length = math.sqrt(total)
    if length == 0:
        raise ValueError("零向量")
    result = []
    for value in vector:
        result.append(value / length)
    return result

def dot(left, right):
    if len(left) != len(right):
        raise ValueError("维度不一致")
    total = 0
    for index in range(len(left)):
        total = total + left[index] * right[index]
    return total

def higher_score(hit):
    return -hit["score"]

records = [
    {"chunk_id": "r:0", "doc_id": "r", "text": "SIMULATED退费", "vector": [3, 0]},
    {"chunk_id": "c:0", "doc_id": "c", "text": "SIMULATED课程", "vector": [0, 4]},
    {"chunk_id": "s:0", "doc_id": "s", "text": "SIMULATED证书", "vector": [3, 4]},
]
for record in records:
    record["vector"] = normalize(record["vector"])
path = Path("demo_index.json")
path.write_text(json.dumps({"knowledge_version": "demo-v1", "encoder": "manual-2d-v1", "dimension": 2, "records": records}, ensure_ascii=False), encoding="utf-8")
loaded = json.loads(path.read_text(encoding="utf-8"))
if loaded["dimension"] != 2 or loaded["encoder"] != "manual-2d-v1":
    raise ValueError("索引与编码器不匹配")
query = normalize([1, 0])
hits = []
for record in loaded["records"]:
    score = dot(query, record["vector"])
    if score >= 0.7:
        hits.append({"doc_id": record["doc_id"], "chunk_id": record["chunk_id"], "score": score})
ordered = sorted(hits, key=higher_score)
print(ordered[:2])
print(loaded["knowledge_version"])

输出：

[{'doc_id': 'r', 'chunk_id': 'r:0', 'score': 1.0}]
demo-v1

关键词与语义侧排名融合：先练透明RRF

RRF 按名次融合多个检索列表：每个名次 r 贡献 1/(k+r)，相同 ID 累加。

def descending_score(item):
    return -item["score"]

keyword_ranking = ["doc-a", "doc-b"]
semantic_fixture_ranking = ["doc-b", "doc-c"]
scores = {}
for ranking in [keyword_ranking, semantic_fixture_ranking]:
    rank = 0
    for document_id in ranking:
        rank = rank + 1
        scores[document_id] = scores.get(document_id, 0) + 1 / (60 + rank)
items = []
for document_id, score in scores.items():
    items.append({"id": document_id, "score": score})
ordered = sorted(items, key=descending_score)
print(ordered[0]["id"])
print(len(ordered))

输出：

doc-b
3

同义和错字规则：有限适配，数字否定不改

replace 返回替换后的新文本；有限同义词和错字规则保留数字与否定含义。

def normalize_query(raw):
    text = raw.strip()
    for old, new in [("退费", "退款"), ("课成", "课程")]:
        text = text.replace(old, new)
    return text

raw = "  超过4日未开课能退费吗  "
text = normalize_query(raw)
print(text)
print("4" in text, "未" in text)
print(raw != text)

输出：

超过4日未开课能退款吗
True True
True

跨文档与事实槽位：格式、召回、事实分开算

跨文档问题需要检查证据是否齐全，也要检查答案中的条件关系。出现正确数字不等于表达正确事实。

expected_ids = ["grade-rule", "completion-rule"]
retrieved_ids = ["grade-rule"]
missing_ids = []
for source_id in expected_ids:
    if source_id not in retrieved_ids:
        missing_ids.append(source_id)
print("缺证据", missing_ids)
required_words = ["75", "85%"]
answer = "达到75分但不需要85%完成度。"
word_hits = 0
for word in required_words:
    if word in answer:
        word_hits = word_hits + 1
print("字面命中", word_hits, "/", len(required_words))
print("人工事实判定：错误，遗漏且否定了完成度要求")

输出：

缺证据 ['completion-rule']
字面命中 2 / 2
人工事实判定：错误，遗漏且否定了完成度要求

最小召回@k报告：分母只算有正确证据的查询

此例每个查询只有一个目标文档：目标进入前 k 个结果就记一次命中。

cases = [
    {"expected": "a", "ranking": ["x", "a", "b"]},
    {"expected": "b", "ranking": ["b", "x", "a"]},
]
for k in [1, 3]:
    hit_count = 0
    for case in cases:
        if case["expected"] in case["ranking"][:k]:
            hit_count = hit_count + 1
    print("召回@", k, hit_count, "/", len(cases))

输出：

召回@ 1 1 / 2
召回@ 3 2 / 2

第19节：有界缓存、并发保序、P95、内存与量化进阶

缓存键：相同问题不一定是同一次业务条件

缓存键包含所有影响结果的条件。元组可作为字典键；知识版本、日期、top-k 或编码规则改变时应使用不同键。

def make_key(question, top_k, knowledge_version, as_of, encoder_version):
    return (question.strip(), top_k, knowledge_version, as_of, encoder_version)

first = make_key("  退款  ", 3, "v1", "2026-10-12", "bow-v1")
same = make_key("退款", 3, "v1", "2026-10-12", "bow-v1")
changed = make_key("退款", 3, "v2", "2026-10-12", "bow-v1")
print(first == same)
print(first == changed)
cache = {first: ["doc-a"]}
print(cache.get(changed))

输出：

True
False
None

容量有界LRU：最近访问不是普通插入顺序

LRU 淘汰最久未访问的项。OrderedDict.move_to_end 更新访问位置，popitem(last=False) 从头淘汰。

from collections import OrderedDict

class LRU:
    def __init__(self, capacity):
        if type(capacity) is not int or capacity < 1:
            raise ValueError("容量必须是正整数")
        self.capacity = capacity
        self.items = OrderedDict()

    def put(self, key, value):
        self.items[key] = value
        self.items.move_to_end(key)
        if len(self.items) > self.capacity:
            self.items.popitem(last=False)

    def get(self, key):
        if key not in self.items:
            return None
        self.items.move_to_end(key)
        return self.items[key]

cache = LRU(2)
cache.put("a", 1)
cache.put("b", 2)
print(cache.get("a"))
cache.put("c", 3)
print(list(cache.items))
print(cache.get("b"))

输出：

1
['a', 'c']
None

TTL失效：用可控时钟先验边界

TTL 表示存活时间，到达过期时刻就失效；下例传入固定时钟值观察边界。

class TTLCache:
    def __init__(self, ttl):
        if ttl <= 0:
            raise ValueError("TTL必须为正")
        self.ttl = ttl
        self.items = {}

    def put(self, key, value, now):
        self.items[key] = {"value": value, "expires": now + self.ttl}

    def get(self, key, now):
        item = self.items.get(key)
        if item is None:
            return None
        if now >= item["expires"]:
            self.items.pop(key)
            return None
        return item["value"]

cache = TTLCache(5)
cache.put("a", "result", 100)
print(cache.get("a", 104))
print(cache.get("a", 105))

输出：

result
None

预加载/复用：证明少构建，再测同条件耗时

把重复构建移到循环外，可复用同一对象；用 perf_counter 测量时间间隔。

from time import perf_counter

def build_index():
    return list(range(1000))

start = perf_counter()
old_results = []
for index in range(20):
    values = build_index()
    old_results.append(values[0])
old_ms = (perf_counter() - start) * 1000

start = perf_counter()
shared = build_index()
new_results = []
for index in range(20):
    new_results.append(shared[0])
new_ms = (perf_counter() - start) * 1000
print("构建次数", 20, "变为", 1)
print("输出保持", old_results == new_results)
print("玩具耗时ms", round(old_ms, 3), round(new_ms, 3))

预期结果：

构建次数 20 变为 1
输出保持 True
玩具耗时ms <本次旧耗时> <本次新耗时>

锁与复合状态：GIL不替你保证所有逻辑安全

线程共享对象；用 Lock 包住读、计算、写回的复合操作。

from threading import Lock
from concurrent.futures import ThreadPoolExecutor

class Counter:
    def __init__(self):
        self.value = 0
        self.lock = Lock()

    def add_once(self, unused):
        with self.lock:
            self.value = self.value + 1

counter = Counter()
with ThreadPoolExecutor(max_workers=4) as pool:
    list(pool.map(counter.add_once, range(20)))
print(counter.value)

输出：

20

批量保序：完成顺序与输入顺序不同

ThreadPoolExecutor.map 按输入顺序返回结果，即使任务完成顺序不同。

from concurrent.futures import ThreadPoolExecutor
from time import sleep

def work(number):
    sleep((3 - number) * 0.002)
    return {"index": number, "result": str(number)}

with ThreadPoolExecutor(max_workers=3) as pool:
    results = list(pool.map(work, [0, 1, 2]))
for item in results:
    print(item["index"], item["result"])

输出：

0 0
1 1
2 2

平均/P95/成功率：先手算，再自动算

平均值为总耗时除以数量；这里 P95 使用 nearest-rank：排序后取第 ceil(0.95N) 项。

import math

latencies = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
ordered = sorted(latencies)
average = sum(latencies) / len(latencies)
p95 = ordered[math.ceil(0.95 * len(ordered)) - 1]
statuses = [200, 200, 200, 200, 200, 200, 200, 200, 500, 500]
http_ok = 0
for status in statuses:
    if status == 200:
        http_ok = http_ok + 1
print("平均ms", average, "P95ms", p95)
print("HTTP成功", http_ok, "/", len(statuses))
print("人工业务正确", 5, "/", len(statuses))

输出：

平均ms 5.5 P95ms 10
HTTP成功 8 / 10
人工业务正确 5 / 10

峰值内存：只读自己的进程，单位说清楚

峰值内存与当前内存不同；Linux 的 ru_maxrss 以 KiB 计，Windows 的 PeakWorkingSet64 以字节计。

import json
import os
import subprocess
import sys

payload = bytearray(4 * 1024 * 1024)
if sys.platform == "linux":
    import resource
    peak_bytes = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss * 1024
elif sys.platform == "win32":
    command = "[Console]::OutputEncoding=[System.Text.UTF8Encoding]::new(); (Get-Process -Id " + str(os.getpid()) + ").PeakWorkingSet64 | ConvertTo-Json"
    result = subprocess.run(["powershell.exe", "-NoProfile", "-NonInteractive", "-Command", command], capture_output=True, text=True, encoding="utf-8", check=True)
    peak_bytes = int(json.loads(result.stdout))
else:
    raise RuntimeError("本例只核对Linux/Windows单位")
print("测量PID", os.getpid())
print("进程峰值MiB", round(peak_bytes / (1024 * 1024), 3))
print("大于零", peak_bytes > 0)

预期结果：

测量PID <本次PID>
进程峰值MiB <本次测量值>
大于零 True

量化进阶：数值误差与裸字节，不承诺业务保持

用 scale 把浮点数映射到有限整数，再乘 scale 还原；压缩会引入数值误差。

import struct

original = [-1.0, 0.25, 0.99]
largest = max(abs(value) for value in original)
scale = largest / 127 if largest != 0 else 1
codes = []
restored = []
errors = []
for value in original:
    code = max(-127, min(127, round(value / scale)))
    codes.append(code)
    back = code * scale
    restored.append(back)
    errors.append(abs(back - value))
float_bytes = struct.pack("fff", original[0], original[1], original[2])
int_bytes = struct.pack("bbb", codes[0], codes[1], codes[2])
print(codes)
print(len(float_bytes), len(int_bytes))
print("最大数值误差", round(max(errors), 4))

输出：

[-127, 32, 126]
12 3
最大数值误差 0.0021

第20节：JS/fetch、来源、配置追踪、manifest与故障回归

JavaScript 与 DOM

let 声明可重新赋值的变量，const 声明不可重新赋值的绑定。querySelector 找元素，事件处理函数响应点击，textContent 更新文字。

<!doctype html>
<html lang="zh-CN"><meta charset="utf-8"><title>JS第一步</title>
<p id="output">尚未更新</p>
<button id="button">加1</button>
<script>
let count = 0;
const output = document.querySelector("#output");
document.querySelector("#button").addEventListener("click", () => {
  count = count + 1;
  output.textContent = "次数=" + count;
});
</script></html>

预期结果：

点击一次显示“次数=1”，再点显示“次数=2”。

配置外置：保存文件与让服务读取是两个动作

配置放入 JSON 文件，启动时读取。Path(__file__).resolve().parent 把路径基准固定到脚本所在目录。

from pathlib import Path
import json

root = Path(__file__).resolve().parent
config = {"version": "ui-demo-v1", "prefix": "[MOCK前端夹具]收到："}
(root / "config20.json").write_text(json.dumps(config, ensure_ascii=False, indent=2), encoding="utf-8")
print("已生成自己的config20.json")

输出：

已生成自己的config20.json

同源 API、响应与追踪

同源页面和 API 共享协议、主机与端口。request_id 关联请求、响应和 JSONL 日志，config_version 标识当前配置。下面以固定数据展示日志字段。

import json
row = {"request_id": "demo-1", "status_code": 200,
       "config_version": "ui-demo-v1", "mode": "mock-fixture"}
line = json.dumps(row, ensure_ascii=False)
record = json.loads(line)
print(record["request_id"], record["status_code"], record["config_version"])

输出：

demo-1 200 ui-demo-v1

fetch 请求闭环

提交前清空旧答案、来源和错误；请求中显示等待状态。fetch 得到响应后检查 response.ok，finally 恢复按钮。以下是接入页面元素后的处理片段。

async function submit(question) {
  answer.textContent = "";
  sources.textContent = "来源：无";
  errorBox.textContent = "";
  button.disabled = true;
  loading.hidden = false;
  try {
    const response = await fetch("/demo_answer", {
      method: "POST",
      headers: {"Content-Type": "application/json"},
      body: JSON.stringify({question})
    });
    const data = await response.json();
    if (!response.ok) throw new Error(data.message);
    answer.textContent = data.answer;
    sources.textContent = data.sources.length
      ? "来源：" + data.sources.map(s => s.title).join("；")
      : "来源：无";
  } catch (error) {
    errorBox.textContent = error.message;
  } finally {
    button.disabled = false;
    loading.hidden = true;
  }
}

预期结果：

成功时更新答案和来源；空来源显示“来源：无”；错误时显示消息；结束后按钮恢复。

按 request_id 查 JSONL 日志

逐行解析记录，再按请求 ID 筛选，就能对应某次响应的状态和配置版本。

import json
lines = [
    '{"request_id":"a","status_code":200,"config_version":"v1"}',
    '{"request_id":"b","status_code":500,"config_version":"v1"}'
]
for line in lines:
    record = json.loads(line)
    if record["request_id"] == "b":
        print(record["status_code"], record["config_version"])

输出：

500 v1

manifest 与分层

manifest 声明入口和所需文件；用路径检查确认它们存在。服务代码可按 model、retrieval、service 分层，路由组织请求与响应。

from pathlib import Path

root = Path("demo_package")
root.mkdir(exist_ok=True)
(root / "app.py").write_text("# demo\n", encoding="utf-8")
manifest = {"entrypoint": "app.py", "required_files": ["app.py"]}
for relative in manifest["required_files"]:
    path = (root / relative).resolve()
    assert path.is_relative_to(root.resolve()) and path.is_file()
print(manifest["entrypoint"])
print("文件检查通过")

输出：

app.py
文件检查通过

按清单打包

ZipFile 创建压缩包，arcname 指定包内名称；按清单打包可控制交付内容。

from pathlib import Path
import zipfile

root = Path("package_example")
root.mkdir(exist_ok=True)
(root / "app.py").write_text("# demo\n", encoding="utf-8")
(root / "config.json").write_text('{"version":"v1"}', encoding="utf-8")
names = ["app.py", "config.json"]
with zipfile.ZipFile("demo.zip", "w", zipfile.ZIP_DEFLATED) as archive:
    for name in names:
        archive.write(root / name, arcname=name)
with zipfile.ZipFile("demo.zip") as archive:
    print(archive.namelist())

输出：

['app.py', 'config.json']