← 学习路线
AI 学习

FastAPI 到 Memory 多轮对话 · 完整笔记

从接口起步,逐步接入模型、环境配置、会话 ID 和内存历史;保留完整代码、排错和附录。

康雨 · 发布 · 约 61 分钟

阅读基础:按章节逐步阅读;示例环境见正文

读完可以:查阅原始解释、完整代码和排错上下文

完整笔记:约 30.5 千字与代码字符 · 207 个代码/流程块。可按目录分次查阅;内容来自现有笔记,不补写未记录的经历。

本文目录

完整学习档案:保留原有章节、代码、失败示例、排错过程和练习。代码反映记录时的环境,不代表每段都是最终版本,也不代表本次发布重新运行了全部示例。请在练习项目中使用自己的环境配置,勿填入生产密钥。

学习阶段:FastAPI → Uvicorn → HTTP API → Pydantic → 真实 LLM 调用 → 配置与错误处理 → messages 多轮对话 → conversation_id → 服务器端 Memory

这份笔记不是知识点摘要,而是一次完整的学习复盘。重点记录:为什么需要这个设计、数据怎样流动、代码为什么这样写、实际遇到了什么问题、问题是怎样定位的,以及这些知识将怎样继续演进成真正的 AI Agent。


一、这一阶段在整条 Agent 路线中的位置

在进入 FastAPI 之前,我已经完成了 Python 最少必要基础,例如:

text
变量、字符串、数字、布尔值
list、dict、if、for
函数、参数、默认参数、return
import、module、package
pip、venv、requirements.txt
文件、JSON、异常、环境变量
HTTP、requests
class、object、Type Hint
async、await、Event Loop

这些知识不是互相孤立的。到了 FastAPI 项目,它们开始第一次在同一个程序里汇合:

text
Python 基础
    ↓
Python 工程结构
    ↓
HTTP / API
    ↓
FastAPI / Uvicorn
    ↓
真实 LLM API
    ↓
多轮对话 messages
    ↓
服务器端 Memory

这一阶段最重要的变化是:

text
以前:我在本地运行一个 Python 程序

现在:我让其他客户端通过 HTTP 调用我的 Python 能力

再进一步:

text
以前:每次只把一个问题发给模型

现在:服务器能根据 conversation_id 找回历史,形成多轮上下文

这意味着项目已经从“调用一次模型的脚本”,发展成了一个最小可用的聊天后端。


二、为什么 AI 应用需要后端服务

2.1 一个普通 Python 函数的边界

例如:

python
def add(a, b):
    return a + b

这个函数只能在当前 Python 程序中被调用。另一台电脑、一个网页或手机 App,并不能仅仅因为这个函数存在,就直接通过网络使用它。

普通函数缺少几样东西:

  • 没有网络监听能力;
  • 没有公开的 URL;
  • 不理解 HTTP Request;
  • 不会生成 HTTP Response;
  • 不知道怎样把 JSON 转成 Python 数据;
  • 不知道怎样把 Python 返回值转成 JSON。

因此,需要在 Python 函数外面增加一层 Web API:

text
客户端
  ↓ HTTP Request
Web API
  ↓ 调用
Python 函数
  ↓ 返回结果
Web API
  ↓ HTTP Response
客户端

FastAPI 的价值,就是帮助我们构建这层 Web API。

2.2 “能调用模型”不等于“已经做成 AI 应用”

最简单的大模型脚本可能是:

text
问题
 ↓
调用模型
 ↓
打印回答

但真实产品往往是:

text
用户
 ↓
网页 / 手机 / 其他客户端
 ↓
HTTP API
 ↓
身份、参数和业务规则
 ↓
对话历史 / 数据库 / 工具
 ↓
大模型
 ↓
整理结果
 ↓
HTTP Response
 ↓
用户界面

所以,大模型只是系统中的一个能力组件。FastAPI 后端负责把客户端、业务逻辑、模型、Memory、数据库和工具组织起来。

2.3 在 Agent 架构中的位置

以后完整的 Agent 服务会更像:

text
Client
  ↓
FastAPI
  ↓
Agent Loop
  ├── LLM
  ├── Memory
  ├── Tools
  └── RAG / Database
  ↓
Response

本阶段还没有完成真正的 Agent Loop,但已经搭好了它未来对外提供服务的外壳。


三、HTTP、Client 与 Server

3.1 Client 是谁

Client 是主动发起请求的一方。它可以是:

  • 浏览器;
  • 网页前端;
  • 手机 App;
  • Python 的 requests;
  • Swagger UI;
  • 另一台服务器。

例如:

python
import requests

response = requests.get("http://127.0.0.1:8000/")
print(response.json())

这里的 requests 工作在客户端一侧。

3.2 Server 是谁

Server 是监听网络请求、执行处理逻辑并返回响应的一方。

在当前项目中:

text
Uvicorn
→ 启动并监听端口

FastAPI
→ 匹配路由、验证数据、调用函数、构造响应

main.py / model.py / memory.py
→ 实际业务代码

3.3 本机地址 127.0.0.1

当前访问:

text
http://127.0.0.1:8000

可以拆成:

text
http
→ 使用 HTTP 协议

127.0.0.1
→ 当前这台电脑,也叫 localhost

8000
→ Uvicorn 正在监听的端口

它不是公网地址。一般只有当前电脑自己能访问,因此还不等于“已经部署到互联网”。


四、FastAPI 与 Uvicorn 各自负责什么

这是学习初期非常容易混淆的一组概念。

4.1 FastAPI 是 Web 框架

FastAPI 主要负责:

  • 定义路由;
  • 规定请求数据结构;
  • 解析和验证请求;
  • 调用 Python 处理函数;
  • 把返回值转换成 HTTP Response;
  • 生成 OpenAPI 文档;
  • 提供 Swagger UI。

例如:

python
from fastapi import FastAPI

app = FastAPI()

逐层理解:

text
FastAPI
→ 一个 class

FastAPI()
→ 调用 class 创建对象

app
→ 创建出来的 FastAPI 应用对象

路由、文档以及请求处理规则都注册在 app 上。

4.2 Uvicorn 是 ASGI Web Server

仅仅写好:

python
app = FastAPI()

并不会自动开始监听网络端口。需要 Uvicorn 真正运行这个应用:

bash
uvicorn main:app --reload

这条命令可以拆成:

text
uvicorn
→ 启动 Web Server

main
→ 找 main.py 这个 Python 模块

app
→ 找模块中的 FastAPI 应用对象

--reload
→ 开发期间监测代码变化,并自动重载后端程序

完整关系是:

text
网络请求
  ↓
Uvicorn 监听到请求
  ↓
交给 FastAPI app
  ↓
FastAPI 匹配路由并执行 Python 函数

4.3 一个常见误区

错误理解:

FastAPI 就是服务器,写完代码以后会自己监听 8000 端口。

正确理解:

text
FastAPI
→ 描述 Web 应用怎样处理请求

Uvicorn
→ 真正运行这个应用并监听网络

五、路由:把网络地址映射到 Python 函数

5.1 根路由示例

python
@app.get("/")
def root():
    return {
        "message": "API正在运行"
    }

@app.get("/") 当前可以理解为一种“注册关系”:

text
如果收到 GET /
        ↓
就执行 root()

因此,访问:

text
http://127.0.0.1:8000/

不是在访问 main.py 文件,而是在访问 FastAPI 应用中注册在 / 路径上的 API。

客户端甚至不需要知道服务器代码文件叫:

text
main.py、server.py,还是 abc.py

客户端只关心:

text
协议 + IP/域名 + 端口 + 路径 + HTTP 方法

5.2 /chat 路由

python
@app.post("/chat")
def chat(request: ChatRequest):
    ...

它建立的是:

text
POST /chat
    ↓
chat(request)

同一个路径配合不同 HTTP 方法,也可以表达不同操作。因此“路径”和“方法”需要放在一起理解。

5.3 Python 装饰器在这里的直观作用

@app.post("/chat") 是装饰器语法。当前阶段不需要先深挖装饰器底层,只要牢牢记住它在这里的工程作用:

把一个普通 Python 函数登记为可以通过 HTTP 访问的路径操作函数。


六、Request、Response、GET 与 POST

6.1 Request 是什么

Request 是客户端发给服务器的一份请求。它通常包含:

text
HTTP 方法
URL / Path
Headers
Query Parameters
Request Body

例如聊天请求:

json
{
  "conversation_id": "abc123",
  "question": "什么是 Agent?"
}

6.2 Response 是什么

Response 是服务器处理以后返回给客户端的结果,通常包含:

text
HTTP 状态码
Response Headers
Response Body

例如:

json
{
  "conversation_id": "abc123",
  "answer": "Agent 是……"
}

当 FastAPI 路由返回 Python 字典时,FastAPI 会把它序列化成 JSON,并设置相应的响应类型。

6.3 GET 与 POST

可以先这样区分:

GET

一般用于读取资源或获取信息。

python
@app.get("/")
def root():
    ...

POST

一般用于向服务器提交数据,让服务器执行某个操作。

python
@app.post("/chat")
def chat(request: ChatRequest):
    ...

聊天问题通常内容较长、结构也会不断扩展,因此适合放在 JSON Request Body 中,而不是全部塞进 URL。

6.4 Path、Query 与 Body

三者的直观区别:

text
Path
→ 表示访问哪个资源,例如 /conversations/abc123

Query
→ 表示额外筛选或选项,例如 ?limit=20

Body
→ 表示提交的一整份结构化数据,例如问题、模型和会话 ID

本项目主要使用 Request Body:

json
{
  "conversation_id": "abc123",
  "question": "我叫什么名字?",
  "model": "qwen-plus"
}

七、Pydantic:让请求数据有明确结构

7.1 为什么不直接接收任意字典

如果服务器随便接收一份字典,就需要自己逐项检查:

text
有没有 question?
question 是不是字符串?
有没有 conversation_id?
model 可不可以不传?

Pydantic 的 BaseModel 可以把这些要求正式写进代码:

python
from typing import Optional
from pydantic import BaseModel


class ChatRequest(BaseModel):
    conversation_id: str
    question: str
    model: Optional[str] = None

这不是“大模型的 Model”,而是“数据模型”。它描述请求应该长什么样。

7.2 必填字段与可选字段

python
question: str

表示:

  • 字段名是 question;
  • 期望类型是字符串;
  • 没有默认值,所以通常必须提供。
python
model: Optional[str] = None

表示:

  • model 可以是字符串;
  • 也可以是 None;
  • 客户端不传时,值就是 None。

后续可在模型层使用默认模型:

python
if model is None:
    model = DEFAULT_MODEL

这形成了:

text
客户端不指定模型
  ↓
使用项目默认模型

客户端主动指定模型
  ↓
使用本次请求给出的模型

7.3 422 从哪里来

如果 question 没有默认值,而客户端没有传,或者传入的数据类型不符合要求,FastAPI 会在进入 chat() 之前完成验证并返回 422 Unprocessable Entity。

数据流是:

text
Request Body
  ↓
Pydantic 验证
  ├── 成功 → 创建 ChatRequest 对象 → 进入 chat()
  └── 失败 → FastAPI 直接返回 422

所以,看到 422 时首先检查请求体字段,而不是立刻怀疑模型 API。

7.4 为什么 Pydantic 还能帮助 Swagger

完整链路:

text
Python Type Hint + Pydantic BaseModel + FastAPI 路由
  ↓
FastAPI 生成 OpenAPI Schema
  ↓
Swagger UI 读取 Schema
  ↓
自动显示字段、类型、必填项和请求示例

八、Swagger 与 OpenAPI:自动生成的测试客户端

8.1 /docs 是什么

启动服务后访问:

text
http://127.0.0.1:8000/docs

会看到 Swagger UI。它可以:

  • 查看所有路由;
  • 查看参数和 Request Body;
  • 点击 Try it out;
  • 直接发送 HTTP 请求;
  • 查看状态码、Response Body 和 Headers。

8.2 Swagger 到底属于客户端还是服务器

这个问题要分两层:

text
FastAPI 服务器
→ 提供 /docs 页面和 /openapi.json

浏览器打开 /docs 后
→ Swagger UI 在浏览器中运行
→ 主动向 API 发送请求
→ 此时扮演 Client

所以可以把 Swagger UI 理解成:

FastAPI 自动提供的一个网页 API 测试客户端。

8.3 /openapi.json 的作用

FastAPI 会根据路由和 Pydantic 模型生成 API 描述:

text
Python 代码
  ↓
FastAPI
  ↓
/openapi.json
  ↓
Swagger /docs

Swagger 并不是猜测请求中有哪些字段,而是读取 OpenAPI Schema 后生成界面。

8.4 实际遇到的“旧页面”问题

修改后端代码以后,--reload 只负责重新加载服务器代码,并不会强制浏览器刷新已经打开的 /docs 页面。

当时实际发生的是:

text
main.py 默认模型已经改成 qwen-plus
  ↓
Uvicorn 已经 reload
  ↓
浏览器里的旧 Swagger 页面没有刷新
  ↓
Request Body 仍然保留 model="Qwen"
  ↓
点击 Execute
  ↓
上游返回 404 model_not_found

因此修改以下内容后,应该刷新 /docs:

text
路由
Pydantic Model
字段类型
默认值
Request Body 结构

如果仍有疑问,可直接打开 /openapi.json 检查后端当前真正生成的 Schema。


九、搭建第一个可运行的 FastAPI 项目

9.1 第一版项目结构

text
fastapi_agent_demo/
│
├── .venv/
├── main.py
├── model.py
└── requirements.txt

职责:

text
main.py
→ HTTP、FastAPI、路由、请求和响应

model.py
→ 模型调用逻辑

requirements.txt
→ 第三方依赖清单

.venv/
→ 当前项目独立的 Python 运行环境

9.2 创建环境与安装依赖

bash
python -m venv .venv
.venv\Scripts\activate
pip install fastapi uvicorn
pip freeze > requirements.txt

理解:

text
.venv
→ 真正安装出来的环境,一般不上传

requirements.txt
→ 重建环境所需的清单,应该保留在项目中

9.3 先写一个假的模型层

model.py:

python
def call_model(
    question: str,
    model: str
) -> str:
    answer = "使用 " + model + " 回答:" + question

    return answer

它没有访问真实模型,只是在模拟“模型层的输入和输出”。

为什么先这样做?因为调试时应该分层验证:

text
第一步:先证明 HTTP 服务可以跑通
第二步:再接真实外部模型

如果一开始同时加入 FastAPI、网络、Key、SDK 和模型,就很难判断错误究竟在哪一层。

9.4 第一版 main.py

python
from fastapi import FastAPI
from pydantic import BaseModel

from model import call_model


app = FastAPI()


class ChatRequest(BaseModel):
    question: str
    model: str = "Qwen"


@app.get("/")
def root():
    return {
        "message": "API正在运行"
    }


@app.post("/chat")
def chat(request: ChatRequest):
    answer = call_model(
        question=request.question,
        model=request.model
    )

    return {
        "answer": answer
    }

9.5 第一次完整数据流

text
Swagger UI
  ↓ POST /chat
Uvicorn
  ↓
FastAPI
  ↓
ChatRequest 解析 Request Body
  ↓
chat()
  ↓
call_model()
  ↓
模拟回答
  ↓
Python dict
  ↓
FastAPI 转成 JSON Response
  ↓
Swagger 显示结果

这一步虽然模型是假的,但 HTTP 服务已经是真实、完整的。


十、从模拟回答升级到真实大模型

10.1 变化发生在哪里

FastAPI 外壳不需要推倒重来。主要替换的是 model.py:

text
以前:
call_model() 拼接字符串

以后:
call_model() 调用真实模型 SDK

这证明分层设计的价值:

text
main.py 的 HTTP 逻辑保持稳定
model.py 内部替换具体模型实现

10.2 使用兼容接口的 SDK

学习项目使用 OpenAI 风格 SDK 调用 Qwen 兼容接口。核心结构如下:

python
from openai import OpenAI


client = OpenAI(
    api_key=API_KEY,
    base_url=BASE_URL
)

然后请求模型:

python
completion = client.chat.completions.create(
    model=model,
    messages=[
        {
            "role": "system",
            "content": "你是一个有帮助的AI助手"
        },
        {
            "role": "user",
            "content": question
        }
    ]
)

最后取出回答:

python
answer = completion.choices[0].message.content

10.3 为什么响应结构不是直接一个字符串

模型服务返回的不只是文本,还可能包含:

text
多个候选结果 choices
消息角色
文本内容
结束原因
Token 使用量
请求元数据

所以需要从响应对象中逐层取值:

text
completion
  ↓
choices[0]
  ↓
message
  ↓
content

10.4 “Qwen”与“qwen-plus”不是同一个概念

text
Qwen
→ 人类使用的产品或模型家族名称

qwen-plus
→ API 能识别的具体模型 ID 示例

API 不会根据宽泛的品牌名自动猜测具体模型。必须传接口支持、账号有权访问的准确模型 ID。


十一、环境变量与配置分层

11.1 为什么不能把 Key 写进代码

不应该这样:

python
API_KEY = "真实密钥"

原因包括:

  • 代码上传 GitHub 时可能泄露;
  • 截图或分享代码时可能泄露;
  • 本地、测试和生产环境可能使用不同配置;
  • 修改配置还要改动业务代码;
  • 一旦泄露,别人可能盗用额度。

更合理的结构:

text
.env
→ 保存本机秘密和环境配置

config.py
→ 读取、检查并向 Python 模块提供配置

model.py
→ 使用配置创建模型客户端

11.2 .env

示例:

env
LLM_API_KEY=你的真实APIKey
LLM_BASE_URL=模型服务兼容接口地址
LLM_MODEL=qwen-plus

注意:真实 Key 不应写进笔记、截图或聊天记录。

11.3 .gitignore

至少应忽略:

gitignore
.env
.venv/
__pycache__/

其中:

text
.env
→ 因为包含秘密

.venv/
→ 因为体积大且可以根据依赖清单重建

__pycache__/
→ Python 自动生成的缓存

11.4 config.py

python
import os

from dotenv import load_dotenv


load_dotenv()


API_KEY = os.getenv("LLM_API_KEY")
BASE_URL = os.getenv("LLM_BASE_URL")
DEFAULT_MODEL = os.getenv("LLM_MODEL")


if API_KEY is None:
    raise RuntimeError("没有找到 LLM_API_KEY")

if BASE_URL is None:
    raise RuntimeError("没有找到 LLM_BASE_URL")

if DEFAULT_MODEL is None:
    raise RuntimeError("没有找到 LLM_MODEL")

11.5 为什么配置缺失时直接 raise RuntimeError

如果没有 API Key,模型调用一定无法正常工作。仅仅:

python
print("API Key读取失败")

会让程序继续运行,直到用户发请求时才以更模糊的方式失败。

更合理的是:

text
启动时读取配置
  ↓
发现必要配置缺失
  ↓
立即停止并给出明确错误

这叫“尽早失败”:让错误在最接近原因的地方暴露。

11.6 RuntimeError 与 HTTPException

text
RuntimeError
→ 程序运行所需的基础条件不成立
→ 例如启动时没有 API Key

HTTPException
→ 某一次 HTTP 请求需要返回错误响应
→ 例如上游模型暂时不可用

11.7 单一事实来源

早期版本可能在多处重复写:

text
.env       → qwen-plus
main.py    → qwen-plus
model.py   → qwen-plus

一旦只改其中一处,就会不一致。

整理后:

text
.env 中的 LLM_MODEL
  ↓
config.py 中的 DEFAULT_MODEL
  ↓
model.py 使用

这就是 Single Source of Truth:同一项配置尽量只维护一个权威来源。

11.8 默认配置 + 允许覆盖

python
def call_model(
    messages,
    model: Optional[str] = None
):
    if model is None:
        model = DEFAULT_MODEL

它同时满足两种需求:

text
普通请求
→ 不传 model,使用默认值

特殊请求
→ 主动传 model,覆盖本次默认值

十二、真实 LLM 请求的数据流

真实模型接入后,请求链变成:

text
Swagger UI / Client
  ↓ POST /chat + JSON
Uvicorn
  ↓
FastAPI app
  ↓
Pydantic ChatRequest
  ↓
chat()
  ↓
model.py 的 call_model()
  ↓
OpenAI 风格 SDK
  ↓ HTTP 请求
模型服务 API
  ↓
Qwen 模型推理
  ↓
SDK 响应 completion
  ↓
choices[0].message.content
  ↓
answer
  ↓
FastAPI JSON Response
  ↓
Swagger UI / Client

还存在一条配置链:

text
.env
  ↓ load_dotenv()
config.py
  ↓ API_KEY / BASE_URL / DEFAULT_MODEL
model.py
  ↓ 创建 client 并调用模型

理解这两条链以后,排错就不再是“所有文件一起乱改”,而是判断:

text
请求链在哪一层停止?
配置链是否把正确值传到了模型层?

十三、错误处理:从吞掉错误到分层传递

13.1 第一版错误处理的问题

早期代码:

python
try:
    ...
except Exception as e:
    print("模型调用失败:", repr(e))
    return None

然后 main.py:

python
if answer is None:
    raise HTTPException(
        status_code=500,
        detail="模型调用失败"
    )

它能防止程序直接崩溃,但丢失了大量信息:

text
究竟是连接失败?
认证失败?
模型不存在?
频率限制?
上游服务故障?

全部被压成:

text
None → 500 模型调用失败

13.2 为什么 return None 会“吞掉”错误

异常本来携带类型、状态码和详细信息。转换成 None 后,上层只知道“没有结果”,却不知道原因。

text
上游明确错误
  ↓ except Exception
return None
  ↓
错误身份丢失
  ↓
main.py 只能统一返回 500

13.3 更合理的职责划分

text
model.py
→ 了解模型 SDK
→ 识别或保留底层异常

main.py
→ 了解 HTTP API
→ 决定客户端应该收到什么状态码和说明

简化写法可以让模型层记录后重新抛出:

python
def call_model(...):
    try:
        ...
        return answer
    except Exception as error:
        print("模型调用失败:", repr(error))
        raise

这里的裸 raise 表示:

把当前正在处理的原始异常继续抛给调用者。

它与 return None 完全不同。

13.4 在 API 层转换成 HTTP 错误

概念示例:

python
from fastapi import HTTPException
from openai import (
    APIConnectionError,
    AuthenticationError,
    NotFoundError,
    RateLimitError,
    APIStatusError,
)


try:
    answer = call_model(...)
except AuthenticationError:
    raise HTTPException(
        status_code=500,
        detail="模型服务认证配置错误"
    )
except NotFoundError:
    raise HTTPException(
        status_code=400,
        detail="指定的模型不存在或当前账号无权访问"
    )
except RateLimitError:
    raise HTTPException(
        status_code=429,
        detail="模型服务请求过于频繁,请稍后再试"
    )
except APIConnectionError:
    raise HTTPException(
        status_code=503,
        detail="暂时无法连接模型服务"
    )
except APIStatusError:
    raise HTTPException(
        status_code=502,
        detail="上游模型服务返回异常"
    )
except Exception:
    raise HTTPException(
        status_code=500,
        detail="服务器处理模型请求时发生未知错误"
    )

这里的映射不是永恒不变的标准答案,而是一种工程设计。重点是:不要把所有不同失败都伪装成完全相同的 500。

13.5 两层 API 为什么会出现两个状态码

当前系统实际调用了两层 API:

text
客户端
  ↓
自己的 FastAPI /chat
  ↓
模型服务 API

上游模型可能返回 404,但如果自己的代码把异常转换成了统一 500,那么:

text
模型服务:404 model_not_found
  ↓
model.py 捕获后 return None
  ↓
main.py 发现 None
  ↓
自己的 API:500 Internal Server Error

因此,终端中看到的上游错误和 Swagger 中看到的本地 API 状态码不一定相同。

13.6 常用状态码在本项目中的理解

状态码 当前阶段的直观含义
200 请求成功并得到回答
400 客户端提供的参数有问题,例如模型名无效
401 认证失败,通常需要检查 Key 或权限
404 请求的上游资源或模型不存在
422 FastAPI/Pydantic 对 Request Body 验证失败
429 请求过于频繁或触发额度限制
500 自己的服务发生未预期内部错误
502 自己作为网关调用上游时,上游返回异常
503 当前暂时无法连接或使用依赖服务

十四、实际排错记录与方法

这一阶段最有价值的部分之一,不只是“最终运行成功”,而是亲自遇到并定位了不同层的错误。

14.1 APIConnectionError: Connection error

数据流定位:

text
POST /chat 已经进入 FastAPI       ✅
chat() 已经调用 call_model()      ✅
SDK 尝试连接外部模型服务          ✅
没有正常建立连接或取得 HTTP 响应  ❌

因此它和 401、404 不在同一层。

排查顺序应包括:

  1. BASE_URL 是否为 None;
  2. .env 的变量名是否和 os.getenv() 完全一致;
  3. base_url=BASE_URL 是否误写成字符串 base_url="BASE_URL";
  4. 地址协议、路径和地域配置是否正确;
  5. 网络、代理、VPN、防火墙是否影响连接;
  6. 修改 config.py 后是否正确导入。

可临时打印非敏感信息:

python
print("BASE_URL =", BASE_URL)
print("DEFAULT_MODEL =", DEFAULT_MODEL)

但不要打印:

python
print(API_KEY)

14.2 404 model_not_found

这个错误反而证明:

text
电脑网络 → 模型服务器

通常已经走通,因为服务器明确收到了请求并返回了结构化错误。

当时原因是传入:

json
{
  "model": "Qwen"
}

而接口需要具体模型 ID,例如学习时使用的:

json
{
  "model": "qwen-plus"
}

排错原则:错误信息已经明确指向模型名时,不要又同时乱改网络、Key 和 FastAPI 路由。

14.3 后端已更新,Swagger 仍发送旧值

原因不是后端没重载,而是浏览器 /docs 页面仍保留旧 Schema 或旧输入内容。

解决思路:

text
确认 Uvicorn 已 reload
  ↓
刷新 /docs
  ↓
重新检查 Request Body
  ↓
必要时查看 /openapi.json

14.4 调试的通用原则:从最近变动处开始

如果程序原来能运行,而把配置读取从 model.py 移到 config.py 后立刻报错,第一优先级应该检查新增加的配置链:

text
.env
  ↓
config.py
  ↓
model.py

而不是一开始就怀疑所有可能原因。

14.5 根据错误判断“请求走到了哪里”

text
422
→ 多数还没进入 chat(),请求验证阶段失败

本地 404
→ 可能没有匹配到 FastAPI 路由

APIConnectionError
→ 已进入业务函数,但连接上游失败

上游 401
→ 通常已经连接上服务器,认证没有通过

上游 404 model_not_found
→ 通常已经连接成功,但模型标识或权限有问题

200
→ 整条请求与响应链成功完成

这是一种非常重要的后端调试思维:

不要只看“失败了”,要根据错误判断请求已经成功经过了哪些层,停止在哪一层。


十五、同步、异步与 Agent 的等待时间

15.1 同步函数与异步函数

同步路由:

python
@app.post("/chat")
def chat(...):
    ...

异步路由:

python
@app.post("/chat")
async def chat(...):
    ...

异步函数中可以等待真正的异步操作:

python
answer = await call_model(...)

15.2 为什么 Agent 特别需要关注异步

Agent 经常需要等待:

text
大模型 API
网页搜索
天气 API
数据库查询
远程工具
文件服务

这些通常属于 I/O 等待型任务。等待一个外部服务返回时,CPU 不一定需要一直空耗在那里。

text
用户 A 等待模型
  ↓ 暂时让出执行机会
Event Loop 推进用户 B
  ↓ B 也等待
Event Loop 推进用户 C

异步提升的是服务在等待型任务下的并发处理能力,而不是让模型本身推理得更快。

15.3 并发与并行

text
并发 Concurrency
→ 多个任务都在推进
→ 一个任务等待时切换到另一个任务

并行 Parallelism
→ 同一时刻由多个执行单元真正执行多个任务

二者不是同一个概念。

15.4 async def 不是异步魔法

错误示例:

python
async def get_data():
    response = requests.get(url)

虽然函数写成 async def,但 requests.get() 仍是同步阻塞操作。它不会自动变成可被 Event Loop 高效调度的异步网络请求。

合理选择:

text
同步 SDK / 同步 HTTP 库
→ 可以使用同步函数和普通 def 路由

异步 SDK / 异步 HTTP 客户端
→ 使用 async def + await

不是所有代码加上 async 就更高级。关键是内部调用的库是否真正支持异步,以及任务是否适合异步。

15.5 连续 await 不一定并发

python
a = await task_a()
b = await task_b()
c = await task_c()

这表示:完成 A 后才开始 B,完成 B 后才开始 C。

若三个任务互相独立,需要一起推进,可以使用类似:

python
results = await asyncio.gather(
    task_a(),
    task_b(),
    task_c()
)

但只有在任务确实独立、库真正异步并且业务允许并发时才这样设计。


十六、从单轮调用进入多轮对话

16.1 单轮调用是什么

最初发送给模型的内容:

python
messages = [
    {
        "role": "system",
        "content": "你是一个有帮助的AI助手"
    },
    {
        "role": "user",
        "content": question
    }
]

每次请求只有当前问题。第一次:

text
我的名字叫小明

第二次如果只发送:

text
我叫什么名字?

模型通常并不知道答案,因为第二次请求中没有“小明”这个信息。

16.2 一个关键认识:模型 API 本身不会自动记住上一次请求

常见误解:

第一次告诉了模型名字,第二次调用同一个模型,它自然应该记得。

实际机制:

text
第一次 HTTP 请求结束
  ↓
第二次是新的 HTTP 请求
  ↓
如果应用没有重新提供历史
  ↓
模型只看到本次输入

看起来“有记忆”的系统,本质通常是:

text
应用保存历史
  +
每次重新把相关历史发送给模型
  =
模型能够根据上下文回答

16.3 多轮 messages

python
messages = [
    {
        "role": "system",
        "content": "你是一个有帮助的AI助手"
    },
    {
        "role": "user",
        "content": "我的名字叫小明"
    },
    {
        "role": "assistant",
        "content": "好的,我记住了,你叫小明"
    },
    {
        "role": "user",
        "content": "我叫什么名字?"
    }
]

模型现在能回答“小明”,不是因为它在上一次请求后改变了参数或永久存储了用户,而是因为当前请求再次把历史放进了上下文。

16.4 为什么使用 list + dict

外层是列表:

text
list
→ 保存多条消息
→ 保留先后顺序

每条消息是字典:

text
dict
→ 描述一条消息的多个字段
→ role 表示谁说的
→ content 表示说了什么

因此:

text
messages
  ↓ list
一组按时间排列的消息
  ↓ dict
每条消息都有 role 和 content

消息顺序非常重要,因为对话的意义依赖上下文顺序。


十七、system、user、assistant 三种角色

17.1 system

python
{
    "role": "system",
    "content": "你是一个耐心的 Python 老师"
}

它主要描述模型在本次上下文中的行为、身份、边界或输出要求。

它不是普通用户消息,而更接近应用提供的全局指导。

17.2 user

python
{
    "role": "user",
    "content": "什么是 FastAPI?"
}

表示真实用户输入的问题、任务或补充信息。

17.3 assistant

python
{
    "role": "assistant",
    "content": "FastAPI 是一个 Python Web 框架……"
}

表示模型过去已经给出的回答。

为什么连模型自己的历史回答也要保存?因为下一轮中,模型需要知道:

  • 自己之前回答过什么;
  • 用户说的“你刚才那一点”指什么;
  • 当前对话已经形成了哪些约定;
  • 避免前后矛盾或重复。

17.4 三种角色形成的对话结构

text
system
→ 规定总体行为

user
→ 提出任务

assistant
→ 产生回答

user
→ 根据回答继续追问

这比把所有文字拼成一个大字符串更清楚,也让模型知道每段内容来自谁。


十八、为什么需要 conversation_id

18.1 只保存一个全局消息列表会怎样

如果服务器只有:

python
messages = []

所有用户都向同一份列表写入历史,就可能发生:

text
用户 A 告诉模型自己的名字
  ↓
用户 B 发来问题
  ↓
用户 B 的上下文里混入用户 A 的历史

这不仅会答错,还是严重的数据隔离问题。

18.2 conversation_id 是会话的索引

请求:

json
{
  "conversation_id": "abc123",
  "question": "我叫什么名字?"
}

服务器可以理解为:

text
先找到编号为 abc123 的聊天窗口
  ↓
读取它自己的历史
  ↓
加入当前问题
  ↓
调用模型

它像不同聊天窗口的编号:

text
abc123
→ 一组独立历史

xyz456
→ 另一组独立历史

18.3 会话 ID 不完全等于用户 ID

同一个用户可以创建多个聊天窗口,因此通常:

text
一个 user
  ↓
可以拥有多个 conversation
  ↓
每个 conversation 拥有多条 message

本阶段尚未加入用户系统,只使用 conversation_id 完成最小会话隔离。

18.4 谁应该生成 conversation_id

学习 Demo 中可以手动填写 abc123,方便观察机制。正式系统中通常由服务器创建唯一 ID,再交给客户端保存和继续使用。

如果长期手动随便写,会产生:

  • ID 冲突;
  • 拼写错误导致历史丢失;
  • 用户可能访问不属于自己的会话;
  • 无法可靠建立用户与会话的权限关系。

十九、用 Python 字典实现服务器端 Memory

19.1 数据结构

python
from typing import Dict, List


memory: Dict[str, List[Dict[str, str]]] = {}

从外到内拆解:

text
memory
→ 整个服务器当前保存的会话集合

Dict 的 key:str
→ conversation_id,例如 abc123

Dict 的 value:List[...]
→ 这个会话的多条历史消息

List 中的每个 Dict[str, str]
→ 一条消息,例如 role + content

运行一段时间后可能变成:

python
memory = {
    "abc123": [
        {
            "role": "user",
            "content": "我的名字叫小明"
        },
        {
            "role": "assistant",
            "content": "好的,我记住了,你叫小明"
        }
    ],
    "xyz456": [
        {
            "role": "user",
            "content": "你好"
        },
        {
            "role": "assistant",
            "content": "你好!有什么可以帮助你?"
        }
    ]
}

19.2 get_messages()

python
def get_messages(
    conversation_id: str
) -> List[Dict[str, str]]:
    return memory.get(
        conversation_id,
        []
    )

memory.get(conversation_id, []) 表示:

text
如果 key 存在
→ 返回对应历史列表

如果 key 不存在
→ 返回默认空列表 []

为什么需要 []?新会话第一次请求时还没有历史,但这不是错误,而是“历史为空”。返回空列表后,后续仍可以正常:

python
messages.extend(history)

19.3 一个重要细节:get() 不会创建 key

python
history = memory.get("new-id", [])

得到空列表,不等于 memory 中已经出现了 "new-id"。

text
get()
→ 读取
→ 找不到时返回默认值
→ 不负责把默认值写回字典

真正创建会话要在写入逻辑中完成。

19.4 add_message()

python
def add_message(
    conversation_id: str,
    role: str,
    content: str
) -> None:
    if conversation_id not in memory:
        memory[conversation_id] = []

    memory[conversation_id].append(
        {
            "role": role,
            "content": content
        }
    )

逐步执行:

text
收到 conversation_id
  ↓
检查 memory 中是否已经有这个会话
  ↓ 没有
创建一个空消息列表
  ↓
把新消息 append 到该会话列表末尾

返回类型是 None,因为这个函数的主要作用不是计算一个结果,而是修改服务器内存中的状态。

19.5 append() 与 extend()

append():加入一个整体元素

python
messages.append(
    {
        "role": "user",
        "content": question
    }
)

表示把“一条消息字典”作为一个元素放到列表末尾。

extend():加入另一个列表中的所有元素

python
messages.extend(history)

如果 history 有两条消息,extend() 会把这两条分别接到 messages 后面。

区别:

python
messages.append(history)

会把整个历史列表当成一个元素,形成嵌套列表,这不符合模型 API 期望的消息结构。

19.6 为什么先构建临时 messages

每次请求中,可以创建一份本次发送给模型的临时上下文:

python
messages = [
    {
        "role": "system",
        "content": "你是一个有帮助的AI助手"
    }
]

messages.extend(history)

messages.append(
    {
        "role": "user",
        "content": request.question
    }
)

这样做有几个好处:

  • system 指令可以在每次调用时统一添加;
  • 历史 Memory 只保存真正的 user/assistant 对话;
  • 本次问题可以先进入临时上下文,而不是立刻污染长期状态;
  • 如果模型失败,Memory 可以保持在上一轮成功状态。

19.7 为什么模型成功以后再保存两条消息

错误顺序:

text
一收到用户问题就保存 user 消息
  ↓
调用模型失败
  ↓
没有 assistant 回答
  ↓
Memory 留下一条未完成对话

这种历史会让后续上下文与真实交互状态不一致。

当前更稳妥的顺序:

text
读取历史
  ↓
构建临时 messages
  ↓
调用模型
  ↓ 成功
保存 user 消息
  ↓
保存 assistant 消息

核心思想是:

只有这一轮对话成功完成后,才把完整的一问一答提交到 Memory。

这是最基础的数据一致性意识。未来使用数据库时,会进一步学习事务、失败回滚和消息状态。


二十、最终版代码与逐层解释

下面代码用于还原这一阶段最终形成的结构。不同 SDK 版本的具体异常类或接口细节可能需要按实际环境调整,但职责划分和数据流不变。

20.1 项目结构

text
fastapi_agent_demo/
│
├── .venv/
├── .env
├── .gitignore
├── requirements.txt
│
├── config.py
├── memory.py
├── model.py
└── main.py

职责:

text
.env
→ 保存原始环境配置和秘密

config.py
→ 读取并验证配置

model.py
→ 调用大模型

memory.py
→ 按 conversation_id 保存和读取历史

main.py
→ HTTP 路由、请求验证、编排调用、错误响应

Uvicorn
→ 运行整个 FastAPI 应用

20.2 .env

env
LLM_API_KEY=你的真实APIKey
LLM_BASE_URL=你的模型服务兼容接口地址
LLM_MODEL=qwen-plus

20.3 config.py

python
import os

from dotenv import load_dotenv


load_dotenv()


API_KEY = os.getenv("LLM_API_KEY")
BASE_URL = os.getenv("LLM_BASE_URL")
DEFAULT_MODEL = os.getenv("LLM_MODEL")


if API_KEY is None:
    raise RuntimeError("没有找到 LLM_API_KEY")

if BASE_URL is None:
    raise RuntimeError("没有找到 LLM_BASE_URL")

if DEFAULT_MODEL is None:
    raise RuntimeError("没有找到 LLM_MODEL")

20.4 memory.py

python
from typing import Dict, List


Message = Dict[str, str]
Memory = Dict[str, List[Message]]


memory: Memory = {}


def get_messages(
    conversation_id: str
) -> List[Message]:
    return memory.get(
        conversation_id,
        []
    )


def add_message(
    conversation_id: str,
    role: str,
    content: str
) -> None:
    if conversation_id not in memory:
        memory[conversation_id] = []

    memory[conversation_id].append(
        {
            "role": role,
            "content": content
        }
    )

这里增加 Message 和 Memory 类型别名,只是为了让嵌套类型更容易阅读:

text
Message
→ 一条消息 Dict[str, str]

Memory
→ conversation_id 到消息列表的映射

20.5 model.py

python
from typing import Dict, List, Optional

from openai import OpenAI

from config import API_KEY, BASE_URL, DEFAULT_MODEL


client = OpenAI(
    api_key=API_KEY,
    base_url=BASE_URL
)


def call_model(
    messages: List[Dict[str, str]],
    model: Optional[str] = None
) -> str:
    if model is None:
        model = DEFAULT_MODEL

    try:
        completion = client.chat.completions.create(
            model=model,
            messages=messages
        )

        answer = completion.choices[0].message.content

        if answer is None:
            raise RuntimeError("模型没有返回文本内容")

        return answer

    except Exception as error:
        print("模型调用失败:", repr(error))
        raise

相比单轮版本,最大的变化是:

text
以前 call_model(question)
→ model.py 自己临时构建两条消息

现在 call_model(messages)
→ 上层已经把 system、历史和当前问题组织好
→ model.py 只负责把 messages 发给模型

这种分工让模型层不必知道历史来自内存、数据库还是其他地方。

20.6 main.py

python
from typing import Optional

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from memory import add_message, get_messages
from model import call_model


app = FastAPI()


class ChatRequest(BaseModel):
    conversation_id: str
    question: str
    model: Optional[str] = None


@app.get("/")
def root():
    return {
        "message": "API正在运行"
    }


@app.post("/chat")
def chat(request: ChatRequest):
    history = get_messages(
        request.conversation_id
    )

    messages = [
        {
            "role": "system",
            "content": "你是一个有帮助的AI助手"
        }
    ]

    messages.extend(history)

    messages.append(
        {
            "role": "user",
            "content": request.question
        }
    )

    try:
        answer = call_model(
            messages=messages,
            model=request.model
        )
    except Exception:
        raise HTTPException(
            status_code=503,
            detail="暂时无法完成模型请求"
        )

    add_message(
        conversation_id=request.conversation_id,
        role="user",
        content=request.question
    )

    add_message(
        conversation_id=request.conversation_id,
        role="assistant",
        content=answer
    )

    return {
        "conversation_id": request.conversation_id,
        "answer": answer
    }

20.7 main.py 为什么像“总指挥”

main.py 没有亲自实现模型调用,也没有亲自维护字典细节,而是负责组织顺序:

text
接收请求
  ↓
读取 Memory
  ↓
构建本次上下文
  ↓
调用模型
  ↓
成功后保存 Memory
  ↓
返回 Response

这就是关注点分离:每个模块承担清晰职责,通过函数协作。

20.8 为什么 system 没有保存进 Memory

当前实现每次都会主动把 system 放在临时 messages 的第一项,因此没有必要在每轮都重复写入历史。

当前 Memory 保存:

text
user 消息
assistant 消息

调用前再组合:

text
system + history + current user

未来如果不同会话需要不同 system prompt,可以把它保存为会话配置,而不是每轮重复保存为普通消息。


二十一、一次完整请求到底经历了什么

以请求为例:

json
{
  "conversation_id": "abc123",
  "question": "我叫什么名字?"
}

第 1 步:客户端发送 HTTP Request

text
POST /chat
Content-Type: application/json
Body: conversation_id + question

第 2 步:Uvicorn 接收

Uvicorn 正在监听 127.0.0.1:8000,收到请求后把它交给 FastAPI 应用。

第 3 步:FastAPI 匹配路由

text
POST /chat
  ↓
@app.post("/chat")
  ↓
chat()

第 4 步:Pydantic 验证

请求体被解析成:

python
request.conversation_id
request.question
request.model

第 5 步:读取历史

python
history = get_messages("abc123")

假设得到:

python
[
    {
        "role": "user",
        "content": "我的名字叫小明"
    },
    {
        "role": "assistant",
        "content": "好的,我记住了,你叫小明"
    }
]

第 6 步:构建临时上下文

text
system
  ↓
历史 user
  ↓
历史 assistant
  ↓
本次 user

最终模型看到的内容相当于:

python
[
    {
        "role": "system",
        "content": "你是一个有帮助的AI助手"
    },
    {
        "role": "user",
        "content": "我的名字叫小明"
    },
    {
        "role": "assistant",
        "content": "好的,我记住了,你叫小明"
    },
    {
        "role": "user",
        "content": "我叫什么名字?"
    }
]

第 7 步:调用模型

model.py 把完整 messages 发送给 Qwen。模型根据当前上下文回答:

text
你叫小明。

第 8 步:提交 Memory

模型成功后保存:

python
{"role": "user", "content": "我叫什么名字?"}
{"role": "assistant", "content": "你叫小明。"}

第 9 步:返回 HTTP Response

json
{
  "conversation_id": "abc123",
  "answer": "你叫小明。"
}

这条链可以压缩为:

text
Client
  ↓
Uvicorn
  ↓
FastAPI
  ↓
conversation_id
  ↓
memory.py 查历史
  ↓
组合 messages
  ↓
model.py
  ↓
Qwen
  ↓
保存完整一问一答
  ↓
Response

二十二、真实测试:记住小明与会话隔离

22.1 第一次请求:告诉 abc123 名字

json
{
  "conversation_id": "abc123",
  "question": "我的名字叫小明"
}

成功后,服务器 Memory 中形成:

python
memory["abc123"] = [
    {
        "role": "user",
        "content": "我的名字叫小明"
    },
    {
        "role": "assistant",
        "content": "好的,我记住了,你叫小明"
    }
]

22.2 第二次请求:仍使用 abc123

json
{
  "conversation_id": "abc123",
  "question": "我叫什么名字?"
}

客户端没有再次传“小明”,但服务器通过 abc123 找到了历史,并把历史重新发给模型,所以得到类似:

json
{
  "conversation_id": "abc123",
  "answer": "你叫小明。"
}

这证明:

text
历史保存成功
历史读取成功
上下文拼接成功
多轮对话成功

22.3 第三次请求:换成 xyz456

json
{
  "conversation_id": "xyz456",
  "question": "我叫什么名字?"
}

服务器执行:

python
get_messages("xyz456")

得到:

python
[]

模型没有看到 abc123 的历史,所以会回答类似:

text
我还不知道你的名字,你可以告诉我。

这证明:

text
abc123
→ 知道小明

xyz456
→ 不知道小明

也就是:

会话隔离成功。

22.4 这个实验真正验证了什么

它同时验证了四件事:

  1. Memory 能保存消息;
  2. 后续请求能取回历史;
  3. 模型可以根据重新发送的历史回答;
  4. 不同 conversation_id 不会共用同一份历史。

二十三、当前 Memory 的边界

23.1 它是服务器端 Memory,但不是持久化 Memory

python
memory = {}

存在于 Python 进程的 RAM 中:

text
Uvicorn 正在运行
→ memory 存在

Ctrl + C 停止 Uvicorn
→ Python 进程结束
→ RAM 数据消失

重新启动
→ 再次执行 memory = {}
→ 所有会话从空字典开始

因此,当前可以准确称为:

进程内、内存级、服务器端会话 Memory。

不能称为长期持久化记忆。

23.2 多 Worker 不共享普通字典

正式部署可能启动多个进程:

text
Worker A
→ 自己的 memory {}

Worker B
→ 另一份 memory {}

如果第一次请求落到 A,第二次请求落到 B,B 可能看不到 A 中保存的历史。

因此普通字典适合:

  • 学习 Memory 原理;
  • 单进程 Demo;
  • 快速原型。

不适合:

  • 多进程生产环境;
  • 服务器重启后仍需保留记录;
  • 多台服务器共享会话;
  • 大量用户与长历史;
  • 权限、搜索、统计和审计。

23.3 历史会无限增长

当前每轮都把完整历史重新发给模型:

text
第 1 轮:2 条消息
第 10 轮:20 条消息
第 100 轮:200 条消息

可能产生:

  • 超过模型上下文窗口;
  • Token 成本不断上升;
  • 响应变慢;
  • 早期无关信息干扰当前问题。

未来需要:

text
限制最近 N 轮
对旧历史做摘要
区分短期上下文与长期记忆
从长期记忆中按需检索

23.4 Memory 不等于模型永久学习

这里没有训练模型,也没有修改模型参数。

text
Memory
→ 应用保存的数据

上下文
→ 本次请求发送给模型的信息

模型参数
→ 训练形成的内部能力

停止服务器后字典消失,足以证明“小明”并没有永久进入模型本身。

23.5 当前还缺少安全与权限

如果任何客户端只要猜到 abc123 就能访问该会话,就存在越权风险。正式系统必须建立:

text
用户登录身份
  ↓
校验 conversation 是否属于该用户
  ↓
只读取有权限的 messages

因此 conversation_id 解决了“技术上的索引和隔离入口”,但还没有自动解决身份认证与访问控制。


二十四、常见误区总整理

误区 1:访问 URL 就是在访问 main.py

不是。客户端访问的是“HTTP 方法 + 路径”对应的 API。main.py 只是服务器内部代码组织方式。

误区 2:FastAPI 自己就是监听端口的服务器

FastAPI 负责应用规则,Uvicorn 负责运行 ASGI 应用并监听网络。

误区 3:Swagger 更接近服务器端,所以不是 Client

/docs 页面由服务器提供,但页面中的 Swagger UI 在浏览器里发请求时,就是 Client。

误区 4:写了 BaseModel,请求就一定进入业务函数

如果请求体不符合模型,FastAPI 会先返回 422,业务函数可能根本没有执行。

误区 5:模型产品名可以直接作为 API 模型 ID

接口通常需要准确模型 ID。Qwen 这种家族名不一定是有效的 API 参数。

误区 6:Uvicorn reload 后浏览器页面也自动刷新

--reload 只重载后端。已经打开的 /docs 仍需要浏览器刷新。

误区 7:async def 里面的所有操作都会自动异步

同步阻塞库仍然同步阻塞。要真正异步,内部调用也要支持异步并配合 await。

误区 8:await 多次就一定是并发

连续逐行 await 往往仍是按顺序等待。并发需要先安排多个互相独立的异步任务。

误区 9:模型天然记得上一轮 API 请求

应用必须保存并重新发送历史;否则新的请求通常只有新的上下文。

误区 10:只保存用户问题就够了

模型还需要看到自己过去的回答,才能理解后续追问并维持对话一致性。

误区 11:memory.get(id, []) 会自动创建会话

它只返回默认空列表,不会向原字典写入 key。

误区 12:append(history) 与 extend(history) 一样

append 会把整个历史列表作为一个元素;extend 才会把其中每条消息依次接入当前列表。

误区 13:一收到问题就应该立刻写 Memory

如果随后模型调用失败,会留下没有回答的半轮数据。当前版本选择模型成功后再保存完整一问一答。

误区 14:Python 字典已经是数据库

字典只在当前进程内存中存在,重启丢失,多进程也不天然共享。

误区 15:捕获所有异常并 return None 就是完整错误处理

它只能避免异常继续传播,却会丢失错误类型和原因。工程化处理需要分层识别、记录并映射错误。


二十五、本阶段学习成果

25.1 已经能够解释的完整架构

text
Client
  ↓ HTTP Request
Uvicorn
  ↓
FastAPI
  ↓
Pydantic
  ↓
main.py 路由函数
  ├── memory.py
  └── model.py
          ↓
       OpenAI 风格 SDK
          ↓
       Qwen API
  ↓
HTTP Response

25.2 已经真正完成的功能

  • 创建并运行 FastAPI 项目;
  • 使用 Uvicorn 监听本机端口;
  • 编写 GET 和 POST 路由;
  • 使用 Pydantic 定义 Request Body;
  • 使用 Swagger UI 发请求并检查响应;
  • 把模拟模型替换成真实 Qwen 调用;
  • 使用 .env 管理 Key、Base URL 和默认模型;
  • 使用 config.py 集中读取与验证配置;
  • 将 API、模型和 Memory 分到不同模块;
  • 识别连接错误、认证错误、模型不存在和请求验证错误;
  • 理解同步与异步的选择;
  • 使用 messages 构建多轮上下文;
  • 使用 conversation_id 隔离会话;
  • 使用 Python 字典实现服务器端 Memory;
  • 验证“abc123 知道小明,xyz456 不知道小明”。

25.3 这一阶段形成的软件工程思想

关注点分离

text
main.py
→ HTTP / 路由 / 编排

model.py
→ 模型 SDK

memory.py
→ 会话历史

config.py
→ 配置

单一事实来源

默认模型等配置只在一个权威位置维护。

尽早失败

必要配置缺失时,在启动阶段就给出明确错误。

分层排错

根据错误判断请求走到了哪一层,而不是同时乱改所有代码。

数据一致性

模型成功后再提交完整的一问一答,避免半轮历史污染 Memory。

先跑通最小链路,再逐层增加复杂度

text
模拟函数
  ↓
真实模型
  ↓
配置层
  ↓
错误处理
  ↓
多轮 messages
  ↓
服务器 Memory

这条方法以后同样适用于工具调用、RAG、数据库和部署。


二十六、下一阶段学习路线

当前所在位置:

text
Python 基础                         ✅
Python 工程基础                    ✅
HTTP / API                         ✅
FastAPI / Uvicorn                  ✅ 基础完成
Pydantic                           ✅ 基础完成
真实 LLM API 调用                  ✅
.env / config.py                   ✅
基础错误处理                       ✅
messages 多轮对话                  ✅
conversation_id                    ✅
内存级服务器 Memory                ✅ 当前完成
────────────────────────────────────
数据库持久化 Memory                ← 下一阶段
Agent Loop
Tool Calling
RAG
Streaming
日志 / 认证 / 测试
Linux / Docker / 云部署

26.1 下一步:持久化 Memory

把:

python
memory = {}

升级成数据库。

建议先建立三个核心实体:

text
users
→ 谁在使用系统

conversations
→ 用户创建的聊天窗口

messages
→ 每个聊天窗口中的具体消息

关系:

text
User 1
  ├── Conversation A
  │     ├── Message 1
  │     ├── Message 2
  │     └── Message 3
  └── Conversation B
        ├── Message 1
        └── Message 2

数据库阶段需要理解:

  • 表为什么拆成三张,而不是把所有内容塞进一张表;
  • 主键、外键和 ID;
  • 消息顺序与创建时间;
  • 如何根据 conversation_id 查询历史;
  • 如何保证会话属于当前用户;
  • 写入一问一答时如何处理失败和事务;
  • 如何分页,不一次读出无限历史。

26.2 再下一步:Agent Loop

Chatbot 是根据上下文直接回答;Agent 还要判断是否需要行动:

text
用户任务
  ↓
LLM 判断下一步
  ↓
是否需要工具?
  ├── 否 → 最终回答
  └── 是
       ↓
     调用 Tool
       ↓
     得到 Observation
       ↓
     再交给 LLM
       ↓
     继续判断,直到完成

一个最小 Agent 通常包含:

text
LLM + Tools + Memory + Loop

26.3 Tool Calling

需要继续学习:

text
工具函数
工具 Schema
模型如何选择工具
参数如何生成和验证
程序如何真正执行工具
工具结果如何回填 messages
如何防止无限循环和危险调用

26.4 RAG

text
文档
  ↓
切块 Chunk
  ↓
Embedding
  ↓
Vector Database
  ↓
相似度检索
  ↓
把相关内容放入上下文
  ↓
LLM 回答

RAG 与 Memory 有联系,但不是同一件事:

text
对话 Memory
→ 用户与系统过去说过什么

RAG
→ 从外部知识库取回当前问题需要的资料

26.5 Streaming 与生产工程

当前响应是:

text
等待模型全部生成
  ↓
一次性返回

后续 Streaming 会变成:

text
模型产生一部分 Token
  ↓
服务器立即发送一部分
  ↓
客户端逐步显示

之后还需要补齐:

text
logging
认证与权限
CORS
pytest
超时与重试
限流
监控
Linux
Docker
域名与 HTTPS
云服务器部署

附录 A:项目运行与验证清单

A.1 启动前

  • 已创建并激活 .venv;
  • 已安装 fastapi、uvicorn、openai、python-dotenv;
  • .env 位于正确项目目录;
  • .env 变量名和 config.py 完全一致;
  • .gitignore 已忽略 .env;
  • 没有在日志或截图中打印真实 API Key;
  • requirements.txt 已记录依赖。

A.2 启动

bash
uvicorn main:app --reload

检查:

  • 没有配置缺失的 RuntimeError;
  • Uvicorn 正在监听预期端口;
  • 打开 / 能收到“API正在运行”;
  • 打开 /docs 能看到 POST /chat。

A.3 单轮模型测试

json
{
  "conversation_id": "single-test",
  "question": "什么是 Agent?"
}

检查:

  • 状态码是 200;
  • 回答来自真实模型而不是旧的模拟拼接字符串;
  • Response Body 中包含 conversation_id 和 answer。

A.4 多轮测试

第一次:

json
{
  "conversation_id": "abc123",
  "question": "我的名字叫小明"
}

第二次:

json
{
  "conversation_id": "abc123",
  "question": "我叫什么名字?"
}

检查:

  • 第二次能够回答“小明”;
  • 客户端第二次没有重新发送“小明”;
  • 服务器确实从 Memory 中读取了历史。

A.5 会话隔离测试

json
{
  "conversation_id": "xyz456",
  "question": "我叫什么名字?"
}

检查:

  • xyz456 不知道 abc123 中的小明;
  • 没有发生会话串线。

A.6 RAM Memory 边界测试

text
先让 abc123 记住小明
  ↓
Ctrl + C 停止 Uvicorn
  ↓
重新启动服务
  ↓
再次询问 abc123:我叫什么名字?

预期:模型不再知道小明。这证明当前数据只存在于进程内存,还没有持久化。


附录 B:自测题

B.1 FastAPI 与 HTTP

  1. 为什么普通 Python 函数不能让另一台电脑直接通过网络调用?
  2. FastAPI、FastAPI() 和 app 分别是什么?
  3. Uvicorn 与 FastAPI 的职责有什么区别?
  4. 为什么访问 / 不是在访问 main.py 文件?
  5. @app.post("/chat") 建立了什么映射?
  6. Python 字典为什么能被客户端收到为 JSON?
  7. Swagger UI 为什么可以被看作一个 Client?

B.2 Pydantic 与配置

  1. question: str 为什么通常是必填字段?
  2. model: Optional[str] = None 表示什么?
  3. 422 通常发生在哪一层?业务函数是否一定执行?
  4. 为什么 API Key 不能硬编码在 model.py?
  5. .env → config.py → model.py 的配置链如何工作?
  6. 为什么必要配置缺失时适合 raise RuntimeError?
  7. 什么叫 Single Source of Truth?

B.3 错误处理

  1. return None 和裸 raise 的区别是什么?
  2. 为什么 APIConnectionError 与 401 的排查方向不同?
  3. 为什么 model_not_found 反而能证明请求大概率已经到达上游服务器?
  4. 为什么上游返回 404,而自己的 API 可能返回 500?
  5. model.py 与 main.py 应分别承担什么错误处理职责?
  6. 修改 Pydantic 默认值后,为什么要刷新 /docs?

B.4 异步

  1. 并发和并行有什么区别?
  2. await 等待时,Event Loop 可以做什么?
  3. 为什么在 async def 中使用 requests.get() 不会自动变成真正异步?
  4. 为什么连续三次 await 不一定并发?
  5. 同步 SDK 是否一定意味着代码“不合理”?

B.5 Messages 与 Memory

  1. 为什么模型 API 不会天然记住上一轮请求?
  2. 为什么 messages 使用外层 list、内层 dict?
  3. system、user、assistant 分别表示什么?
  4. 为什么需要保存历史 assistant 消息?
  5. conversation_id 解决什么问题?它为什么不完全等于用户 ID?
  6. Dict[str, List[Dict[str, str]]] 应怎样从外到内解释?
  7. memory.get(conversation_id, []) 为什么需要默认空列表?
  8. 为什么 get() 不等于创建了新会话?
  9. append() 和 extend() 在构建消息列表时有什么区别?
  10. 为什么应该在模型成功后再保存完整一问一答?
  11. 为什么 abc123 知道小明,而 xyz456 不知道?
  12. 为什么 memory = {} 是服务器端 Memory,却不是持久化 Memory?

最终总结

这一阶段真正建立起来的,不只是几个 FastAPI 写法,而是一条完整的软件理解链:

text
Python 函数
  ↓ 通过 FastAPI 暴露
HTTP API
  ↓ 通过 Uvicorn 运行
可被 Client 调用的服务
  ↓ 接入模型 SDK
真实 LLM Service
  ↓ 传递完整 messages
多轮 Chat
  ↓ 使用 conversation_id 查历史
服务器端 Memory

最关键的三个认识是:

  1. 模型能力只是系统的一层。 FastAPI、配置、错误处理、Memory 和数据库共同决定应用是否可靠。
  2. 模型看起来有记忆,不等于模型永久记住了用户。 当前机制是应用保存历史,并在下一次请求中重新发送。
  3. 会运行只是起点,知道数据如何流动、错误停在哪一层、为什么这样拆模块,才是真正进入 Agent 工程。

当前项目已经是一个简化版 ChatGPT 后端的雏形。下一步应把进程内字典升级为数据库持久化,再在这个基础上加入 Agent Loop、Tools 和 RAG。