完整学习档案:保留原有章节、代码、失败示例、排错过程和练习。代码反映记录时的环境,不代表每段都是最终版本,也不代表本次发布重新运行了全部示例。请在练习项目中使用自己的环境配置,勿填入生产密钥。
学习阶段:FastAPI → Uvicorn → HTTP API → Pydantic → 真实 LLM 调用 → 配置与错误处理 →
messages多轮对话 →conversation_id→ 服务器端 Memory这份笔记不是知识点摘要,而是一次完整的学习复盘。重点记录:为什么需要这个设计、数据怎样流动、代码为什么这样写、实际遇到了什么问题、问题是怎样定位的,以及这些知识将怎样继续演进成真正的 AI Agent。
一、这一阶段在整条 Agent 路线中的位置
在进入 FastAPI 之前,我已经完成了 Python 最少必要基础,例如:
变量、字符串、数字、布尔值
list、dict、if、for
函数、参数、默认参数、return
import、module、package
pip、venv、requirements.txt
文件、JSON、异常、环境变量
HTTP、requests
class、object、Type Hint
async、await、Event Loop这些知识不是互相孤立的。到了 FastAPI 项目,它们开始第一次在同一个程序里汇合:
Python 基础
↓
Python 工程结构
↓
HTTP / API
↓
FastAPI / Uvicorn
↓
真实 LLM API
↓
多轮对话 messages
↓
服务器端 Memory这一阶段最重要的变化是:
以前:我在本地运行一个 Python 程序
现在:我让其他客户端通过 HTTP 调用我的 Python 能力再进一步:
以前:每次只把一个问题发给模型
现在:服务器能根据 conversation_id 找回历史,形成多轮上下文这意味着项目已经从“调用一次模型的脚本”,发展成了一个最小可用的聊天后端。
二、为什么 AI 应用需要后端服务
2.1 一个普通 Python 函数的边界
例如:
def add(a, b):
return a + b这个函数只能在当前 Python 程序中被调用。另一台电脑、一个网页或手机 App,并不能仅仅因为这个函数存在,就直接通过网络使用它。
普通函数缺少几样东西:
- 没有网络监听能力;
- 没有公开的 URL;
- 不理解 HTTP Request;
- 不会生成 HTTP Response;
- 不知道怎样把 JSON 转成 Python 数据;
- 不知道怎样把 Python 返回值转成 JSON。
因此,需要在 Python 函数外面增加一层 Web API:
客户端
↓ HTTP Request
Web API
↓ 调用
Python 函数
↓ 返回结果
Web API
↓ HTTP Response
客户端FastAPI 的价值,就是帮助我们构建这层 Web API。
2.2 “能调用模型”不等于“已经做成 AI 应用”
最简单的大模型脚本可能是:
问题
↓
调用模型
↓
打印回答但真实产品往往是:
用户
↓
网页 / 手机 / 其他客户端
↓
HTTP API
↓
身份、参数和业务规则
↓
对话历史 / 数据库 / 工具
↓
大模型
↓
整理结果
↓
HTTP Response
↓
用户界面所以,大模型只是系统中的一个能力组件。FastAPI 后端负责把客户端、业务逻辑、模型、Memory、数据库和工具组织起来。
2.3 在 Agent 架构中的位置
以后完整的 Agent 服务会更像:
Client
↓
FastAPI
↓
Agent Loop
├── LLM
├── Memory
├── Tools
└── RAG / Database
↓
Response本阶段还没有完成真正的 Agent Loop,但已经搭好了它未来对外提供服务的外壳。
三、HTTP、Client 与 Server
3.1 Client 是谁
Client 是主动发起请求的一方。它可以是:
- 浏览器;
- 网页前端;
- 手机 App;
- Python 的
requests; - Swagger UI;
- 另一台服务器。
例如:
import requests
response = requests.get("http://127.0.0.1:8000/")
print(response.json())这里的 requests 工作在客户端一侧。
3.2 Server 是谁
Server 是监听网络请求、执行处理逻辑并返回响应的一方。
在当前项目中:
Uvicorn
→ 启动并监听端口
FastAPI
→ 匹配路由、验证数据、调用函数、构造响应
main.py / model.py / memory.py
→ 实际业务代码3.3 本机地址 127.0.0.1
当前访问:
http://127.0.0.1:8000可以拆成:
http
→ 使用 HTTP 协议
127.0.0.1
→ 当前这台电脑,也叫 localhost
8000
→ Uvicorn 正在监听的端口它不是公网地址。一般只有当前电脑自己能访问,因此还不等于“已经部署到互联网”。
四、FastAPI 与 Uvicorn 各自负责什么
这是学习初期非常容易混淆的一组概念。
4.1 FastAPI 是 Web 框架
FastAPI 主要负责:
- 定义路由;
- 规定请求数据结构;
- 解析和验证请求;
- 调用 Python 处理函数;
- 把返回值转换成 HTTP Response;
- 生成 OpenAPI 文档;
- 提供 Swagger UI。
例如:
from fastapi import FastAPI
app = FastAPI()逐层理解:
FastAPI
→ 一个 class
FastAPI()
→ 调用 class 创建对象
app
→ 创建出来的 FastAPI 应用对象路由、文档以及请求处理规则都注册在 app 上。
4.2 Uvicorn 是 ASGI Web Server
仅仅写好:
app = FastAPI()并不会自动开始监听网络端口。需要 Uvicorn 真正运行这个应用:
uvicorn main:app --reload这条命令可以拆成:
uvicorn
→ 启动 Web Server
main
→ 找 main.py 这个 Python 模块
app
→ 找模块中的 FastAPI 应用对象
--reload
→ 开发期间监测代码变化,并自动重载后端程序完整关系是:
网络请求
↓
Uvicorn 监听到请求
↓
交给 FastAPI app
↓
FastAPI 匹配路由并执行 Python 函数4.3 一个常见误区
错误理解:
FastAPI 就是服务器,写完代码以后会自己监听 8000 端口。
正确理解:
FastAPI
→ 描述 Web 应用怎样处理请求
Uvicorn
→ 真正运行这个应用并监听网络五、路由:把网络地址映射到 Python 函数
5.1 根路由示例
@app.get("/")
def root():
return {
"message": "API正在运行"
}@app.get("/") 当前可以理解为一种“注册关系”:
如果收到 GET /
↓
就执行 root()因此,访问:
http://127.0.0.1:8000/不是在访问 main.py 文件,而是在访问 FastAPI 应用中注册在 / 路径上的 API。
客户端甚至不需要知道服务器代码文件叫:
main.py、server.py,还是 abc.py客户端只关心:
协议 + IP/域名 + 端口 + 路径 + HTTP 方法5.2 /chat 路由
@app.post("/chat")
def chat(request: ChatRequest):
...它建立的是:
POST /chat
↓
chat(request)同一个路径配合不同 HTTP 方法,也可以表达不同操作。因此“路径”和“方法”需要放在一起理解。
5.3 Python 装饰器在这里的直观作用
@app.post("/chat") 是装饰器语法。当前阶段不需要先深挖装饰器底层,只要牢牢记住它在这里的工程作用:
把一个普通 Python 函数登记为可以通过 HTTP 访问的路径操作函数。
六、Request、Response、GET 与 POST
6.1 Request 是什么
Request 是客户端发给服务器的一份请求。它通常包含:
HTTP 方法
URL / Path
Headers
Query Parameters
Request Body例如聊天请求:
{
"conversation_id": "abc123",
"question": "什么是 Agent?"
}6.2 Response 是什么
Response 是服务器处理以后返回给客户端的结果,通常包含:
HTTP 状态码
Response Headers
Response Body例如:
{
"conversation_id": "abc123",
"answer": "Agent 是……"
}当 FastAPI 路由返回 Python 字典时,FastAPI 会把它序列化成 JSON,并设置相应的响应类型。
6.3 GET 与 POST
可以先这样区分:
GET
一般用于读取资源或获取信息。
@app.get("/")
def root():
...POST
一般用于向服务器提交数据,让服务器执行某个操作。
@app.post("/chat")
def chat(request: ChatRequest):
...聊天问题通常内容较长、结构也会不断扩展,因此适合放在 JSON Request Body 中,而不是全部塞进 URL。
6.4 Path、Query 与 Body
三者的直观区别:
Path
→ 表示访问哪个资源,例如 /conversations/abc123
Query
→ 表示额外筛选或选项,例如 ?limit=20
Body
→ 表示提交的一整份结构化数据,例如问题、模型和会话 ID本项目主要使用 Request Body:
{
"conversation_id": "abc123",
"question": "我叫什么名字?",
"model": "qwen-plus"
}七、Pydantic:让请求数据有明确结构
7.1 为什么不直接接收任意字典
如果服务器随便接收一份字典,就需要自己逐项检查:
有没有 question?
question 是不是字符串?
有没有 conversation_id?
model 可不可以不传?Pydantic 的 BaseModel 可以把这些要求正式写进代码:
from typing import Optional
from pydantic import BaseModel
class ChatRequest(BaseModel):
conversation_id: str
question: str
model: Optional[str] = None这不是“大模型的 Model”,而是“数据模型”。它描述请求应该长什么样。
7.2 必填字段与可选字段
question: str表示:
- 字段名是
question; - 期望类型是字符串;
- 没有默认值,所以通常必须提供。
model: Optional[str] = None表示:
model可以是字符串;- 也可以是
None; - 客户端不传时,值就是
None。
后续可在模型层使用默认模型:
if model is None:
model = DEFAULT_MODEL这形成了:
客户端不指定模型
↓
使用项目默认模型
客户端主动指定模型
↓
使用本次请求给出的模型7.3 422 从哪里来
如果 question 没有默认值,而客户端没有传,或者传入的数据类型不符合要求,FastAPI 会在进入 chat() 之前完成验证并返回 422 Unprocessable Entity。
数据流是:
Request Body
↓
Pydantic 验证
├── 成功 → 创建 ChatRequest 对象 → 进入 chat()
└── 失败 → FastAPI 直接返回 422所以,看到 422 时首先检查请求体字段,而不是立刻怀疑模型 API。
7.4 为什么 Pydantic 还能帮助 Swagger
完整链路:
Python Type Hint + Pydantic BaseModel + FastAPI 路由
↓
FastAPI 生成 OpenAPI Schema
↓
Swagger UI 读取 Schema
↓
自动显示字段、类型、必填项和请求示例八、Swagger 与 OpenAPI:自动生成的测试客户端
8.1 /docs 是什么
启动服务后访问:
http://127.0.0.1:8000/docs会看到 Swagger UI。它可以:
- 查看所有路由;
- 查看参数和 Request Body;
- 点击
Try it out; - 直接发送 HTTP 请求;
- 查看状态码、Response Body 和 Headers。
8.2 Swagger 到底属于客户端还是服务器
这个问题要分两层:
FastAPI 服务器
→ 提供 /docs 页面和 /openapi.json
浏览器打开 /docs 后
→ Swagger UI 在浏览器中运行
→ 主动向 API 发送请求
→ 此时扮演 Client所以可以把 Swagger UI 理解成:
FastAPI 自动提供的一个网页 API 测试客户端。
8.3 /openapi.json 的作用
FastAPI 会根据路由和 Pydantic 模型生成 API 描述:
Python 代码
↓
FastAPI
↓
/openapi.json
↓
Swagger /docsSwagger 并不是猜测请求中有哪些字段,而是读取 OpenAPI Schema 后生成界面。
8.4 实际遇到的“旧页面”问题
修改后端代码以后,--reload 只负责重新加载服务器代码,并不会强制浏览器刷新已经打开的 /docs 页面。
当时实际发生的是:
main.py 默认模型已经改成 qwen-plus
↓
Uvicorn 已经 reload
↓
浏览器里的旧 Swagger 页面没有刷新
↓
Request Body 仍然保留 model="Qwen"
↓
点击 Execute
↓
上游返回 404 model_not_found因此修改以下内容后,应该刷新 /docs:
路由
Pydantic Model
字段类型
默认值
Request Body 结构如果仍有疑问,可直接打开 /openapi.json 检查后端当前真正生成的 Schema。
九、搭建第一个可运行的 FastAPI 项目
9.1 第一版项目结构
fastapi_agent_demo/
│
├── .venv/
├── main.py
├── model.py
└── requirements.txt职责:
main.py
→ HTTP、FastAPI、路由、请求和响应
model.py
→ 模型调用逻辑
requirements.txt
→ 第三方依赖清单
.venv/
→ 当前项目独立的 Python 运行环境9.2 创建环境与安装依赖
python -m venv .venv
.venv\Scripts\activate
pip install fastapi uvicorn
pip freeze > requirements.txt理解:
.venv
→ 真正安装出来的环境,一般不上传
requirements.txt
→ 重建环境所需的清单,应该保留在项目中9.3 先写一个假的模型层
model.py:
def call_model(
question: str,
model: str
) -> str:
answer = "使用 " + model + " 回答:" + question
return answer它没有访问真实模型,只是在模拟“模型层的输入和输出”。
为什么先这样做?因为调试时应该分层验证:
第一步:先证明 HTTP 服务可以跑通
第二步:再接真实外部模型如果一开始同时加入 FastAPI、网络、Key、SDK 和模型,就很难判断错误究竟在哪一层。
9.4 第一版 main.py
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 第一次完整数据流
Swagger UI
↓ POST /chat
Uvicorn
↓
FastAPI
↓
ChatRequest 解析 Request Body
↓
chat()
↓
call_model()
↓
模拟回答
↓
Python dict
↓
FastAPI 转成 JSON Response
↓
Swagger 显示结果这一步虽然模型是假的,但 HTTP 服务已经是真实、完整的。
十、从模拟回答升级到真实大模型
10.1 变化发生在哪里
FastAPI 外壳不需要推倒重来。主要替换的是 model.py:
以前:
call_model() 拼接字符串
以后:
call_model() 调用真实模型 SDK这证明分层设计的价值:
main.py 的 HTTP 逻辑保持稳定
model.py 内部替换具体模型实现10.2 使用兼容接口的 SDK
学习项目使用 OpenAI 风格 SDK 调用 Qwen 兼容接口。核心结构如下:
from openai import OpenAI
client = OpenAI(
api_key=API_KEY,
base_url=BASE_URL
)然后请求模型:
completion = client.chat.completions.create(
model=model,
messages=[
{
"role": "system",
"content": "你是一个有帮助的AI助手"
},
{
"role": "user",
"content": question
}
]
)最后取出回答:
answer = completion.choices[0].message.content10.3 为什么响应结构不是直接一个字符串
模型服务返回的不只是文本,还可能包含:
多个候选结果 choices
消息角色
文本内容
结束原因
Token 使用量
请求元数据所以需要从响应对象中逐层取值:
completion
↓
choices[0]
↓
message
↓
content10.4 “Qwen”与“qwen-plus”不是同一个概念
Qwen
→ 人类使用的产品或模型家族名称
qwen-plus
→ API 能识别的具体模型 ID 示例API 不会根据宽泛的品牌名自动猜测具体模型。必须传接口支持、账号有权访问的准确模型 ID。
十一、环境变量与配置分层
11.1 为什么不能把 Key 写进代码
不应该这样:
API_KEY = "真实密钥"原因包括:
- 代码上传 GitHub 时可能泄露;
- 截图或分享代码时可能泄露;
- 本地、测试和生产环境可能使用不同配置;
- 修改配置还要改动业务代码;
- 一旦泄露,别人可能盗用额度。
更合理的结构:
.env
→ 保存本机秘密和环境配置
config.py
→ 读取、检查并向 Python 模块提供配置
model.py
→ 使用配置创建模型客户端11.2 .env
示例:
LLM_API_KEY=你的真实APIKey
LLM_BASE_URL=模型服务兼容接口地址
LLM_MODEL=qwen-plus注意:真实 Key 不应写进笔记、截图或聊天记录。
11.3 .gitignore
至少应忽略:
.env
.venv/
__pycache__/其中:
.env
→ 因为包含秘密
.venv/
→ 因为体积大且可以根据依赖清单重建
__pycache__/
→ Python 自动生成的缓存11.4 config.py
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,模型调用一定无法正常工作。仅仅:
print("API Key读取失败")会让程序继续运行,直到用户发请求时才以更模糊的方式失败。
更合理的是:
启动时读取配置
↓
发现必要配置缺失
↓
立即停止并给出明确错误这叫“尽早失败”:让错误在最接近原因的地方暴露。
11.6 RuntimeError 与 HTTPException
RuntimeError
→ 程序运行所需的基础条件不成立
→ 例如启动时没有 API Key
HTTPException
→ 某一次 HTTP 请求需要返回错误响应
→ 例如上游模型暂时不可用11.7 单一事实来源
早期版本可能在多处重复写:
.env → qwen-plus
main.py → qwen-plus
model.py → qwen-plus一旦只改其中一处,就会不一致。
整理后:
.env 中的 LLM_MODEL
↓
config.py 中的 DEFAULT_MODEL
↓
model.py 使用这就是 Single Source of Truth:同一项配置尽量只维护一个权威来源。
11.8 默认配置 + 允许覆盖
def call_model(
messages,
model: Optional[str] = None
):
if model is None:
model = DEFAULT_MODEL它同时满足两种需求:
普通请求
→ 不传 model,使用默认值
特殊请求
→ 主动传 model,覆盖本次默认值十二、真实 LLM 请求的数据流
真实模型接入后,请求链变成:
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还存在一条配置链:
.env
↓ load_dotenv()
config.py
↓ API_KEY / BASE_URL / DEFAULT_MODEL
model.py
↓ 创建 client 并调用模型理解这两条链以后,排错就不再是“所有文件一起乱改”,而是判断:
请求链在哪一层停止?
配置链是否把正确值传到了模型层?十三、错误处理:从吞掉错误到分层传递
13.1 第一版错误处理的问题
早期代码:
try:
...
except Exception as e:
print("模型调用失败:", repr(e))
return None然后 main.py:
if answer is None:
raise HTTPException(
status_code=500,
detail="模型调用失败"
)它能防止程序直接崩溃,但丢失了大量信息:
究竟是连接失败?
认证失败?
模型不存在?
频率限制?
上游服务故障?全部被压成:
None → 500 模型调用失败13.2 为什么 return None 会“吞掉”错误
异常本来携带类型、状态码和详细信息。转换成 None 后,上层只知道“没有结果”,却不知道原因。
上游明确错误
↓ except Exception
return None
↓
错误身份丢失
↓
main.py 只能统一返回 50013.3 更合理的职责划分
model.py
→ 了解模型 SDK
→ 识别或保留底层异常
main.py
→ 了解 HTTP API
→ 决定客户端应该收到什么状态码和说明简化写法可以让模型层记录后重新抛出:
def call_model(...):
try:
...
return answer
except Exception as error:
print("模型调用失败:", repr(error))
raise这里的裸 raise 表示:
把当前正在处理的原始异常继续抛给调用者。
它与 return None 完全不同。
13.4 在 API 层转换成 HTTP 错误
概念示例:
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:
客户端
↓
自己的 FastAPI /chat
↓
模型服务 API上游模型可能返回 404,但如果自己的代码把异常转换成了统一 500,那么:
模型服务: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
数据流定位:
POST /chat 已经进入 FastAPI ✅
chat() 已经调用 call_model() ✅
SDK 尝试连接外部模型服务 ✅
没有正常建立连接或取得 HTTP 响应 ❌因此它和 401、404 不在同一层。
排查顺序应包括:
BASE_URL是否为None;.env的变量名是否和os.getenv()完全一致;base_url=BASE_URL是否误写成字符串base_url="BASE_URL";- 地址协议、路径和地域配置是否正确;
- 网络、代理、VPN、防火墙是否影响连接;
- 修改
config.py后是否正确导入。
可临时打印非敏感信息:
print("BASE_URL =", BASE_URL)
print("DEFAULT_MODEL =", DEFAULT_MODEL)但不要打印:
print(API_KEY)14.2 404 model_not_found
这个错误反而证明:
电脑网络 → 模型服务器通常已经走通,因为服务器明确收到了请求并返回了结构化错误。
当时原因是传入:
{
"model": "Qwen"
}而接口需要具体模型 ID,例如学习时使用的:
{
"model": "qwen-plus"
}排错原则:错误信息已经明确指向模型名时,不要又同时乱改网络、Key 和 FastAPI 路由。
14.3 后端已更新,Swagger 仍发送旧值
原因不是后端没重载,而是浏览器 /docs 页面仍保留旧 Schema 或旧输入内容。
解决思路:
确认 Uvicorn 已 reload
↓
刷新 /docs
↓
重新检查 Request Body
↓
必要时查看 /openapi.json14.4 调试的通用原则:从最近变动处开始
如果程序原来能运行,而把配置读取从 model.py 移到 config.py 后立刻报错,第一优先级应该检查新增加的配置链:
.env
↓
config.py
↓
model.py而不是一开始就怀疑所有可能原因。
14.5 根据错误判断“请求走到了哪里”
422
→ 多数还没进入 chat(),请求验证阶段失败
本地 404
→ 可能没有匹配到 FastAPI 路由
APIConnectionError
→ 已进入业务函数,但连接上游失败
上游 401
→ 通常已经连接上服务器,认证没有通过
上游 404 model_not_found
→ 通常已经连接成功,但模型标识或权限有问题
200
→ 整条请求与响应链成功完成这是一种非常重要的后端调试思维:
不要只看“失败了”,要根据错误判断请求已经成功经过了哪些层,停止在哪一层。
十五、同步、异步与 Agent 的等待时间
15.1 同步函数与异步函数
同步路由:
@app.post("/chat")
def chat(...):
...异步路由:
@app.post("/chat")
async def chat(...):
...异步函数中可以等待真正的异步操作:
answer = await call_model(...)15.2 为什么 Agent 特别需要关注异步
Agent 经常需要等待:
大模型 API
网页搜索
天气 API
数据库查询
远程工具
文件服务这些通常属于 I/O 等待型任务。等待一个外部服务返回时,CPU 不一定需要一直空耗在那里。
用户 A 等待模型
↓ 暂时让出执行机会
Event Loop 推进用户 B
↓ B 也等待
Event Loop 推进用户 C异步提升的是服务在等待型任务下的并发处理能力,而不是让模型本身推理得更快。
15.3 并发与并行
并发 Concurrency
→ 多个任务都在推进
→ 一个任务等待时切换到另一个任务
并行 Parallelism
→ 同一时刻由多个执行单元真正执行多个任务二者不是同一个概念。
15.4 async def 不是异步魔法
错误示例:
async def get_data():
response = requests.get(url)虽然函数写成 async def,但 requests.get() 仍是同步阻塞操作。它不会自动变成可被 Event Loop 高效调度的异步网络请求。
合理选择:
同步 SDK / 同步 HTTP 库
→ 可以使用同步函数和普通 def 路由
异步 SDK / 异步 HTTP 客户端
→ 使用 async def + await不是所有代码加上 async 就更高级。关键是内部调用的库是否真正支持异步,以及任务是否适合异步。
15.5 连续 await 不一定并发
a = await task_a()
b = await task_b()
c = await task_c()这表示:完成 A 后才开始 B,完成 B 后才开始 C。
若三个任务互相独立,需要一起推进,可以使用类似:
results = await asyncio.gather(
task_a(),
task_b(),
task_c()
)但只有在任务确实独立、库真正异步并且业务允许并发时才这样设计。
十六、从单轮调用进入多轮对话
16.1 单轮调用是什么
最初发送给模型的内容:
messages = [
{
"role": "system",
"content": "你是一个有帮助的AI助手"
},
{
"role": "user",
"content": question
}
]每次请求只有当前问题。第一次:
我的名字叫小明第二次如果只发送:
我叫什么名字?模型通常并不知道答案,因为第二次请求中没有“小明”这个信息。
16.2 一个关键认识:模型 API 本身不会自动记住上一次请求
常见误解:
第一次告诉了模型名字,第二次调用同一个模型,它自然应该记得。
实际机制:
第一次 HTTP 请求结束
↓
第二次是新的 HTTP 请求
↓
如果应用没有重新提供历史
↓
模型只看到本次输入看起来“有记忆”的系统,本质通常是:
应用保存历史
+
每次重新把相关历史发送给模型
=
模型能够根据上下文回答16.3 多轮 messages
messages = [
{
"role": "system",
"content": "你是一个有帮助的AI助手"
},
{
"role": "user",
"content": "我的名字叫小明"
},
{
"role": "assistant",
"content": "好的,我记住了,你叫小明"
},
{
"role": "user",
"content": "我叫什么名字?"
}
]模型现在能回答“小明”,不是因为它在上一次请求后改变了参数或永久存储了用户,而是因为当前请求再次把历史放进了上下文。
16.4 为什么使用 list + dict
外层是列表:
list
→ 保存多条消息
→ 保留先后顺序每条消息是字典:
dict
→ 描述一条消息的多个字段
→ role 表示谁说的
→ content 表示说了什么因此:
messages
↓ list
一组按时间排列的消息
↓ dict
每条消息都有 role 和 content消息顺序非常重要,因为对话的意义依赖上下文顺序。
十七、system、user、assistant 三种角色
17.1 system
{
"role": "system",
"content": "你是一个耐心的 Python 老师"
}它主要描述模型在本次上下文中的行为、身份、边界或输出要求。
它不是普通用户消息,而更接近应用提供的全局指导。
17.2 user
{
"role": "user",
"content": "什么是 FastAPI?"
}表示真实用户输入的问题、任务或补充信息。
17.3 assistant
{
"role": "assistant",
"content": "FastAPI 是一个 Python Web 框架……"
}表示模型过去已经给出的回答。
为什么连模型自己的历史回答也要保存?因为下一轮中,模型需要知道:
- 自己之前回答过什么;
- 用户说的“你刚才那一点”指什么;
- 当前对话已经形成了哪些约定;
- 避免前后矛盾或重复。
17.4 三种角色形成的对话结构
system
→ 规定总体行为
user
→ 提出任务
assistant
→ 产生回答
user
→ 根据回答继续追问这比把所有文字拼成一个大字符串更清楚,也让模型知道每段内容来自谁。
十八、为什么需要 conversation_id
18.1 只保存一个全局消息列表会怎样
如果服务器只有:
messages = []所有用户都向同一份列表写入历史,就可能发生:
用户 A 告诉模型自己的名字
↓
用户 B 发来问题
↓
用户 B 的上下文里混入用户 A 的历史这不仅会答错,还是严重的数据隔离问题。
18.2 conversation_id 是会话的索引
请求:
{
"conversation_id": "abc123",
"question": "我叫什么名字?"
}服务器可以理解为:
先找到编号为 abc123 的聊天窗口
↓
读取它自己的历史
↓
加入当前问题
↓
调用模型它像不同聊天窗口的编号:
abc123
→ 一组独立历史
xyz456
→ 另一组独立历史18.3 会话 ID 不完全等于用户 ID
同一个用户可以创建多个聊天窗口,因此通常:
一个 user
↓
可以拥有多个 conversation
↓
每个 conversation 拥有多条 message本阶段尚未加入用户系统,只使用 conversation_id 完成最小会话隔离。
18.4 谁应该生成 conversation_id
学习 Demo 中可以手动填写 abc123,方便观察机制。正式系统中通常由服务器创建唯一 ID,再交给客户端保存和继续使用。
如果长期手动随便写,会产生:
- ID 冲突;
- 拼写错误导致历史丢失;
- 用户可能访问不属于自己的会话;
- 无法可靠建立用户与会话的权限关系。
十九、用 Python 字典实现服务器端 Memory
19.1 数据结构
from typing import Dict, List
memory: Dict[str, List[Dict[str, str]]] = {}从外到内拆解:
memory
→ 整个服务器当前保存的会话集合
Dict 的 key:str
→ conversation_id,例如 abc123
Dict 的 value:List[...]
→ 这个会话的多条历史消息
List 中的每个 Dict[str, str]
→ 一条消息,例如 role + content运行一段时间后可能变成:
memory = {
"abc123": [
{
"role": "user",
"content": "我的名字叫小明"
},
{
"role": "assistant",
"content": "好的,我记住了,你叫小明"
}
],
"xyz456": [
{
"role": "user",
"content": "你好"
},
{
"role": "assistant",
"content": "你好!有什么可以帮助你?"
}
]
}19.2 get_messages()
def get_messages(
conversation_id: str
) -> List[Dict[str, str]]:
return memory.get(
conversation_id,
[]
)memory.get(conversation_id, []) 表示:
如果 key 存在
→ 返回对应历史列表
如果 key 不存在
→ 返回默认空列表 []为什么需要 []?新会话第一次请求时还没有历史,但这不是错误,而是“历史为空”。返回空列表后,后续仍可以正常:
messages.extend(history)19.3 一个重要细节:get() 不会创建 key
history = memory.get("new-id", [])得到空列表,不等于 memory 中已经出现了 "new-id"。
get()
→ 读取
→ 找不到时返回默认值
→ 不负责把默认值写回字典真正创建会话要在写入逻辑中完成。
19.4 add_message()
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
}
)逐步执行:
收到 conversation_id
↓
检查 memory 中是否已经有这个会话
↓ 没有
创建一个空消息列表
↓
把新消息 append 到该会话列表末尾返回类型是 None,因为这个函数的主要作用不是计算一个结果,而是修改服务器内存中的状态。
19.5 append() 与 extend()
append():加入一个整体元素
messages.append(
{
"role": "user",
"content": question
}
)表示把“一条消息字典”作为一个元素放到列表末尾。
extend():加入另一个列表中的所有元素
messages.extend(history)如果 history 有两条消息,extend() 会把这两条分别接到 messages 后面。
区别:
messages.append(history)会把整个历史列表当成一个元素,形成嵌套列表,这不符合模型 API 期望的消息结构。
19.6 为什么先构建临时 messages
每次请求中,可以创建一份本次发送给模型的临时上下文:
messages = [
{
"role": "system",
"content": "你是一个有帮助的AI助手"
}
]
messages.extend(history)
messages.append(
{
"role": "user",
"content": request.question
}
)这样做有几个好处:
system指令可以在每次调用时统一添加;- 历史 Memory 只保存真正的
user/assistant对话; - 本次问题可以先进入临时上下文,而不是立刻污染长期状态;
- 如果模型失败,Memory 可以保持在上一轮成功状态。
19.7 为什么模型成功以后再保存两条消息
错误顺序:
一收到用户问题就保存 user 消息
↓
调用模型失败
↓
没有 assistant 回答
↓
Memory 留下一条未完成对话这种历史会让后续上下文与真实交互状态不一致。
当前更稳妥的顺序:
读取历史
↓
构建临时 messages
↓
调用模型
↓ 成功
保存 user 消息
↓
保存 assistant 消息核心思想是:
只有这一轮对话成功完成后,才把完整的一问一答提交到 Memory。
这是最基础的数据一致性意识。未来使用数据库时,会进一步学习事务、失败回滚和消息状态。
二十、最终版代码与逐层解释
下面代码用于还原这一阶段最终形成的结构。不同 SDK 版本的具体异常类或接口细节可能需要按实际环境调整,但职责划分和数据流不变。
20.1 项目结构
fastapi_agent_demo/
│
├── .venv/
├── .env
├── .gitignore
├── requirements.txt
│
├── config.py
├── memory.py
├── model.py
└── main.py职责:
.env
→ 保存原始环境配置和秘密
config.py
→ 读取并验证配置
model.py
→ 调用大模型
memory.py
→ 按 conversation_id 保存和读取历史
main.py
→ HTTP 路由、请求验证、编排调用、错误响应
Uvicorn
→ 运行整个 FastAPI 应用20.2 .env
LLM_API_KEY=你的真实APIKey
LLM_BASE_URL=你的模型服务兼容接口地址
LLM_MODEL=qwen-plus20.3 config.py
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
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 类型别名,只是为了让嵌套类型更容易阅读:
Message
→ 一条消息 Dict[str, str]
Memory
→ conversation_id 到消息列表的映射20.5 model.py
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相比单轮版本,最大的变化是:
以前 call_model(question)
→ model.py 自己临时构建两条消息
现在 call_model(messages)
→ 上层已经把 system、历史和当前问题组织好
→ model.py 只负责把 messages 发给模型这种分工让模型层不必知道历史来自内存、数据库还是其他地方。
20.6 main.py
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 没有亲自实现模型调用,也没有亲自维护字典细节,而是负责组织顺序:
接收请求
↓
读取 Memory
↓
构建本次上下文
↓
调用模型
↓
成功后保存 Memory
↓
返回 Response这就是关注点分离:每个模块承担清晰职责,通过函数协作。
20.8 为什么 system 没有保存进 Memory
当前实现每次都会主动把 system 放在临时 messages 的第一项,因此没有必要在每轮都重复写入历史。
当前 Memory 保存:
user 消息
assistant 消息调用前再组合:
system + history + current user未来如果不同会话需要不同 system prompt,可以把它保存为会话配置,而不是每轮重复保存为普通消息。
二十一、一次完整请求到底经历了什么
以请求为例:
{
"conversation_id": "abc123",
"question": "我叫什么名字?"
}第 1 步:客户端发送 HTTP Request
POST /chat
Content-Type: application/json
Body: conversation_id + question第 2 步:Uvicorn 接收
Uvicorn 正在监听 127.0.0.1:8000,收到请求后把它交给 FastAPI 应用。
第 3 步:FastAPI 匹配路由
POST /chat
↓
@app.post("/chat")
↓
chat()第 4 步:Pydantic 验证
请求体被解析成:
request.conversation_id
request.question
request.model第 5 步:读取历史
history = get_messages("abc123")假设得到:
[
{
"role": "user",
"content": "我的名字叫小明"
},
{
"role": "assistant",
"content": "好的,我记住了,你叫小明"
}
]第 6 步:构建临时上下文
system
↓
历史 user
↓
历史 assistant
↓
本次 user最终模型看到的内容相当于:
[
{
"role": "system",
"content": "你是一个有帮助的AI助手"
},
{
"role": "user",
"content": "我的名字叫小明"
},
{
"role": "assistant",
"content": "好的,我记住了,你叫小明"
},
{
"role": "user",
"content": "我叫什么名字?"
}
]第 7 步:调用模型
model.py 把完整 messages 发送给 Qwen。模型根据当前上下文回答:
你叫小明。第 8 步:提交 Memory
模型成功后保存:
{"role": "user", "content": "我叫什么名字?"}
{"role": "assistant", "content": "你叫小明。"}第 9 步:返回 HTTP Response
{
"conversation_id": "abc123",
"answer": "你叫小明。"
}这条链可以压缩为:
Client
↓
Uvicorn
↓
FastAPI
↓
conversation_id
↓
memory.py 查历史
↓
组合 messages
↓
model.py
↓
Qwen
↓
保存完整一问一答
↓
Response二十二、真实测试:记住小明与会话隔离
22.1 第一次请求:告诉 abc123 名字
{
"conversation_id": "abc123",
"question": "我的名字叫小明"
}成功后,服务器 Memory 中形成:
memory["abc123"] = [
{
"role": "user",
"content": "我的名字叫小明"
},
{
"role": "assistant",
"content": "好的,我记住了,你叫小明"
}
]22.2 第二次请求:仍使用 abc123
{
"conversation_id": "abc123",
"question": "我叫什么名字?"
}客户端没有再次传“小明”,但服务器通过 abc123 找到了历史,并把历史重新发给模型,所以得到类似:
{
"conversation_id": "abc123",
"answer": "你叫小明。"
}这证明:
历史保存成功
历史读取成功
上下文拼接成功
多轮对话成功22.3 第三次请求:换成 xyz456
{
"conversation_id": "xyz456",
"question": "我叫什么名字?"
}服务器执行:
get_messages("xyz456")得到:
[]模型没有看到 abc123 的历史,所以会回答类似:
我还不知道你的名字,你可以告诉我。这证明:
abc123
→ 知道小明
xyz456
→ 不知道小明也就是:
会话隔离成功。
22.4 这个实验真正验证了什么
它同时验证了四件事:
- Memory 能保存消息;
- 后续请求能取回历史;
- 模型可以根据重新发送的历史回答;
- 不同
conversation_id不会共用同一份历史。
二十三、当前 Memory 的边界
23.1 它是服务器端 Memory,但不是持久化 Memory
memory = {}存在于 Python 进程的 RAM 中:
Uvicorn 正在运行
→ memory 存在
Ctrl + C 停止 Uvicorn
→ Python 进程结束
→ RAM 数据消失
重新启动
→ 再次执行 memory = {}
→ 所有会话从空字典开始因此,当前可以准确称为:
进程内、内存级、服务器端会话 Memory。
不能称为长期持久化记忆。
23.2 多 Worker 不共享普通字典
正式部署可能启动多个进程:
Worker A
→ 自己的 memory {}
Worker B
→ 另一份 memory {}如果第一次请求落到 A,第二次请求落到 B,B 可能看不到 A 中保存的历史。
因此普通字典适合:
- 学习 Memory 原理;
- 单进程 Demo;
- 快速原型。
不适合:
- 多进程生产环境;
- 服务器重启后仍需保留记录;
- 多台服务器共享会话;
- 大量用户与长历史;
- 权限、搜索、统计和审计。
23.3 历史会无限增长
当前每轮都把完整历史重新发给模型:
第 1 轮:2 条消息
第 10 轮:20 条消息
第 100 轮:200 条消息可能产生:
- 超过模型上下文窗口;
- Token 成本不断上升;
- 响应变慢;
- 早期无关信息干扰当前问题。
未来需要:
限制最近 N 轮
对旧历史做摘要
区分短期上下文与长期记忆
从长期记忆中按需检索23.4 Memory 不等于模型永久学习
这里没有训练模型,也没有修改模型参数。
Memory
→ 应用保存的数据
上下文
→ 本次请求发送给模型的信息
模型参数
→ 训练形成的内部能力停止服务器后字典消失,足以证明“小明”并没有永久进入模型本身。
23.5 当前还缺少安全与权限
如果任何客户端只要猜到 abc123 就能访问该会话,就存在越权风险。正式系统必须建立:
用户登录身份
↓
校验 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 已经能够解释的完整架构
Client
↓ HTTP Request
Uvicorn
↓
FastAPI
↓
Pydantic
↓
main.py 路由函数
├── memory.py
└── model.py
↓
OpenAI 风格 SDK
↓
Qwen API
↓
HTTP Response25.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 这一阶段形成的软件工程思想
关注点分离
main.py
→ HTTP / 路由 / 编排
model.py
→ 模型 SDK
memory.py
→ 会话历史
config.py
→ 配置单一事实来源
默认模型等配置只在一个权威位置维护。
尽早失败
必要配置缺失时,在启动阶段就给出明确错误。
分层排错
根据错误判断请求走到了哪一层,而不是同时乱改所有代码。
数据一致性
模型成功后再提交完整的一问一答,避免半轮历史污染 Memory。
先跑通最小链路,再逐层增加复杂度
模拟函数
↓
真实模型
↓
配置层
↓
错误处理
↓
多轮 messages
↓
服务器 Memory这条方法以后同样适用于工具调用、RAG、数据库和部署。
二十六、下一阶段学习路线
当前所在位置:
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
把:
memory = {}升级成数据库。
建议先建立三个核心实体:
users
→ 谁在使用系统
conversations
→ 用户创建的聊天窗口
messages
→ 每个聊天窗口中的具体消息关系:
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 还要判断是否需要行动:
用户任务
↓
LLM 判断下一步
↓
是否需要工具?
├── 否 → 最终回答
└── 是
↓
调用 Tool
↓
得到 Observation
↓
再交给 LLM
↓
继续判断,直到完成一个最小 Agent 通常包含:
LLM + Tools + Memory + Loop26.3 Tool Calling
需要继续学习:
工具函数
工具 Schema
模型如何选择工具
参数如何生成和验证
程序如何真正执行工具
工具结果如何回填 messages
如何防止无限循环和危险调用26.4 RAG
文档
↓
切块 Chunk
↓
Embedding
↓
Vector Database
↓
相似度检索
↓
把相关内容放入上下文
↓
LLM 回答RAG 与 Memory 有联系,但不是同一件事:
对话 Memory
→ 用户与系统过去说过什么
RAG
→ 从外部知识库取回当前问题需要的资料26.5 Streaming 与生产工程
当前响应是:
等待模型全部生成
↓
一次性返回后续 Streaming 会变成:
模型产生一部分 Token
↓
服务器立即发送一部分
↓
客户端逐步显示之后还需要补齐:
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 启动
uvicorn main:app --reload检查:
- 没有配置缺失的
RuntimeError; - Uvicorn 正在监听预期端口;
- 打开
/能收到“API正在运行”; - 打开
/docs能看到POST /chat。
A.3 单轮模型测试
{
"conversation_id": "single-test",
"question": "什么是 Agent?"
}检查:
- 状态码是 200;
- 回答来自真实模型而不是旧的模拟拼接字符串;
- Response Body 中包含
conversation_id和answer。
A.4 多轮测试
第一次:
{
"conversation_id": "abc123",
"question": "我的名字叫小明"
}第二次:
{
"conversation_id": "abc123",
"question": "我叫什么名字?"
}检查:
- 第二次能够回答“小明”;
- 客户端第二次没有重新发送“小明”;
- 服务器确实从 Memory 中读取了历史。
A.5 会话隔离测试
{
"conversation_id": "xyz456",
"question": "我叫什么名字?"
}检查:
-
xyz456不知道abc123中的小明; - 没有发生会话串线。
A.6 RAM Memory 边界测试
先让 abc123 记住小明
↓
Ctrl + C 停止 Uvicorn
↓
重新启动服务
↓
再次询问 abc123:我叫什么名字?预期:模型不再知道小明。这证明当前数据只存在于进程内存,还没有持久化。
附录 B:自测题
B.1 FastAPI 与 HTTP
- 为什么普通 Python 函数不能让另一台电脑直接通过网络调用?
FastAPI、FastAPI()和app分别是什么?- Uvicorn 与 FastAPI 的职责有什么区别?
- 为什么访问
/不是在访问main.py文件? @app.post("/chat")建立了什么映射?- Python 字典为什么能被客户端收到为 JSON?
- Swagger UI 为什么可以被看作一个 Client?
B.2 Pydantic 与配置
question: str为什么通常是必填字段?model: Optional[str] = None表示什么?- 422 通常发生在哪一层?业务函数是否一定执行?
- 为什么 API Key 不能硬编码在
model.py? .env → config.py → model.py的配置链如何工作?- 为什么必要配置缺失时适合
raise RuntimeError? - 什么叫 Single Source of Truth?
B.3 错误处理
return None和裸raise的区别是什么?- 为什么
APIConnectionError与 401 的排查方向不同? - 为什么
model_not_found反而能证明请求大概率已经到达上游服务器? - 为什么上游返回 404,而自己的 API 可能返回 500?
model.py与main.py应分别承担什么错误处理职责?- 修改 Pydantic 默认值后,为什么要刷新
/docs?
B.4 异步
- 并发和并行有什么区别?
await等待时,Event Loop 可以做什么?- 为什么在
async def中使用requests.get()不会自动变成真正异步? - 为什么连续三次
await不一定并发? - 同步 SDK 是否一定意味着代码“不合理”?
B.5 Messages 与 Memory
- 为什么模型 API 不会天然记住上一轮请求?
- 为什么
messages使用外层 list、内层 dict? system、user、assistant分别表示什么?- 为什么需要保存历史
assistant消息? conversation_id解决什么问题?它为什么不完全等于用户 ID?Dict[str, List[Dict[str, str]]]应怎样从外到内解释?memory.get(conversation_id, [])为什么需要默认空列表?- 为什么
get()不等于创建了新会话? append()和extend()在构建消息列表时有什么区别?- 为什么应该在模型成功后再保存完整一问一答?
- 为什么
abc123知道小明,而xyz456不知道? - 为什么
memory = {}是服务器端 Memory,却不是持久化 Memory?
最终总结
这一阶段真正建立起来的,不只是几个 FastAPI 写法,而是一条完整的软件理解链:
Python 函数
↓ 通过 FastAPI 暴露
HTTP API
↓ 通过 Uvicorn 运行
可被 Client 调用的服务
↓ 接入模型 SDK
真实 LLM Service
↓ 传递完整 messages
多轮 Chat
↓ 使用 conversation_id 查历史
服务器端 Memory最关键的三个认识是:
- 模型能力只是系统的一层。 FastAPI、配置、错误处理、Memory 和数据库共同决定应用是否可靠。
- 模型看起来有记忆,不等于模型永久记住了用户。 当前机制是应用保存历史,并在下一次请求中重新发送。
- 会运行只是起点,知道数据如何流动、错误停在哪一层、为什么这样拆模块,才是真正进入 Agent 工程。
当前项目已经是一个简化版 ChatGPT 后端的雏形。下一步应把进程内字典升级为数据库持久化,再在这个基础上加入 Agent Loop、Tools 和 RAG。