Agent 服务底座学习笔记（三）：MySQL、Redis 与异步任务到底该怎么配合

前言到了这一部分真正在从“会写接口”走向“理解后端系统”。真正有系统感的后端服务除了搭接口以外往往还要解决下面这些问题哪些数据需要长期保存哪些数据只是临时状态变化很快哪些任务不适合让用户一直同步等待提交任务之后系统如何告诉前端当前进度如果任务失败了如何知道它失败在哪一步这些问题一出现后端开发就不再只是“写几个接口”那么简单了而会开始进入更真实的系统设计层面。所以这篇文章不只写“数据库怎么连、Redis 怎么用、异步任务怎么起”而是想从更完整的后端视角把这三个部分的分工和协作讲清楚。一、为什么后端服务不能什么都同步做完这部分内容如果只从技术名词看很容易觉得它们彼此没有关系MySQLRedis异步任务但如果从真实服务场景出发这三者其实是在回应同一个问题当一个服务开始变得更真实之后数据和长任务应该如何组织我们可以先想一个很常见的场景假设你写了一个 AI/Agent 服务用户发来一个请求服务要做的事情可能包括记录这次请求内容调模型生成结果处理文件记录调用过程可能还要执行一段比较长的后续流程如果这些事情都硬塞在一次请求里同步完成会怎样最直观的几个问题是用户要一直等接口响应时间会很长超时风险变高中间过程不可见失败后不好排查结果和状态不好管理这时候你就会发现后端服务不能只靠“接口函数里把事情做完”这一种模式。必须开始区分几类东西要长期保留的数据。比如交互记录、调用记录、任务执行痕迹这些东西不能请求结束就消失。服务运行过程中的状态数据。比如某个任务当前是排队中、执行中还是失败了这种数据变化快但很重要。执行时间比较长的任务。这类任务不适合一直卡住主请求而更适合拆成“先提交再查询”。从这个角度重新看 MySQL、Redis 和异步任务时它们的角色就会一下子清楚很多MySQL 更适合处理长期结构化数据Redis 更适合处理快速变化的状态数据异步任务模型更适合处理长流程执行所以这部分内容真正学的不是三个工具而是一个后端系统如何把“历史数据”“当前状态”“长任务执行”分开管理。二、为什么有些数据适合放数据库而不是放内存里就算了很多人一开始做项目时会下意识地把“数据存起来”理解成一件很简单的事比如先放在内存里或者先随便写个列表、字典保存一下。在 demo 阶段这样当然能跑。但只要你开始稍微认真一点做服务就会发现有些数据天然就不适合只存在内存里。比如下面这些数据用户发过什么请求系统回复过什么某次模型调用用了什么参数某次任务执行到了第几步某次任务失败时的错误信息这类数据有一个共同点它们不只是“当下处理要用”而是以后还可能要查、要追踪、要分析。1. 数据库更适合承载“历史沉淀”例如一条交互记录很适合长成这样class InteractionRecord: id: int trace_id: str interaction_type: str prompt: str response_preview: str created_at: datetime这类数据的特点非常典型结构相对稳定字段清晰希望长期保存后续可能需要查询历史如果这些数据只放在内存里一旦服务重启记录就全没了。而一旦放进数据库你就可以做很多后续事情查最近 20 条交互按 trace id 定位某次请求看某类交互出现得多不多后续做分析和可视化2. 数据库里不一定只存“业务主数据”很多人刚接触数据库时会觉得数据库主要就是存用户表、订单表、商品表这种“主业务数据”。但在后端服务里数据库也很适合存一些“过程型记录”。例如除了交互记录还可以记录LLM 调用记录Agent 执行步骤记录文档元信息用户记忆信息这类数据虽然不一定直接给用户看但它们对排查、分析、复盘非常有价值。3. Repository 层到底在解决什么问题当项目开始接数据库时另一个很容易踩的坑就是路由层或 service 层直接到处写 SQL。短期看似乎很方便但后面会出现几个问题SQL 到处散数据访问方式不统一业务逻辑和数据访问耦在一起改动和排查都变麻烦所以更合理的方式通常是把数据库访问逻辑收口到 repository 层。比如一条“写入交互记录”的逻辑可以长成这样class InteractionRepository: async def add_record( self, trace_id: str, interaction_type: str, prompt: str, response_preview: str, ) - None: await connection.execute( INSERT INTO interaction_records ( trace_id, interaction_type, prompt, response_preview, created_at ) VALUES (?, ?, ?, ?, ?) , (trace_id, interaction_type, prompt, response_preview, utcnow()), ) await connection.commit()这里最重要的不是 SQL 本身而是分层思路service 层只关心“我要记录一次交互”repository 层负责“怎么把它写进数据库”这个边界一旦建立起来整个服务会清楚很多。所以这一部分真正要学会的是持久化数据应该有专门的数据访问层而不是在业务代码里到处乱写。三、为什么不是所有数据都应该进 MySQLRedis 的角色到底是什么理解了数据库之后很多人会自然问一个问题“既然数据库也能存数据那 Redis 到底多出来是干什么的”答案并不在于“Redis 更快”这么简单而在于不同数据的性质不同。数据库更适合“长期结构化数据”而 Redis 更适合“变化快、查询方式简单、主要关注当前值”的数据。而任务状态正是最典型的例子1. 任务状态为什么更像“看板数据”假设你有一个后台任务它的状态可能会这样变化queuedrunningwaiting_toolsuccessfailed你会发现这类数据和交互记录非常不一样。交互记录更像“历史档案”写进去了就留着后面偶尔查重点是沉淀历史而任务状态更像“当前看板”更新很频繁查询通常按 task id 直接查更关心此刻状态更像运行时数据所以这时候 Redis 的角色就很清楚了它不是数据库的替代品而是状态管理的好位置。2. 一个最小任务状态存储是什么样例如可以设计一个任务后端抽象class TaskBackend(ABC): async def create(self) - TaskRecord: ... async def get(self, task_id: str) - TaskRecord | None: ... async def update(self, record: TaskRecord) - None: ...然后 Redis 版本的实现会更像这样class RedisTaskBackend(TaskBackend): def __init__(self, redis_url: str): self.client Redis.from_url(redis_url, decode_responsesTrue) staticmethod def _key(task_id: str) - str: return ftask:{task_id} async def get(self, task_id: str) - TaskRecord | None: raw await self.client.get(self._key(task_id)) if raw is None: return None return TaskRecord.model_validate_json(raw) async def update(self, record: TaskRecord) - None: await self.client.set(self._key(record.task_id), record.model_dump_json())这段代码背后的设计非常直观任务按task:{task_id}这样的 key 保存每次查询都可以直接按 key 拿每次状态更新都可以直接覆盖最新快照四、异步任务模型真正重要的不是“后台跑”而是“状态可描述”很多人第一次接触异步任务时会把它理解成“把某个函数放后台执行一下。”但真正重要的问题其实是当一个任务被放到后台执行后你如何知道它现在怎么样了也就是说异步任务系统真正的核心不只是“异步”而是任务有身份任务有状态任务有进度任务有结果任务有失败信息1. 一个任务记录应该至少长什么样from datetime import datetime from typing import Literal from pydantic import BaseModel, Field TaskStatus Literal[queued, running, waiting_tool, success, failed] class TaskStep(BaseModel): step_no: int title: str status: Literal[running, success, failed] detail: str started_at: datetime finished_at: datetime | None None duration_ms: int | None None class TaskRecord(BaseModel): task_id: str status: TaskStatus trace_id: str | None None current_step: int 0 provider: str | None None result: str | None None result_data: dict | None None error: str | None None steps: list[TaskStep] Field(default_factorylist) created_at: datetime updated_at: datetime可以看到一条任务记录不只是“有个 id”它还明确描述了当前状态当前跑到第几步结果是什么错误是什么每一步做了什么创建和更新时间这让我真正开始理解后台任务最重要的不是“任务已经丢出去了”而是“任务过程是可见的”。2. 为什么长任务不能强塞在一次请求里异步任务之所以存在通常不是因为大家喜欢把事情搞复杂而是因为有些任务天然不适合同步做完。例如长文本生成文件解析索引构建多步骤工具调用后台批量处理如果这些流程都放在一次请求里同步做完典型问题就是用户一直等接口很容易超时失败时很难表达中间状态后续重试和排查都很别扭所以更合理的方式通常是先提交任务返回 task id后台继续执行前端或调用方再查询状态普通同步接口更像请求进来立刻做完返回最终结果任务型接口更像请求进来服务先接单返回“我已经收到了”后续再查执行结果这也是为什么任务提交接口通常会返回202 Accepted。它表达的不是“已经做完了”而是你的请求我接收了但真正执行还在继续。3. 一个最小任务提交流程是什么样class TaskService: async def submit_chat_task(self, payload: ChatRequest) - TaskRecord: record await self.task_backend.create() record.trace_id uuid4().hex await self.task_backend.update(record) asyncio.create_task(self._run_chat_task(record.task_id, payload)) return record这几行代码虽然不长但已经把异步任务的核心流程说明白了先创建任务记录。给这条任务一个 trace_id。把任务状态写回存储层。后台启动真正执行逻辑。当前请求先返回 task 记录。4. 任务执行时为什么状态流转特别重要再看后台执行部分async def _run_chat_task(self, task_id: str, payload: ChatRequest) - None: record await self.task_backend.get(task_id) if record is None: return record.status running await self.task_backend.update(record) try: response await self.agent_runner.run(...) record.status success record.result response.reply record.result_data response.structured_data except Exception as exc: record.status failed record.error str(exc) await self.task_backend.update(record)这段代码想表达的是一条任务不是“有或没有”这么简单而是会经历状态变化刚提交时是queued真正开始跑时变成running成功结束时变成success出错时变成failed如果中间还涉及多步骤过程甚至可以进一步细化成当前跑到第几步这一步成功还是失败每一步花了多久所以任务系统真正难的地方是怎么把任务从黑盒变成一个可被查询和理解的状态机。五、把 MySQL、Redis 和异步任务重新串起来它们到底是怎么协作的可以想象一个比较真实的场景用户发起了一个比较长的聊天或处理请求系统不想让他一直同步等待于是采用任务模式。整个过程大概是这样1. 前端提交任务接口接收到请求后不是直接同步把全部逻辑做完而是先调用任务服务创建一条任务记录。这时任务后端会生成task_id初始状态queued创建时间然后接口立即返回task_id当前状态trace_id前端拿到这些数据后就知道这件事已经被系统接收了。2. 任务状态放进 Redis 或内存后端这时最关键的不是把最终结果立刻算出来而是先让这条任务变得可查。所以任务状态会先进入任务后端。如果是本地练习环境可能存内存如果更接近真实环境就更适合存 Redis。这时 Redis 承担的角色就是“让我随时能按 task id 查到当前状态。”3. 后台开始真正执行任务后台协程开始跑真正的业务流程一开始会把状态从queued改成running。如果任务执行过程中有多个步骤还会不断更新当前第几步这一步是否成功中间输出摘要这让任务变得不再是“只知道还没结束”而是“知道它进行到哪了”。4. 过程型结果沉淀进数据库执行过程中如果有值得长期保留的信息例如交互摘要模型调用记录步骤记录文档处理记录这些就更适合落进数据库这些数据不是为了“当前状态查询”而是为了历史追踪排查分析后续复盘所以这里 MySQL / SQLite 的角色更像“把这次任务执行过程里值得长期留下来的信息沉淀下来。”5. 任务结束后状态回写任务成功了就把状态更新成success并把结果写进任务记录。任务失败了就把状态更新成failed并带上错误信息。此时前端再去查任务状态就能看到已完成还是失败最终结果是什么是否有错误信息这样一个完整的“提交任务 - 后台执行 - 查询状态”的闭环就形成了。所以你会发现这三者根本不是割裂的数据库负责沉淀历史Redis 负责保存当前状态异步任务负责组织长流程执行这就是今天这部分内容真正要建立的系统感。六、这一部分对后面学习 Agent / RAG / 任务系统有什么帮助这一块非常重要因为它决定了后面学更复杂能力时会不会乱。比如1. 对 Agent 长流程执行很重要Agent 很多时候并不是“一步出答案”而是调模型调工具走多步流程可能中间等待这天然适合任务状态模型。2. 对 RAG 文档处理很重要文档解析、切分、索引构建往往不是瞬时完成的很适合异步任务化处理。3. 对系统排查很重要一旦服务出了问题你需要的不只是“接口报错了”而是这次请求有没有落记录任务跑到哪一步失败了状态什么时候从 running 变 failed4. 对后端系统感非常重要一个服务不是几个接口函数的集合而是由接口层、数据层、状态层、任务执行层共同组成的系统。总结这一部分最重要的不是我学会了“怎么连数据库”“怎么用 Redis”“怎么起异步任务”而是我开始真正理解一个后端系统必须区分历史数据、当前状态和长任务执行。以前我更容易把服务理解成请求进来代码跑完结果返回。但现在我会更自然地意识到有些数据应该长期沉淀到数据库有些状态更适合放在 Redis 这样的状态存储里有些任务不该强塞在主请求里而应该拆成提交和查询两段这部分内容真正帮我建立起来的是一种“系统视角”MySQL、Redis 和异步任务这一层真正学的不是三个工具而是一个后端系统如何管理历史、状态和长流程。下一篇预告下一篇我会继续复盘 Docker / Linux / Git / 排错这一层为什么这些看起来不像“业务代码”的能力反而是后端服务能否真正交付的关键部分。