Python 专题：AI 胶水 · 脚本自动化 · 大模型集成

写在最前面的边界声明：本模块不教你把 Python 学成第二主线。07 报告的语言版图结论在这里原样执行。

1.1 边界三条（闭关期间刻在脑子里）

• 不部署 Python 在线服务：需要长期运行、要扛并发、要鉴权要数据库的服务，全部交给 Go 工作台（模块 11）。Python 脚本是"跑完即走"的胶水。
• 每周 1 个时间盒：按 08 总纲路径 B/C，Python 从第 3 个月起每周一个时间盒（约半天）做脚本与 LLM 练习；主线（Go/Vue3/数据库）时间被挤占即暂停 Python。
• AI 五步工作流同样适用：写 Python 脚本也走 spec 先行 → 失败测试先行 → 小步实现 → 双会话评审 → 人掌 merge（08 总纲 §6）。胶水脚本不代表可以乱写。

安装步骤本身在 09 模块 §1.6（Windows 安装 Python 3.14.8）。本节只讲三件本模块独有的事：装 uv、用 uv 建项目、配国内镜像。

2.1 为什么用 uv 而不是 pip/venv

2.2 照做：安装 uv 并建好第一个胶水项目

# 1) 安装 uv（PowerShell 一次性执行；脚本来自 astral 官方发布页，本模块核验版本 0.13.0）
powershell -ExecutionPolicy Bypass -c "irm https://releases.astral.sh/github/uv/releases/download/0.13.0/uv-installer.ps1 | iex"

# 2) 关闭当前终端，新开一个 PowerShell，确认版本
uv --version
# 期望输出: uv 0.13.0

# 3) 国内镜像（每个新终端会话临时生效；setx UV_DEFAULT_INDEX 可写入用户环境变量永久生效）
$env:UV_DEFAULT_INDEX = "https://pypi.tuna.tsinghua.edu.cn/simple"

# 4) 建项目：uv 自动创建 .venv 目录并锁定 Python 3.14.8
mkdir ai-glue; cd ai-glue
uv init --name ai-glue --python 3.14.8
uv add httpx                  # 加依赖：自动写 pyproject.toml + uv.lock
uv run python --version       # 期望输出: Python 3.14.8

venv 概念一句话：venv 就是项目目录里一个隔离的依赖文件夹（.venv/），让这个项目装的包不污染系统 Python。用 uv 后你几乎不用手写 python -m venv、不用记 activate 命令——uv 全自动接管，跑脚本统一用 uv run python xxx.py。

有多年编程经验的人不需要教科书。下表是概念对照，下面是一段把要点全装进去的代码，敲一遍即上手。

3.1 概念对照表

3.2 一段代码装下全部要点

# 类型注解只是提示：Python 是动态类型，运行时才检查
def summarize(items: list[float], label: str = "总计") -> str:
    total = sum(items)
    avg = total / len(items) if items else 0.0      # 三元表达式一行搞定
    return f"{label}: {total:.1f}, 均值 {avg:.1f} 元"   # f-string = JS 模板字符串

class Report:
    def __init__(self, title: str):      # 构造器；self = 显式 this（PHP 是 $this）
        self.title = title
    def render(self, items):             # 方法第一个参数永远是 self
        return f"<{self.title}> {summarize(items)}"

data = [12.5, 8.0, 30.2]
print(Report("7月").render(data))        # 输出: <7月> 总计: 50.7, 均值 16.9 元

# 推导式：PHP/Go/JS 都没有的杀手特性
big = [x for x in data if x >= 10]       # 列表推导式 → [12.5, 30.2]
doubled = {x: x*2 for x in data}         # 字典推导式

try:
    _ = 1 / 0
except ZeroDivisionError as e:          # 异常：except 对应 catch
    print("捕获:", e)

# 文件读写：永远显式 encoding（Windows 下不写会吃 GBK 报错，见 §8）
with open("note.txt", "w", encoding="utf-8") as f:
    f.write("hello")

3.3 装饰器与文件 IO（够用到你写脚本为止）

import time
def timer(fn):                          # 装饰器 = 接收函数、返回包装函数的函数
    def wrap(*args, **kwargs):
        t0 = time.perf_counter()
        r = fn(*args, **kwargs)
        print(f"{fn.__name__} 耗时 {time.perf_counter()-t0:.3f}s")
        return r
    return wrap

@timer                                  # = work = timer(work)，PHP/Go 无此语法
def work():
    time.sleep(0.1)

给有经验者的三句提醒：①缩进是语法，Tab/空格混用必报 IndentationError；②没有 Go 的编译期保证，类型注解写错不报错——所以"小步实现 + 真跑通"比写注解重要；③标准库极厚（pathlib/json/csv/subprocess/httpx 之外的一切），胶水脚本优先用标准库。

以下脚本全部整文件可复制，在 ai-glue 项目里用 uv run python 文件名.py 运行。这是本模块每周时间盒的练习材料。

4.1 第一类：文件与数据处理（csv/json/Excel 简述）

脚本①：批量重命名 rename_dated.py

"""批量重命名：把目录下 *.txt 重编号为 doc-001.txt …（对照 PHP 里的 rename()）"""
import sys
from pathlib import Path          # pathlib = Go 的 os/filepath + filepath.Walk 的合体

def main(folder: str):
    files = sorted(Path(folder).glob("*.txt"))
    if not files:
        print("没找到 .txt 文件")
        return
    for i, p in enumerate(files, start=1):
        new_name = f"doc-{i:03d}{p.suffix}"
        p.rename(p.with_name(new_name))      # with_name 只换文件名不改目录
        print(f"{p.name} -> {new_name}")

if __name__ == "__main__":                  # = Go 的 main() 入口惯例
    main(sys.argv[1] if len(sys.argv) > 1 else ".")

脚本②：日志聚合 log_agg.py

"""日志聚合：扫描目录下 *.log，按级别统计 ERROR/WARN，输出 report.txt"""
from pathlib import Path
from collections import Counter

def main(folder: str):
    stats = Counter()
    samples: list[str] = []
    for p in Path(folder).glob("*.log"):
        text = p.read_text(encoding="utf-8", errors="replace")  # errors=replace 防坏字节崩
        for line in text.splitlines():
            for level in ("ERROR", "WARN"):
                if level in line:
                    stats[level] += 1
                    if len(samples) < 10:
                        samples.append(f"[{p.name}] {line.strip()[:160]}")
    out = ["# 日志聚合报告",
           f"ERROR={stats['ERROR']}  WARN={stats['WARN']}", ""]
    out += samples or ["（无 ERROR/WARN 行）"]
    Path("report.txt").write_text("\n".join(out), encoding="utf-8")
    print("\n".join(out))

if __name__ == "__main__":
    import sys
    main(sys.argv[1] if len(sys.argv) > 1 else "logs")

4.2 第二类：系统自动化（子进程 / 定时 / 与 Go 互操作）

脚本③：subprocess 调 Go 二进制（互操作范式）

"""系统自动化：调用 Go 编译出的 CLI 二进制，解析其文本输出"""
import subprocess, sys

def run(cmd: list[str]) -> str:
    r = subprocess.run(cmd, capture_output=True, text=True,
                       encoding="utf-8", errors="replace", timeout=60)
    if r.returncode != 0:
        sys.exit(f"命令失败({r.returncode}): {r.stderr.strip()[:300]}")
    return r.stdout

# 示例：调用工作台项目里交叉编译出的 Go 二进制（路径按你实际位置改）
print(run(["..\\workbench\\bin\\workbench.exe", "--version"]))

# 更常见的用法：批量跑一个处理命令，把结果攒起来
for line in run(["git", "log", "--oneline", "-5"]).splitlines():
    print("git:", line)

脚本④：等待工作台就绪并自动建任务（HTTP 调工作台 API）

"""系统自动化：轮询工作台 /healthz，就绪后登录并自动建一条任务（端点严格按契约）"""
import time, sys
import httpx

BASE = "http://localhost:8080"

def wait_ready(timeout_s: int = 30):
    t0 = time.time()
    while time.time() - t0 < timeout_s:
        try:
            r = httpx.get(f"{BASE}/healthz", timeout=2.0)
            if r.status_code == 200 and r.json().get("status") == "ok":
                print("工作台已就绪:", r.json())
                return
        except httpx.TransportError:
            pass                        # 还没起来，继续等
        print("等待工作台启动…")
        time.sleep(2)
    sys.exit("超时：工作台 30 秒内未就绪")

def login(username: str, password: str) -> str:
    r = httpx.post(f"{BASE}/api/v1/auth/login",
                   json={"username": username, "password": password}, timeout=5.0)
    body = r.json()                 # 统一响应: {"code":0,"data":{...},"msg":""}
    if body.get("code") != 0:
        sys.exit(f"登录失败: {body.get('msg')}")
    return body["data"]["token"]

def create_task(token: str, title: str, note: str):
    r = httpx.post(f"{BASE}/api/v1/tasks",
                   headers={"Authorization": f"Bearer {token}"},
                   json={"title": title, "note": note, "done": False}, timeout=5.0)
    print("建任务响应:", r.json())

if __name__ == "__main__":
    wait_ready()
    tok = login("admin", "admin123")   # 工作台首次启动时建的管理员账号
    create_task(tok, "来自 Python 胶水的任务", "由 wait_workbench.py 自动创建")

4.3 第三类：网络请求（requests vs httpx 选型）

脚本⑤：httpx 基础（带超时/鉴权/错误分类的 GET）

"""httpx 基础范式：所有网络请求照这个模板改"""
import httpx, sys

def list_tasks(token: str):
    try:
        r = httpx.get("http://localhost:8080/api/v1/tasks",
                      headers={"Authorization": f"Bearer {token}"},
                      timeout=(5.0, 30.0))       # (连接超时, 读取超时)，必须显式给
    except httpx.TimeoutException:
        sys.exit("请求超时：检查工作台是否在 8080 端口运行")
    if r.status_code == 401:
        sys.exit("401：token 过期，重新登录拿 token")
    body = r.json()
    tasks = body["data"]
    for t in tasks:
        mark = "x" if t["done"] else " "
        print(f"[{mark}] {t['title']}")

范式记住三条：①timeout=(连接秒, 读取秒)永远显式写，裸请求会挂死；②先判 status_code（401/404/500）再 .json()；③解析 JSON 前先 print(r.status_code, r.text[:200]) 是排错起点（见 §8 报错 6）。

这就是 Python 卫星存在的理由。国产大模型（DeepSeek、通义千问、Kimi 等）全部兼容 OpenAI 的 HTTP 协议——一套代码换三个环境变量即可切换厂商。

5.1 协议与端点（一张表）

脚本⑥：最小 LLM 调用 llm_chat.py（非流式）

"""最小 LLM 调用：OpenAI 兼容协议，换环境变量即换厂商"""
import os, sys, httpx

BASE = os.environ["LLM_BASE_URL"]      # e.g. https://api.deepseek.com
KEY  = os.environ["LLM_API_KEY"]
MODEL = os.environ.get("LLM_MODEL", "deepseek-chat")

def chat(messages: list[dict]) -> str:
    r = httpx.post(
        f"{BASE}/chat/completions",
        headers={"Authorization": f"Bearer {KEY}"},
        json={"model": MODEL,
              "messages": messages,
              "temperature": 0.3},
        timeout=(10.0, 120.0),          # 大模型响应慢，读取超时给足
    )
    if r.status_code != 200:
        sys.exit(f"LLM 错误 {r.status_code}: {r.text[:300]}")
    return r.json()["choices"][0]["message"]["content"]

if __name__ == "__main__":
    print(chat([{"role": "user", "content": "用一句话解释什么是 RAG"}]))

# PowerShell 里配置环境变量后运行（key 在各厂商控制台申请）
$env:LLM_BASE_URL = "https://api.deepseek.com"
$env:LLM_API_KEY  = "sk-你的key"
$env:LLM_MODEL    = "deepseek-chat"
uv run python llm_chat.py

脚本⑦：流式输出 llm_stream.py（逐 token 打印）

"""流式输出：响应是 SSE 文本流，每行 data: {...}，[DONE] 结束"""
import os, json, httpx

BASE, KEY = os.environ["LLM_BASE_URL"], os.environ["LLM_API_KEY"]
MODEL = os.environ.get("LLM_MODEL", "deepseek-chat")

with httpx.stream("POST", f"{BASE}/chat/completions",
                  headers={"Authorization": f"Bearer {KEY}"},
                  json={"model": MODEL,
                        "messages": [{"role": "user",
                                      "content": "写三句产品上线前 Checklist"}],
                        "stream": True},
                  timeout=(10.0, 120.0)) as r:
    for line in r.iter_lines():
        if not line or not line.startswith("data:"):
            continue
        payload = line[5:].strip()
        if payload == "[DONE]":
            break
        delta = json.loads(payload)["choices"][0]["delta"].get("content", "")
        print(delta, end="", flush=True)     # 不换行、不缓冲，逐字刷出

5.2 工具调用（函数调用）简介——知道有这回事即可

OpenAI 兼容协议的工具调用是三段式循环，本模块只需要知道形状，真要用时照此扩展：

1. 请求里多带一个 tools 字段：描述你本地有哪些函数（名字、参数 JSON Schema）；
2. 模型不直接回答，返回 finish_reason == "tool_calls"，告诉你"该调哪个函数、传什么参数"；
3. 你在本地真的执行这个函数，把结果以 role: "tool" 的消息追加进 messages，再发一次请求——模型拿到结果后继续生成回答。

边界提醒：工具调用是"模型出主意、你执行"，不是模型真的能跑你的代码。安全上执行前必须人工或白名单确认——这和 AI 五步工作流里"人掌 merge"是同一个哲学。

5.3 验收示例：读文件 → LLM 总结 → 写回工作台

下面这个脚本把 §4 的工作台契约（登录拿 JWT → POST /api/v1/tasks）和本节的 LLM 调用焊在一起，就是你两个时间盒要跑通的东西。端点严格按 workbench 契约，不新增任何端点。

"""summarize_to_workbench.py：读本地文件 → LLM 总结 → 写回工作台任务列表"""
import os, json, httpx

WB = "http://localhost:8080"
LLM_BASE  = os.environ["LLM_BASE_URL"]
LLM_KEY   = os.environ["LLM_API_KEY"]
LLM_MODEL = os.environ.get("LLM_MODEL", "deepseek-chat")

def llm_summarize(text: str) -> str:
    r = httpx.post(f"{LLM_BASE}/chat/completions",
                   headers={"Authorization": f"Bearer {LLM_KEY}"},
                   json={"model": LLM_MODEL,
                         "messages": [
                             {"role": "system",
                              "content": "你是摘要助手：用不超过 150 字中文总结用户给的文本。"},
                             {"role": "user", "content": text}],
                         "temperature": 0.2},
                   timeout=(10.0, 120.0))
    r.raise_for_status()
    return r.json()["choices"][0]["message"]["content"]

def wb_login() -> str:
    r = httpx.post(f"{WB}/api/v1/auth/login",
                   json={"username": "admin", "password": "admin123"}, timeout=5.0)
    return r.json()["data"]["token"]

def wb_add_task(token: str, title: str, note: str):
    r = httpx.post(f"{WB}/api/v1/tasks",
                   headers={"Authorization": f"Bearer {token}"},
                   json={"title": title, "note": note, "done": False}, timeout=5.0)
    body = r.json()
    assert body["code"] == 0, body          # 统一响应：code=0 才算成功
    print("已写回工作台:", json.dumps(body["data"], ensure_ascii=False))

if __name__ == "__main__":
    src = open("note.md", encoding="utf-8").read()   # 换成你自己的 md/txt
    summary = llm_summarize(src)
    print("LLM 摘要:\n" + summary)
    tok = wb_login()
    wb_add_task(tok, "AI 摘要: note.md", summary)
    # 打开工作台前端任务列表，应能看到这条新任务

默认不部署 Python 服务。只有当某个 AI 功能必须暴露成 HTTP 接口（给前端临时演示、给内网其他工具调用）时，才用 FastAPI 包一层。这是例外，不是常态。

# 先装依赖（FastAPI 0.143.0，本模块自核验；[standard] 含 uvicorn 服务器）
uv add "fastapi[standard]"

# main.py：整个文件就这些
from fastapi import FastAPI
from pydantic import BaseModel
import os, httpx

app = FastAPI(title="ai-glue-service")
LLM_BASE, LLM_KEY = os.environ["LLM_BASE_URL"], os.environ["LLM_API_KEY"]

class Req(BaseModel):
    text: str

@app.get("/healthz")
def healthz():
    return {"status": "ok"}

@app.post("/api/v1/ai/summarize")
def summarize(req: Req):
    r = httpx.post(f"{LLM_BASE}/chat/completions",
                   headers={"Authorization": f"Bearer {LLM_KEY}"},
                   json={"model": os.environ.get("LLM_MODEL", "deepseek-chat"),
                         "messages": [{"role": "user", "content": "用 100 字总结：" + req.text}]},
                   timeout=(10.0, 120.0))
    return {"summary": r.json()["choices"][0]["message"]["content"]}

# 启动（端口 8000，避开工作台的 8080）
# uv run uvicorn main:app --port 8000

本节与 08 总纲 §7 版本纪律规则 3 严格对齐：锁文件是唯一事实来源。Python 生态的历史包袱是 requirements.txt，必须把它的位置摆正。

7.1 日常命令三件套

uv add httpx              # 加依赖：写 pyproject.toml，解析后写入 uv.lock
uv remove httpx           # 删依赖：同步更新两个文件
uv sync                   # 按 uv.lock 精确还原 .venv（换机器/CI 第一步）
uv run python xxx.py      # 永远用 uv run 跑，不用裸 python（见 §8 报错 9）

与 AI 协作时（对齐 08 §6.2）：每次让 AI 写 Python 脚本，把 uv.lock 内容喂给它，要求"只准用锁文件里已有的包与版本"；AI 说"建议升级到 XX"= 默认拒绝（除非命中升级三条件：安全通告 / 解决当前真实问题 / 全新项目）。

闭关卡住时先查这张表。报错原文要完整复制，不要只复制半句。

完成下面每一项才算通关。建议排在两个时间盒内：第一个时间盒环境+脚本三件，第二个时间盒 LLM 集成。

过了验收 6，本模块即完成。之后 Python 回到"每周 1 个时间盒"的节奏：从你的真实工作里挑一个重复劳动（周报汇总、日志巡检、文档摘要、批量数据清洗），用五步工作流做成脚本沉淀进四件套 snippets——这就是卫星的全部使命。

Python 本体以 09 版本总表为唯一事实来源；Python 生态四个工具由本模块生产时（2026-10-14）访问官方源自核验，记录如下。

来源表

（编者整理：本模块语法对照表、脚本范式、报错表为基于 07/08 系列口径与编程常识的执行化设计，无外部独立来源。）