前言

最近在系统补 AI/Agent 应用开发的工程基础。刚开始我以为,自己最需要补的是 Agent 框架、RAG、模型调用这些“看起来更像 AI 的部分”。但真正开始做项目之后,我越来越强烈地意识到:做不稳 AI/Agent 应用,并不是因为不会调模型,而是因为服务底座不够扎实。

下面这些问题,如果没有真正写过后端服务代码,往往很容易停留在“看得懂”阶段:

  • 异步到底解决什么问题,为什么不能乱用?
  • 类型标注到底是形式主义,还是确实能提高代码质量?
  • dataclass 和 Pydantic 应该怎么分工?
  • 后端服务为什么不能只靠 print() 调试?
  • 为什么很多服务都要有 request id?
  • 为什么统一异常处理比到处写 try/except 更好?
  • 装饰器在后端项目里到底适合做什么?

这些内容单独看都不算复杂,但如果没有把它们放进真实服务场景里,很容易变成一堆零散知识点。于是我决定先用一个训练型项目,把 Python 服务基础这一层彻底捋清楚。这篇文章就是我对这部分学习的第一轮完整复盘。

这篇文章主要想讲清楚 4 件事:

  • Python 服务基础到底在学什么
  • 这些基础能力在真实后端服务里分别解决什么问题
  • 它们之间是怎样联动起来的
  • 为什么这部分内容会直接影响后面学 FastAPI、Redis、任务系统、RAG 和 Agent 编排的效率

一、Python 服务基础是服务思维训练

很多人学 Python,最开始接触的都是脚本式写法。
比如:

  • 读一个文件
  • 调一个接口
  • 写一个小工具
  • 做一点数据处理
    这些场景里,代码通常是“从上往下执行一次”,逻辑简单时确实不需要太多工程组织。

但后端服务完全不是这个思路,一个真实的服务通常有下面这些特点:

  • 不是只执行一次,而是要长期运行
  • 不是只服务一个人,而是要同时处理很多请求
  • 不是只给自己看结果,而是要对外暴露稳定接口
  • 不是出错就结束,而是要把错误规范地返回给调用方
  • 不是只要能跑,而是要能排查、能扩展、能维护

所以Python 服务基础强化真正要练的不是这些语法名词本身,而是下面这些能力:

  • 如何让配置集中管理
  • 如何让函数输入输出边界更清楚
  • 如何给内部数据建模
  • 如何让日志可追踪
  • 如何把请求上下文在服务链路里传下去
  • 如何把错误变成稳定的接口响应
  • 如何把通用逻辑从业务代码里抽出去
  • 如何判断什么时候该同步、什么时候该异步

二、先建立一个整体视角:一条请求在服务里是怎么流动的

当一个请求进入服务后,通常会经历下面这些步骤:

  1. 服务启动时先读取配置
  2. 日志系统初始化
  3. 请求进来后先经过中间件
  4. 中间件给这次请求分配一个 request id
  5. 参数被校验和解析
  6. 业务函数开始执行
  7. 业务函数里可能会调用数据库、缓存或其他服务
  8. 如果出错,需要被统一转换成规范响应
  9. 整个请求过程需要被日志记录下来
  10. 请求结束后响应返回给客户端

三、配置管理:为什么后端服务不能把配置写死在代码里

1. 为什么配置不能散落在代码里

如果这样写:

token = "dev-token"
port = 8000
database_url = "sqlite:///local.db"
redis_url = "redis://localhost:6379/0"

刚开始看起来没问题,服务也确实能跑。但一旦你真的要把它当成项目继续做,很快就会遇到这些问题:

  • 本地环境和线上环境配置不一样
  • 你和别人机器上的配置不一样
  • 数据库地址会变
  • Redis 地址会变
  • Token 不应该写死在仓库里
  • 一个服务可能会有多个运行环境:开发、测试、生产

这时候如果配置值散落在很多文件里,维护会非常痛苦。所以后端服务里通常会做一件很重要的事:把所有“会变的外部参数”统一收口。

2. 一个最小配置类长什么样

下面是一个很典型的做法:

from functools import lru_cache
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    app_name: str = Field(default="Agent Backend Playground", alias="APP_NAME")
    app_env: str = Field(default="development", alias="APP_ENV")
    app_host: str = Field(default="0.0.0.0", alias="APP_HOST")
    app_port: int = Field(default=8000, alias="APP_PORT")
    api_token: str = Field(default="dev-token", alias="API_TOKEN")
    database_url: str = Field(default="sqlite+aiosqlite:///./local.db", alias="DATABASE_URL")
    redis_url: str = Field(default="redis://localhost:6379/0", alias="REDIS_URL")

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
        extra="ignore",
    )

@lru_cache
def get_settings() -> Settings:
    return Settings()

BaseSettings 的核心思想其实很简单:

  • 先定义一个统一的配置对象
  • 用字段描述服务运行所需的参数
  • 允许这些参数从环境变量或 .env 文件中读取
  • 给每个参数一个合理默认值

3. 这段代码到底解决了什么问题

  1. 配置集中管理。

    以后要看服务都有哪些关键配置,不用全项目到处搜,只要看这个配置类就行。

  2. 环境隔离。

    比如代码里字段叫 api_token,但实际可以从环境变量 API_TOKEN 读取。这样本地开发、测试环境、部署环境都可以有各自的值,而不需要改业务代码。

  3. 类型明确。
    比如:

app_port: int

这就明确表示端口应该是整数,而不是任意字符串。配置系统在读值时也可以帮你做一定程度的类型转换和约束。

4. 为什么这里要加 @lru_cache

@lru_cache
def get_settings() -> Settings:
    return Settings()

因为配置对象通常在服务运行期间是稳定的,所以没必要每次使用时都重新创建。

可以把它理解成:

  • 第一次读取配置时真正创建对象
  • 后面重复调用时直接复用

5. 这一部分最值得记住的结论

配置管理的重点不是“会不会用 BaseSettings”,而是要有这个意识:服务里的配置不应该散落在代码各处,而应该有一个统一、清晰、可切换环境的入口。


四、类型标注:它不是为了形式,而是在帮你明确代码边界

很多人第一次接触类型标注时,会有一个常见误解:“Python 是动态语言,写这些类型是不是有点多余?”如果只是写一次性小脚本,这种感觉可以理解。但在后端服务里,类型标注的价值会明显变大。因为服务代码的核心问题之一就是:边界是否清晰。

1. 最基础的函数签名就已经很有价值

看这个非常简单的例子:

def normalize_message(text: str) -> str:
    return " ".join(text.strip().split())

它的好处:

  • 这个函数输入什么,一眼就知道
  • 这个函数输出什么,一眼就知道
  • 后面调用的人不容易乱传参数
  • 编辑器补全和检查也会更友好

2. 为什么服务层特别需要边界清晰

在后端服务里,函数通常不是孤立存在的,而是会一层层被调用:

  • 路由层调 service 层
  • service 层调仓储层
  • service 层调工具函数
  • service 层调外部能力

如果这些函数签名都写得很模糊,项目一复杂,就很容易出现下面这些问题:

  • 调用方不确定该传什么
  • 返回值结构不清楚
  • 重构时容易改坏
  • 业务逻辑越来越难读

3. TypedDict:当你想返回结构化字典时很有用

有时候你不想专门定义一个完整类,但又不想返回一个随便塞字段的 dict,这时 TypedDict 就很适合。

from typing import Literal, TypedDict
class MessageSummary(TypedDict):
    normalized: str
    length: int
    priority: Literal["low", "medium", "high"]

这段代码的含义是:这个字典不是任意结构,而是应该长成这样:

  • normalized 是字符串
  • length 是整数
  • priority 只能是三个固定值之一

然后函数可以这样写:

def summarize_message(text: str) -> MessageSummary:
    normalized = normalize_message(text)
    priority: Literal["low", "medium", "high"] = "low"

    if len(normalized) > 80:
        priority = "high"
    elif len(normalized) > 30:
        priority = "medium"
    return {
        "normalized": normalized,
        "length": len(normalized),
        "priority": priority,
    }

4. Literal:用来约束有限个可选值

刚才代码里还有一个很实用的点:

Literal["low", "medium", "high"]

它表示这个字段不是任意字符串,而是只能从固定几个值里选。

这在后端服务里非常常见,比如:

  • 任务状态:queued / running / success / failed
  • 模型模式:sync / stream
  • 优先级:low / medium / high

为什么这很重要?因为一旦你不做约束,系统里就很容易出现各种不统一的值:

  • "High"
  • "HIGH"
  • "urgent"
  • "middle-level"

5. Protocol:依赖能力,而不是依赖具体类

from typing import Protocol

class Formatter(Protocol):
    def format(self, message: str) -> str: ...

然后你可以写:

def format_for_log(message: str, formatter: Formatter) -> str:
    return formatter.format(message)

这段代码最重要的地方在于:函数不关心你传进来的 formatter 到底是哪个类,它只关心你有没有 format() 这个能力。这其实是一种很重要的设计意识:尽量依赖接口和能力,而不是一开始就绑定某个具体实现。当项目变复杂时,“依赖能力”往往比“依赖具体类名”更灵活。

6. 类型标注和 Pydantic 的区别

  • 普通类型标注:主要是在帮助你把 Python 内部代码写清楚
  • Pydantic:更适合做接口边界校验、解析和序列化

也就是说:

  • service 层、工具函数、内部对象,更适合优先考虑普通类型标注
  • 请求体、响应体、配置对象这些边界型数据,更适合使用 Pydantic

五、dataclass:为什么它特别适合服务内部的轻量对象

在服务项目里,它有一个重要的价值:非常适合表示内部轻量数据对象。

1. 一个简单例子

from dataclasses import dataclass

@dataclass(slots=True)
class PracticeMessage:
    user_id: str
    raw_text: str

    @property
    def normalized_text(self) -> str:
        return " ".join(self.raw_text.strip().split())

这段代码表达的是:

  • 这是一个消息对象
  • 它有两个核心字段:用户 id 和原始文本
  • 它还有一个根据原始文本计算出来的属性:规范化后的文本

这种对象特别像服务内部会用到的中间数据:

  • 结构清晰
  • 字段固定
  • 行为简单
  • 主要在内部使用

2. 为什么适合服务内部,而不是接口边界

这是 dataclass 和 Pydantic 的关键分工之一。

像这种对象:

  • 不是来自外部请求体
  • 不是给前端直接返回的响应模型
  • 不需要特别强的输入校验和序列化能力
    那它就很适合用 dataclass

3. RequestTrace 是一个非常典型的服务内部对象

from dataclasses import dataclass, field
from time import perf_counter
from uuid import uuid4

@dataclass(slots=True)
class RequestTrace:
    path: str
    request_id: str = field(default_factory=lambda: uuid4().hex)
    started_at: float = field(default_factory=perf_counter)

这个对象承载的是一次请求处理过程中的内部信息:

  • 当前请求路径
  • 这次请求的唯一 id
  • 请求开始时间

这类对象的特点特别明显:

  • 生命周期很短
  • 只在服务内部使用
  • 字段明确
  • 不需要复杂校验

4. default_factory 为什么值得注意

这里有一个小细节很重要:

request_id: str = field(default_factory=lambda: uuid4().hex)
started_at: float = field(default_factory=perf_counter)

它的意思是:每次创建对象时,都动态生成新的默认值。

为什么要这么写:

  • request_id 每次都应该不同
  • started_at 每次都应该是当前时间点

5. slots=True 的意义

@dataclass(slots=True)

它传达的设计意图:

  • 这个对象结构是相对固定的
  • 它更像一个明确的轻量数据容器

六、日志:为什么后端服务不能只靠 print()

1. 服务日志和脚本打印的区别

脚本通常是:

  • 自己跑
  • 自己看输出
  • 出错就停

而服务日志通常需要满足:

  • 同时记录很多请求
  • 支持排查线上问题
  • 支持过滤、检索和分析
  • 能把同一次请求的链路串起来

所以在服务场景里,日志最好不是一堆随手写的字符串,而是:结构化日志。

2. 什么叫结构化日志

import json
import logging
from datetime import UTC, datetime

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "time": datetime.now(UTC).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "request_id": request_id_var.get(),
        }
        return json.dumps(payload, ensure_ascii=True)

这里最重要的不是 JSON 本身,而是它把日志拆成了明确字段:

  • 时间
  • 日志级别
  • logger 名
  • 消息内容
  • request id

相比之下,如果只是:

print("request finished")

短期看能用,但长期几乎没法支持复杂服务排查。

3. 为什么 request_id 是日志里的关键字段

后端服务同时会处理很多请求,日志天然会交错。如果没有 request id,可能会看到一堆类似这样的输出:

  • request started
  • db write ok
  • request completed
  • stream chunk sent
  • task saved
    但并不知道这些日志分别属于哪一次请求,所以 request id 的价值是:把一条请求链路上的日志统一串起来。

4. 日志初始化为什么要在服务启动时做

一个最小初始化过程通常像这样:

def configure_logging(enable_request_log: bool) -> None:
    root = logging.getLogger()
    root.handlers.clear()
    root.setLevel(logging.INFO if enable_request_log else logging.WARNING)
    handler = logging.StreamHandler()
    handler.setFormatter(JsonFormatter())
    root.addHandler(handler)

需要先统一定义:

  • 输出格式
  • 输出级别
  • 输出位置
    这样后续不同模块打出来的日志,才会保持风格一致。

5. 这一部分最该记住的结论

日志在后端服务里不是调试碎片,而是:服务可观测性的入口。尤其当请求开始变多、链路开始变长、问题开始变复杂时,结构化日志和 request id 的价值会迅速放大。


七、请求上下文:为什么 request id 能自动在服务里往下传

第一次看到 request id 可能会下意识觉得:“是不是每个函数都要手动传一个 request_id 参数?”如果真这么做,代码很快会变得特别笨重。所以服务里通常会用一种更好的办法:请求上下文。

1. ContextVar 可以先怎么理解

from contextvars import ContextVar

request_id_var: ContextVar[str] = ContextVar("request_id", default="system")

第一次接触 ContextVar,可以先把它理解成:当前执行链路上的“上下文变量”。

它和普通全局变量最大的区别是:

  • 全局变量容易被所有请求共享
  • ContextVar 更适合保存当前请求自己的上下文信息

这在异步服务里尤其重要,因为很多请求会并发执行。如果用全局变量记录 request id,很容易串数据。

2. 为什么不用普通全局变量

current_request_id = "xxx"

然后每来一个请求就修改一次。这样在单请求脚本里也许问题不明显,但在并发服务里就很危险,因为多个请求可能同时运行,它们会互相覆盖这个值。所以这里的关键认知是:请求级状态不要随便存在普通全局变量里。

3. 一个最小请求上下文管理器

from contextlib import asynccontextmanager

@asynccontextmanager
async def request_trace(path: str):
    trace = RequestTrace(path=path)
    token = request_id_var.set(trace.request_id)
    try:
        yield trace
    finally:
        request_id_var.reset(token)

可以把它拆成 4 步:

  1. 为这次请求创建一个 trace 对象
  2. 把 request id 写进上下文变量
  3. 把控制权交给真正请求处理逻辑
  4. 请求结束后恢复上下文

这就是上下文管理器特别适合的场景:进入时设置,退出时清理。

4. 为什么这比手动层层传参更好

有了这种机制之后,真正业务代码里如果需要 request id,就可以直接拿:

trace_id = request_id_var.get()

好处是:

  • 不是每个函数都要强行加 request_id 参数
  • 代码签名更干净
  • 只有真正需要 request id 的地方才去取

八、中间件:为什么请求级公共逻辑适合放在这里

如果 request id、请求耗时、统一请求日志这些逻辑要在整个服务里生效,那么最适合放在哪里?答案是:中间件。

1. 什么是中间件

可以先把中间件理解成:请求真正进入业务逻辑前后,都会经过的一层通用处理。

from time import perf_counter

class RequestContextMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        async with request_trace(request.url.path) as trace:
            started = perf_counter()
            response = await call_next(request)
            duration = round(perf_counter() - started, 4)
            response.headers["X-Request-ID"] = trace.request_id
            logger.info(
                "request_completed",
                extra={
                    "method": request.method,
                    "path": request.url.path,
                    "status_code": response.status_code,
                    "duration": duration,
                },
            )
            return response

2. 这段中间件代码做了什么

  • 给当前请求创建 trace
  • 生成 request id
  • 统计请求耗时
  • 调用下游真正业务逻辑
  • 把 request id 写进响应头
  • 记录请求完成日志

这些事情都有一个共同特征:它们不是某个业务接口特有的逻辑,而是整个请求链路都可能需要的公共逻辑,所以适合放在中间件里。

3. 为什么这些逻辑不应该散落在每个接口里

如果把上面的逻辑手动写进每个接口函数,会有三个明显问题:

  • 重复代码很多
  • 很容易漏掉某些接口
  • 业务逻辑和链路逻辑混在一起,越来越难维护

所以后端服务里一个很重要的分层意识是:

  • 请求级公共逻辑,优先考虑中间件
  • 具体业务逻辑,放在业务函数里

九、异常处理:为什么统一收口比到处写 try/except 更好

常见的两个极端是:

  • 几乎不处理异常,报错直接炸出来
  • 到处写 try/except,每个接口自己拼错误响应

1. 先区分两类错误:业务异常和系统异常

业务异常 的意思是:程序不一定坏了,但当前请求不符合业务规则。

例如:

  • 上传文件时文件名为空
  • 查询任务时任务不存在
  • 权限不足
  • 用户输入不满足业务约束

这类错误通常应该是:

  • 可预期的
  • 有明确业务语义的
  • 可以稳定返回给调用方的

系统异常 则更像:

  • 数据库连接失败
  • 某个变量空指针
  • 代码逻辑里出现了未预料错误
    这类错误往往不应该把内部细节完整暴露给调用方,而是应该统一兜底。

2. 一个最小业务异常基类

class AppError(Exception):
    def __init__(self, message: str, code: str, status_code: int = 400):
        super().__init__(message)
        self.message = message
        self.code = code
        self.status_code = status_code

这段代码要求每个业务异常至少说明三件事:

  • message:错误说明
  • code:稳定错误码
  • status_code:HTTP 状态码

这比简单地 raise Exception("xxx") 要好很多,因为它让错误开始具备“系统可识别”的结构。

3. 为什么要定义具体业务异常类

class TaskNotFoundError(AppError):
    def __init__(self, task_id: str):
        super().__init__(
            message=f"Task '{task_id}' was not found",
            code="task_not_found",
            status_code=404,
        )

class UploadValidationError(AppError):
    def __init__(self, detail: str):
        super().__init__(
	        message=detail, 
	        code="invalid_upload", 
	        status_code=400,
        )

这样做有几个好处:

  • 业务语义更清楚
  • 代码可读性更强
  • 前端或调用方可以基于稳定错误码处理
  • 错误返回格式可以保持统一

4. 错误响应为什么也要统一结构

def format_error_response(message: str, code: str, details=None) -> dict:
    return {
        "error": {
            "message": message,
            "code": code,
            "details": list(details or []),
        }
    }

它意味着:不管什么错误,对外尽量都长成统一结构。

这样前端或调用方后续就可以稳定读取:

  • error.message
  • error.code
  • error.details
    而不是这个接口返回一个结构、那个接口返回另一个结构。

5. 全局异常处理比到处 try/except 更好的原因

一个典型全局处理器:

def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(AppError)
    async def app_error_handler(_: Request, exc: AppError) -> JSONResponse:
        return JSONResponse(
            status_code=exc.status_code,
            content=format_error_response(exc.message, exc.code),
        )

    @app.exception_handler(Exception)
    async def unhandled_error_handler(_: Request, exc: Exception) -> JSONResponse:
        return JSONResponse(
            status_code=500,
            content=format_error_response(str(exc), "internal_error"),
        )

这比每个接口自己写 try/except 的好处非常明显:

  • 接口层更干净
  • 错误格式更统一
  • 业务异常和系统异常更容易分层
  • 后续维护和修改成本更低

十、装饰器:为什么它特别适合做横切逻辑

在后端服务里,装饰器最重要的价值是:把通用但非核心业务的逻辑,从业务函数里抽出去。

1. 一个典型的服务装饰器:记录耗时

import asyncio
from functools import wraps
from time import perf_counter

def log_duration(label: str):
    def decorator(func):
	    #异步函数
        if asyncio.iscoroutinefunction(func):
            @wraps(func)
            async def async_wrapper(*args, **kwargs):
                started = perf_counter()
                result = await func(*args, **kwargs)
                logger.info(
                    "duration_recorded",
                    extra={"label": label, "seconds": round(perf_counter() - started, 4)},
                )
                return result
            return async_wrapper
            
		#同步函数
        @wraps(func)
        def sync_wrapper(*args, **kwargs):
            started = perf_counter()
            result = func(*args, **kwargs)
            logger.info(
                "duration_recorded",
                extra={"label": label, "seconds": round(perf_counter() - started, 4)},
            )
            return result
        return sync_wrapper
    return decorator

2. 为什么这个逻辑适合装饰器

“记录函数耗时”不是某个业务函数专属的逻辑,而是很多 service 函数都可能需要的通用能力。

如果把它写在每个函数内部,代码会变成这样:

  • 每个函数都手动计时
  • 每个函数都手动打日志
  • 主业务逻辑被这些样板代码包住

而装饰器的价值就是:把这些横切逻辑统一抽出去,让业务函数本身更专注。

3. 为什么这里要区分同步和异步函数

这里用了:

if asyncio.iscoroutinefunction(func):
  • 同步函数调用方式是直接执行
  • 异步函数必须 await

所以装饰器如果想同时支持两者,就必须分别处理。这也从侧面说明:异步函数不是普通函数前面多写一个 async 那么简单,它会影响整个调用方式。

4. @wraps 为什么重要

这一行很多人会顺手写,但不一定真的理解:

@wraps(func)

它的作用是尽量保留原函数的元信息,比如:

  • 函数名
  • 文档字符串
  • 调试和追踪时的函数身份
    如果没有它,很多被装饰后的函数在调试时会只显示 wrapper,体验会变差。

5. 装饰器和中间件的分工

顺手总结一下:

  • 中间件更适合请求级公共逻辑
  • 装饰器更适合函数级公共逻辑

例如:

  • 请求总耗时、request id 更适合中间件
  • 某个 service 方法耗时统计更适合装饰器

十一、同步、异步与基础并发:

1. 同步函数适合什么

看一个简单的例子:

def normalize_message(text: str) -> str:
    return " ".join(text.strip().split())

这类逻辑的特点是:

  • 本地处理
  • 没有外部等待
  • 执行很快

这种情况下,同步就足够了。所以一个很重要的判断是:不是所有函数都值得写成异步。

2. 异步函数适合什么

再看一个模拟 IO 等待的例子:

import asyncio

async def fake_io_step(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{name} finished in {delay:.2f}s"

虽然这里用的是 sleep,但你可以把它理解成真实服务里的这些场景:

  • 等数据库返回
  • 等 Redis 返回
  • 等文件读取完成
  • 等网络请求返回
  • 等模型流式输出

也就是说,异步更适合:主要成本在等待外部资源的逻辑。

3. 什么叫基础并发

再看这个例子:

async def run_concurrent_steps() -> list[str]:
    return await asyncio.gather(
        fake_io_step("redis-check", 0.05),
        fake_io_step("db-check", 0.05),
        fake_io_step("agent-check", 0.05),
    )

这段代码想表达的是:如果多个任务彼此独立,而且都主要在等待,那么它们可以一起挂起、并发推进,而不是一个接一个串行执行。

最直白的方式理解并发:

  • 串行:做完 A 再做 B,再做 C
  • 并发:A、B、C 一起开始推进,各自等待时互不阻塞

对后端服务来说,这就是异步最常见、最实用的价值来源。

4. 为什么“异步不能乱用”

异步不是“写成 async def 就更高性能”。

如果你在异步函数里做的是:

  • 大量 CPU 计算
  • 阻塞式文件操作
  • 阻塞数据库驱动调用
    那它仍然可能卡住事件循环。

所以更准确的说法应该是:异步擅长处理等待,不擅长假装解决所有性能问题。

5. 流式场景为什么天然适合异步

如果做的是聊天流式输出,这种写法就非常典型:

async for chunk in provider.stream_text(request):
    yield f"data: {chunk}\n\n"
    await asyncio.sleep(0.03)

异步在 AI 应用里的作用(流式输出):

  • 结果不是一次性全部返回
  • 而是边生成边发送
  • 每一小段输出之间都可能有等待

十二、把这些基础能力重新串起来:它们在服务里是怎么协作的

可以把一条最小请求链路理解成这样:

  1. 服务启动时先读取统一配置
  2. 日志系统按统一格式初始化
  3. 请求进入后,中间件先生成 request id
  4. request id 被写进请求上下文
  5. 后续业务代码和日志都可以从上下文里读取 request id
  6. service 层函数通过类型标注明确输入输出边界
  7. 内部轻量数据通过 dataclass 建模
  8. 通用函数耗时统计通过装饰器处理
  9. 如果发生业务错误,抛出业务异常
  10. 全局异常处理把错误转成稳定的 JSON 响应
  11. 如果业务涉及 IO 等待,就使用异步函数和并发工具

十三、这部分内容对后面学习 FastAPI、Redis、任务系统有什么帮助

agent部分内容后面继续学:

  • FastAPI
  • 文件上传
  • 流式输出
  • Redis
  • 异步任务
  • RAG
  • Agent 编排
    几乎都会反复用到这一层能力。

比如:

学 FastAPI 时

  • 请求模型和响应模型的理解,离不开边界意识
  • 中间件的作用,离不开请求链路意识
  • 流式输出的实现,离不开异步理解

学 Redis 和任务系统时

  • 任务状态设计,离不开有限值约束和结构表达
  • 后台任务排查,离不开日志和 request id
  • 长任务拆分,离不开同步和异步的判断

学 RAG 和 Agent 服务时

  • 文件处理、知识构建、模型调用,都会涉及 IO 等待
  • 服务接口、流式响应、错误处理,仍然依赖这套基础秩序

十四、总结:

这部分最大的收获,不是学会了语法,而是开始有了服务思维,真正意识到:

  • 配置不是随手写几个常量,而是服务运行的统一入口
  • 类型标注不是形式,而是在帮助我把函数边界写清楚
  • dataclass 不是玩具,而是服务内部轻量对象的工具
  • 日志不是 print() 的升级版,而是服务排查的基础设施
  • request id 和上下文管理不是“高级技巧”,而是请求链路可追踪的关键
  • 异常处理不是为了不报错,而是为了让错误变得可控、稳定、可维护
  • 装饰器不是炫技,而是抽离通用横切逻辑的实用工具
  • 异步不是银弹,而是一种面向等待场景的执行方式

下一篇预告

下一篇我会继续复盘 FastAPI 这一层:为什么 FastAPI 几乎成了 Python AI/Agent 服务的标配,以及一个最小可用的 Agent 服务骨架到底应该怎么搭起来。

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐