书架 · Python 学习系列 · 08 · 实战项目与学习资源下一章：09 LangChain + LangGraph →

08 · 实战项目与学习资源

本章目标：三个由小到大的综合项目，把前 7 章串成肌肉记忆；附全量官方学习资源。

前置：学完 04；项目二需要 05、07。

📌 项目中所有接入参数沿用第 04 章核实过的官方配置（访问日期 2026-09-05）。

项目一：命令行 AI 问答工具（cli-chat）

练到什么：SDK 基础调用、流式输出、多轮历史、历史持久化、异常兜底。约 80 行。

cli-chat/
├── pyproject.toml
└── cli_chat.py

"""cli_chat.py —— 流式多轮对话，历史存 JSON，/reset 清空，/save 导出
uv add openai
"""
import json
import os
from pathlib import Path

from openai import OpenAI, APIError

BASE_URL = "https://api.deepseek.com"          # 第 04 章核实
MODEL = "deepseek-v4-pro"
HISTORY_FILE = Path("history.json")
MAX_TURNS = 20                                  # 滑动窗口：保留最近 20 条

client = OpenAI(api_key=os.environ["DEEPSEEK_API_KEY"], base_url=BASE_URL)

def load_history() -> list[dict]:
    if HISTORY_FILE.exists():
        return json.loads(HISTORY_FILE.read_text(encoding="utf-8"))
    return []

def save_history(history: list[dict]) -> None:
    HISTORY_FILE.write_text(json.dumps(history, ensure_ascii=False, indent=2),
                            encoding="utf-8")

def chat_stream(messages: list[dict]) -> str:
    """流式调用，逐字打印，返回完整答案（第 04 章第 4 节）"""
    stream = client.chat.completions.create(
        model=MODEL, messages=messages, stream=True,
        stream_options={"include_usage": True},
    )
    parts, usage = [], None
    for chunk in stream:
        if chunk.usage:
            usage = chunk.usage
        if chunk.choices and chunk.choices[0].delta.content:
            parts.append(chunk.choices[0].delta.content)
            print(chunk.choices[0].delta.content, end="", flush=True)
    if usage:
        print(f"\n[token] {usage.total_tokens}")
    return "".join(parts)

def main() -> None:
    history = load_history()
    print(f"历史 {len(history)} 条。命令：/reset 清空 /save 导出 /exit 退出")
    while True:
        try:
            user = input("\n你: ").strip()
        except (EOFError, KeyboardInterrupt):
            break
        if not user:
            continue
        if user == "/exit":
            break
        if user == "/reset":
            history.clear(); save_history(history)
            print("（已清空）"); continue
        if user == "/save":
            print(f"已导出 {len(history)} 条 -> {HISTORY_FILE.resolve()}"); continue

        history.append({"role": "user", "content": user})
        messages = [{"role": "system", "content": "你是简洁的中文技术助手"},
                    *history[-MAX_TURNS:]]
        print("助手: ", end="")
        try:
            answer = chat_stream(messages)
        except APIError as e:                       # fail fast，给出可读信息
            print(f"\n[调用失败] {e!r}")
            history.pop()                            # 撤回本轮 user，避免脏历史
            continue
        history.append({"role": "assistant", "content": answer})
        save_history(history)

if __name__ == "__main__":
    main()

扩展练习：① --model 参数切换 flash/pro（省钱）；② 摘要压缩：超 20 条时用 deepseek-v4-flash 压缩旧历史；③ 支持 /retry 重发上一问。

项目二：RAG 知识库问答 API（rag-api）

练到什么：FastAPI 全流程 + embedding 检索 + SSE 透传 + 依赖注入 + 测试。这是三个项目里最"生产"的一个。

rag-api/
├── pyproject.toml        # uv add "fastapi[standard]" openai numpy
├── ingest.py             # 离线：知识库向量化
└── app.py                # 在线：检索 + 问答 API

"""ingest.py —— 把 docs/ 目录下所有 .md 切块并向量化，存 index.npz"""
from pathlib import Path
import numpy as np
from openai import OpenAI
import os

client = OpenAI(api_key=os.environ["DEEPSEEK_API_KEY"], base_url="https://api.deepseek.com")
EMBED_MODEL = "deepseek-v4-flash"          # embedding 模型名以官方文档为准（第 04 章出处）
DOCS_DIR, CHUNK_SIZE = Path("docs"), 500   # 按字符切块（简单版；生产按 token/段落）

chunks: list[str] = []
for md in DOCS_DIR.glob("*.md"):
    text = md.read_text(encoding="utf-8")
    chunks += [text[i:i + CHUNK_SIZE] for i in range(0, len(text), CHUNK_SIZE)]

vectors = []
for i in range(0, len(chunks), 64):        # 分批向量化
    batch = client.embeddings.create(model=EMBED_MODEL, input=chunks[i:i + 64])
    vectors += [d.embedding for d in batch.data]

np.savez("index.npz",
         matrix=np.array(vectors, dtype=np.float32),
         chunks=np.array(chunks, dtype=object))
print(f"已索引 {len(chunks)} 块 -> index.npz")

"""app.py —— POST /ask 普通回答；POST /ask/stream SSE 流式；GET /health"""
import json
import os
from collections.abc import AsyncIterator

import numpy as np
from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
from pydantic import BaseModel, Field

app = FastAPI(title="RAG 知识库问答")
data = np.load("index.npz", allow_pickle=True)
MATRIX: np.ndarray = data["matrix"]
CHUNKS: list[str] = data["chunks"].tolist()

async def get_llm() -> AsyncOpenAI:                      # 依赖注入：全局单例客户端
    return AsyncOpenAI(api_key=os.environ["DEEPSEEK_API_KEY"],
                       base_url="https://api.deepseek.com")

async def verify_token(authorization: str = Header()) -> str:
    if authorization != f"Bearer {os.environ.get('API_TOKEN', 'dev')}":
        raise HTTPException(401, "invalid token")
    return authorization

def retrieve(query_vec: list[float], top_k: int = 3) -> list[str]:   # 第 04 章检索器
    q = np.array(query_vec)
    sims = MATRIX @ q / (np.linalg.norm(MATRIX, axis=1) * np.linalg.norm(q))
    return [CHUNKS[i] for i in np.argsort(sims)[::-1][:top_k]]

class AskRequest(BaseModel):
    question: str = Field(min_length=1, max_length=2000)
    top_k: int = Field(default=3, ge=1, le=10)

def build_messages(question: str, hits: list[str]) -> list[dict]:
    context = "\n\n".join(f"[资料{i}] {h}" for i, h in enumerate(hits, 1))
    return [
        {"role": "system",
         "content": "仅依据资料回答；资料不足就直说不知道，并标注引用了哪几条资料。\n\n" + context},
        {"role": "user", "content": question},
    ]

@app.get("/health")
def health() -> dict:
    return {"ok": True, "chunks": len(CHUNKS)}

@app.post("/ask")
async def ask(req: AskRequest, llm: AsyncOpenAI = Depends(get_llm),
              _: str = Depends(verify_token)):
    qvec = (await llm.embeddings.create(model="deepseek-v4-flash",
                                        input=[req.question])).data[0].embedding
    hits = retrieve(qvec, req.top_k)
    resp = await llm.chat.completions.create(
        model="deepseek-v4-pro", messages=build_messages(req.question, hits))
    return {"answer": resp.choices[0].message.content,
            "usage": resp.usage.total_tokens}

@app.post("/ask/stream")
async def ask_stream(req: AskRequest, llm: AsyncOpenAI = Depends(get_llm),
                     _: str = Depends(verify_token)) -> StreamingResponse:
    qvec = (await llm.embeddings.create(model="deepseek-v4-flash",
                                        input=[req.question])).data[0].embedding
    hits = retrieve(qvec, req.top_k)

    async def gen() -> AsyncIterator[str]:
        yield f"data: {json.dumps({'refs': len(hits)}, ensure_ascii=False)}\n\n"
        stream = await llm.chat.completions.create(
            model="deepseek-v4-pro", messages=build_messages(req.question, hits),
            stream=True)
        async for chunk in stream:
            if chunk.choices and chunk.choices[0].delta.content:
                d = chunk.choices[0].delta.content
                yield f"data: {json.dumps({'v': d}, ensure_ascii=False)}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(gen(), media_type="text/event-stream")

验证：

uv run fastapi dev app.py
curl -N -X POST http://127.0.0.1:8000/ask/stream -H "Authorization: Bearer dev" ^
     -H "Content-Type: application/json" -d "{\"question\": \"公司年假制度是什么\"}"

扩展练习：① 向量库换 Qdrant（docker 起一个）；② 加 /upload 接口在线灌文档；③ 给 retrieve 写单测（造假向量即可，不花钱）。

项目三：最小 Agent + MCP 工具（参考实现升级）

第 05 章的 100 行 Agent 就是项目三的 v1。在其上做两个升级，各约 30 行：

升级 A：接入 MCP 工具（把第 05 章 6.2 的 server.py 工具挂进 Agent）：

# 在 minimal_agent.py 基础上：启动时从 MCP server 拉工具清单，转成 TOOLS
import asyncio
from mcp import Client

async def import_mcp_tools(url: str = "http://localhost:8000/mcp") -> None:
    async with Client(url) as client:
        tools = await client.list_tools()
        for t in tools:
            TOOL_REGISTRY[t.name] = {
                "fn": (lambda tool_name: lambda **kw: asyncio.run(
                    _call_remote(tool_name, kw)))(t.name),   # 远程代理执行
                "schema": {"type": "function",
                           "function": {"name": t.name,
                                        "description": t.description or "",
                                        "parameters": t.inputSchema}},
            }
            print(f"[MCP] 已挂载工具: {t.name}")

async def _call_remote(name: str, kwargs: dict) -> str:
    async with Client("http://localhost:8000/mcp") as client:
        result = await client.call_tool(name, kwargs)
        return str(result.structured_content or result.content)

在 main() 开头 asyncio.run(import_mcp_tools())，Agent 立刻获得远程调用能力——这就是 ZCode/Claude Code 连接 MCP 的最小复刻（出处与协议版本见第 05 章文末）。

升级 B：敏感操作人工确认：在 run_agent 执行工具前，若工具名在 CONFIRM_REQUIRED = {"delete_file", ...}，input("确认执行 y/n? ") 非 y 则回填 "用户拒绝执行"。

学习资源（按优先级）

官方文档（永远的第一手）

经典书（按阶段）

1. 入门后巩固：《Python Crash Course》/《Python 编程：从入门到实践》
2. 进阶必读：《Fluent Python》（流畅的 Python，Luciano Ramalho，人民邮电有中文版）——讲透 Python 数据模型与惯用法，Java 老兵读它事半功倍
3. AI 工程：Anthropic 工程博客系列（Building Effective Agents 等）+ 各 SDK 官方仓库的 examples 目录

交互式练习

• Python 官方交互教程列表（官方 BeginnersGuide 收录）
• Real Python（realpython.com）：高质量教程站，大量免费
• Exercism Python track：导师制刷题

本套资料内链

• 01-Python基础 · 02-Python进阶 · 03-工程化与异步 · 04-LLM应用开发 · 05-Agent开发 · 06-数据科学与ML · 07-Web服务-FastAPI
• 🔥 react-deep-dive.html（浏览器打开）

⬅️ 返回目录