1. 项目概述为什么Agent需要“持续工作”最近在跟几个做AI应用的朋友聊天发现大家踩的坑都出奇地一致好不容易用LangChain或者AutoGPT搭了个能跑起来的智能体Agent测试时对话流畅、任务完成得也挺漂亮。但一放到实际环境里问题就来了——用户聊到一半刷新了页面Agent的“记忆”就清零了想让它定时去检查一下数据或者发个报告发现它执行完一次就“下班”了更头疼的是一些需要长时间运行的后台任务动不动就因为超时或者进程中断而失败。这其实就是典型的“一次性Agent”困境。我们最初搭建的Agent大多是基于一个会话Session或者一次请求Request来设计的。它的生命周期紧紧绑定在用户的这次交互上交互结束Agent的“大脑”上下文、记忆、任务状态也就随之消散了。这种模式对于简单的问答或许够用但对于真正有价值的、复杂的自动化工作流来说是远远不够的。所以“Agent如何持续工作”这个命题就成了从玩具Demo走向生产级应用必须跨过的一道坎。它本质上是在解决三个核心问题状态不丢如何让Agent记住之前做了什么、正在做什么、以及接下来要做什么这就是任务持久化。离线也能干如何让Agent在用户关闭网页、断开连接后依然能默默地把交代的活儿干完这就是后台执行。到点就开工如何让Agent像闹钟一样在特定的时间自动醒来执行任务这就是定时唤醒。把这三点做好了你的Agent才真正具备了“7x24小时无人值守”的自动化能力。无论是做一个自动化的数据分析机器人、一个智能的客服工单处理系统还是一个个性化的资讯推送助手持续工作能力都是其核心骨架。接下来我就结合自己趟过的坑和实战经验把这套骨架的搭建过程拆解清楚。2. 核心架构设计构建一个“永动”Agent的蓝图在动手写代码之前我们先得把架构想明白。一个能持续工作的Agent其内部状态和生命周期管理远比一次性的复杂。我们不能把所有东西都塞在内存里而是需要一个清晰的分层设计。2.1 状态、任务与执行的分离这是最重要的设计思想。一个健壮的持续工作Agent应该将以下三者解耦状态StateAgent的“记忆”和“工作进度”。这包括会话历史和用户的对话记录。任务上下文当前正在执行的任务的目标、已完成的步骤、产生的中间数据。Agent的自身配置使用的工具Tools列表、系统提示词System Prompt、LLM的配置参数等。用户数据与特定用户相关的偏好、历史记录等。设计要点所有这些状态都必须设计成可序列化的。也就是说要能方便地转换成JSON、Pickle二进制或者数据库记录以便持久化存储。状态本身不应该包含任何无法序列化的对象比如一个打开的数据库连接、一个网络请求的Session对象。任务Task描述一件需要被完成的工作。一个任务应该包含唯一标识符如task_id用于在系统中追踪它。任务定义任务类型、目标描述、输入参数。元数据创建时间、创建者、优先级、超时时间等。生命周期状态pending等待中、running运行中、completed成功完成、failed失败、cancelled已取消。设计要点任务是一个轻量的描述对象。它不包含复杂的执行逻辑只定义“要做什么”和“当前做到哪一步了”。执行器Executor真正干活的部分。它负责从任务队列中领取一个处于pending状态的任务。根据任务定义加载或初始化一个Agent实例可能需要从持久化存储中恢复其状态。驱动Agent按步骤执行任务。在任务执行过程中定期保存Agent的状态回持久化存储检查点Checkpoint。在任务完成后无论成功失败更新任务的状态并可能触发后续操作如发送通知、创建新任务。通过这样的分离状态可以安全地存到数据库或文件里任务可以被排队和管理而执行器则可以随时被启动或停止甚至水平扩展。这就为实现后台执行和容错恢复打下了基础。2.2 持久化存储选型数据库 vs 向量库 vs 文件选择哪种方式存状态取决于你的数据特点和访问模式。存储类型典型代表适合存储的内容优点缺点应用场景关系型数据库PostgreSQL, MySQL结构化的任务元数据、用户信息、简单的键值对状态。ACID事务保证复杂查询能力强成熟稳定。不适合存储大块的、嵌套深的JSON状态检索非结构化内容效率低。任务调度表、用户管理、基础配置。文档数据库MongoDB, CouchDBAgent的完整状态对象大的JSON文档、任务详情、会话历史。模式灵活直接存储JSON读写速度快适合嵌套数据。事务支持相对较弱需要关注数据一致性设计。Agent状态持久化的主力存储每次检查点的快照。向量数据库Pinecone, Weaviate, Qdrant从对话历史或任务上下文中提取的嵌入向量用于语义搜索和长期记忆检索。能实现“基于含义的回忆”让Agent拥有真正的长期记忆。不能替代主状态存储主要用于增强记忆检索能力。成本可能较高。实现“记忆检索”功能例如“上周我让你分析过销售数据结论是什么”分布式缓存Redis高频访问的临时状态、任务队列、分布式锁。内存级速度支持丰富的数据结构List做队列String存状态。数据易失可配置持久化但性能下降容量有限。作为任务队列的承载存储短期会话缓存实现分布式锁防止任务重复执行。文件系统本地磁盘、对象存储S3任务产生的大型文件如生成的报告、图片、模型文件、备份。存储海量非结构化数据成本低易于扩展。管理复杂不适合频繁读写的小状态。存储任务输出产物模型文件托管。实战心得混合存储是常态在实际项目中几乎没有只用一种存储的。一个典型的架构是PostgreSQL存tasks表任务ID、状态、创建时间等和users表。MongoDB存agent_sessions集合每个文档是一个Agent在某个时间点的完整状态快照通过session_id和task_id关联。Redis用List实现一个简单的任务队列pending_tasks用String缓存当前活跃的task_id - executor_id映射用SETNX实现分布式锁。Pinecone索引所有会话历史的关键片段向量供Agent快速检索相关记忆。注意选择文档数据库如MongoDB存储Agent状态时要特别注意文档大小。如果状态对象非常庞大比如包含了很长的对话历史可能会影响读写性能并触发数据库的限制。一个优化策略是进行“状态差分”只存储相对于上一个检查点发生变化的部分而不是全量存储。2.3 后台执行与定时唤醒的引擎有了状态存储和任务定义我们需要一个“发动机”来驱动一切。后台执行的核心任务队列这是实现后台执行最经典的模式。当用户前端触发一个长时间任务时API接口并不直接执行而是在数据库中创建一条pending状态的任务记录。将该任务的task_id推入Redis的pending_tasks列表。立即向用户返回task_id和“任务已提交”的响应。 此时用户的请求立即结束无需等待。后台有一个或多个独立的工作进程Worker在持续监听pending_tasks队列。它们从队列中取出task_id加载对应的任务定义和Agent状态开始执行并更新任务状态。用户可以通过另一个API凭task_id来轮询查询任务进度和结果。定时唤醒的核心定时任务调度器对于“每天上午9点生成报告”这类需求我们需要一个调度器。同样不建议在Agent框架内自己造轮子成熟的方案有Celery Beat如果你用Celery作为任务队列它的Beat组件就是一个内置的定时调度器配置非常方便。APScheduler一个轻量级但功能强大的Python库支持固定间隔、定时cron等多种触发方式可以集成到你的工作进程中。操作系统Cron最原始但最可靠的方式。可以写一个简单的脚本用cron定时调用这个脚本负责向你的任务队列里推送一个定时任务。Kubernetes CronJob如果你的应用部署在K8s上用CronJob来启动一个Pod这个Pod的唯一工作就是触发一次定时任务触发完成后Pod自动销毁非常干净。设计要点无论是后台任务还是定时任务最终都应该被转化为一个标准的“任务”对象推入同一个任务队列由同一套工作进程来处理。这样能极大简化系统架构保证执行逻辑的一致性。3. 实战实现从零搭建一个带持久化能力的Agent系统理论说再多不如一行代码。我们用一个简化但完整的例子来实现一个具备任务持久化、后台执行和定时唤醒能力的“天气查询与报告Agent”。这个Agent能接受用户提交的“生成某城市一周天气报告”的任务在后台执行并且支持每天自动生成指定城市的天气简报。3.1 环境准备与依赖安装我们选择Python作为实现语言使用LangChain框架来构建Agent核心因为它生态成熟易于扩展。# 创建项目并安装核心依赖 pip install langchain langchain-openai # Agent核心框架和LLM pip install pymongo redis # 持久化存储和缓存 pip install celery[redis] apscheduler # 任务队列和定时调度 pip install python-dotenv # 管理环境变量这里我们选择MongoDB作为主状态存储。Redis作为Celery的消息代理Broker和结果后端Result Backend同时也用作缓存。Celery APSchedulerCelery处理异步任务队列APScheduler嵌入到Celery Worker中实现定时调度。3.2 定义数据模型与存储层首先定义我们的核心数据模型。# models.py from pydantic import BaseModel, Field from datetime import datetime from typing import Optional, Dict, Any, List from enum import Enum class TaskStatus(str, Enum): PENDING pending RUNNING running COMPLETED completed FAILED failed CANCELLED cancelled class Task(BaseModel): 任务定义 task_id: str Field(default_factorylambda: str(uuid.uuid4())) name: str # 任务名称如 generate_weather_report params: Dict[str, Any] # 任务参数如 {city: 北京, days: 7} status: TaskStatus TaskStatus.PENDING created_at: datetime Field(default_factorydatetime.utcnow) started_at: Optional[datetime] None finished_at: Optional[datetime] None result: Optional[Dict[str, Any]] None # 任务执行结果 error: Optional[str] None # 如果失败错误信息 class AgentSession(BaseModel): Agent会话状态快照 session_id: str Field(default_factorylambda: str(uuid.uuid4())) task_id: str # 关联的任务ID # 使用LangChain的BaseMessage序列化存储对话历史 message_history: List[Dict[str, Any]] # 存储序列化后的消息 # 存储Agent运行时的其他上下文如已使用的工具调用记录 context: Dict[str, Any] Field(default_factorydict) created_at: datetime Field(default_factorydatetime.utcnow) updated_at: datetime Field(default_factorydatetime.utcnow) checkpoint_id: int 0 # 检查点版本号用于增量保存然后实现一个简单的存储管理类封装对MongoDB的操作。# storage.py from pymongo import MongoClient, ReturnDocument from typing import Optional import json from models import Task, AgentSession, TaskStatus class StorageManager: def __init__(self, mongo_uri: str, db_name: str): self.client MongoClient(mongo_uri) self.db self.client[db_name] self.tasks self.db.tasks self.sessions self.db.agent_sessions # --- 任务管理 --- def create_task(self, task: Task) - str: task_dict task.dict() result self.tasks.insert_one(task_dict) return str(result.inserted_id) def get_task(self, task_id: str) - Optional[Task]: doc self.tasks.find_one({task_id: task_id}) return Task(**doc) if doc else None def update_task_status(self, task_id: str, status: TaskStatus, resultNone, errorNone): update_data {status: status, updated_at: datetime.utcnow()} if status TaskStatus.RUNNING: update_data[started_at] datetime.utcnow() elif status in [TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED]: update_data[finished_at] datetime.utcnow() if result is not None: update_data[result] result if error is not None: update_data[error] error self.tasks.update_one({task_id: task_id}, {$set: update_data}) # --- Agent会话管理 --- def save_agent_session(self, session: AgentSession) - str: 保存或更新Agent会话状态 session.updated_at datetime.utcnow() session_dict session.dict() # 使用task_id和checkpoint_id作为复合查询条件实现版本化保存或更新 result self.sessions.update_one( {task_id: session.task_id, checkpoint_id: session.checkpoint_id}, {$set: session_dict}, upsertTrue ) return session.session_id def load_latest_agent_session(self, task_id: str) - Optional[AgentSession]: 加载指定任务最新的Agent会话状态 doc self.sessions.find_one( {task_id: task_id}, sort[(checkpoint_id, -1)] # 按检查点ID降序取最新的 ) return AgentSession(**doc) if doc else None3.3 构建可持久化的Agent执行器这是最核心的部分。我们需要改造传统的LangChain Agent使其状态可保存、可恢复。# agent_executor.py from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain.memory import ConversationBufferMemory from langchain_core.messages import BaseMessage, HumanMessage, AIMessage from langchain.tools import Tool from storage import StorageManager from models import AgentSession import asyncio from typing import List, Dict, Any class PersistentAgentExecutor: def __init__(self, storage: StorageManager, llm_modelgpt-3.5-turbo): self.storage storage self.llm ChatOpenAI(modelllm_model, temperature0) # 定义工具例如一个模拟的天气查询工具 self.tools [ Tool( nameget_weather, funcself._get_weather_mock, description根据城市名和天数查询天气预报。输入应为 城市,天数例如 北京,7。 ) ] # 注意这里不立即创建LangChain的AgentExecutor因为它包含不可序列化的运行时对象。 # 我们将在每次执行时根据加载的会话状态来动态构建。 def _get_weather_mock(self, query: str) - str: 模拟天气查询工具 # 实际项目中应调用真实的天气API city, days query.split(,) return f{city}未来{days}天的天气情况第一天晴第二天多云...模拟数据 def _messages_to_dict(self, messages: List[BaseMessage]) - List[Dict]: 将LangChain消息对象序列化为字典列表 return [{type: msg.__class__.__name__, content: msg.content} for msg in messages] def _dict_to_messages(self, message_dicts: List[Dict]) - List[BaseMessage]: 将字典列表反序列化为LangChain消息对象 messages [] for msg in message_dicts: if msg[type] HumanMessage: messages.append(HumanMessage(contentmsg[content])) elif msg[type] AIMessage: messages.append(AIMessage(contentmsg[content])) # ... 处理其他消息类型 return messages def _create_agent_executor(self, message_history: List[BaseMessage]): 根据历史消息创建一个新的LangChain AgentExecutor实例 # 1. 创建Memory并注入历史 memory ConversationBufferMemory(return_messagesTrue) for msg in message_history: if isinstance(msg, HumanMessage): memory.chat_memory.add_user_message(msg.content) elif isinstance(msg, AIMessage): memory.chat_memory.add_ai_message(msg.content) # 2. 创建Agent prompt ... # 你的Agent提示词模板 agent create_openai_tools_agent(self.llm, self.tools, prompt) # 3. 创建执行器 agent_executor AgentExecutor(agentagent, toolsself.tools, memorymemory, verboseTrue) return agent_executor, memory async def execute_task(self, task_id: str, user_input: str): 执行一个任务的核心方法 # 1. 从存储加载最新的会话状态 session self.storage.load_latest_agent_session(task_id) if session: # 恢复历史消息 message_history self._dict_to_messages(session.message_history) # 恢复其他上下文如果有 restored_context session.context else: # 全新任务初始化空状态 message_history [] restored_context {} session AgentSession(task_idtask_id, message_history[], context{}) # 2. 创建新的AgentExecutor实例每次执行都新建保证干净 agent_executor, memory self._create_agent_executor(message_history) # 3. 执行这里调用Agent try: # 将用户输入添加到memory以便本次执行能“看到” memory.chat_memory.add_user_message(user_input) # 调用Agent response await agent_executor.ainvoke({input: user_input}) ai_response response[output] memory.chat_memory.add_ai_message(ai_response) # 4. 保存检查点持久化状态 # 获取当前完整的消息历史 current_messages memory.chat_memory.messages session.message_history self._messages_to_dict(current_messages) # 可以在这里保存一些运行时上下文例如本次执行中工具调用的结果摘要 # session.context.update({last_tool_used: ...}) session.checkpoint_id 1 # 版本号递增 self.storage.save_agent_session(session) return ai_response except Exception as e: # 执行失败也需要保存状态吗视情况而定。对于可重试的错误可以保存。 # 这里我们记录错误但不保存失败状态的消息历史因为可能不完整。 raise e关键点解析无状态执行器PersistentAgentExecutor本身不持有对话状态。状态全部来自外部存储session。动态创建每次execute_task被调用时都根据加载的session重新创建LangChain的AgentExecutor和Memory对象。这避免了在长时间运行的后台任务中内存中的对象状态变得复杂或难以序列化的问题。检查点机制每次Agent完成一轮思考和行动即ainvoke返回后我们立即将最新的完整消息历史保存回数据库。这就是“检查点”。如果任务执行到一半进程崩溃重启后可以从上一个检查点恢复而不是从头开始。会话关联任务通过task_id将会话与任务强关联。一个复杂的长期任务可能由多轮对话多次execute_task调用完成它们共享同一个session。3.4 集成Celery实现后台任务队列现在我们将上面的执行器包装成Celery任务。# tasks.py (Celery tasks) from celery import Celery from storage import StorageManager from agent_executor import PersistentAgentExecutor from models import Task, TaskStatus import os # 初始化Celery应用使用Redis作为消息代理 redis_url os.getenv(REDIS_URL, redis://localhost:6379/0) celery_app Celery(agent_worker, brokerredis_url, backendredis_url) # 初始化全局组件Celery Worker启动时加载 storage StorageManager(os.getenv(MONGO_URI), agent_db) agent_executor PersistentAgentExecutor(storage) celery_app.task(bindTrue, namerun_agent_task) def run_agent_task(self, task_id: str, user_input: str): Celery任务执行Agent任务 # 1. 更新任务状态为运行中 storage.update_task_status(task_id, TaskStatus.RUNNING) try: # 2. 执行Agent注意Celery任务函数本身不是async我们需要同步运行async函数 # 这里使用asyncio.run来运行异步的execute_task loop asyncio.get_event_loop() except RuntimeError: # 如果在没有事件循环的线程中就新建一个 loop asyncio.new_event_loop() asyncio.set_event_loop(loop) result loop.run_until_complete( agent_executor.execute_task(task_id, user_input) ) loop.close() else: result loop.run_until_complete( agent_executor.execute_task(task_id, user_input) ) # 3. 更新任务状态为完成并保存结果 storage.update_task_status(task_id, TaskStatus.COMPLETED, result{output: result}) return result # 一个用于定时任务的“触发器”任务 celery_app.task(nametrigger_daily_weather_report) def trigger_daily_weather_report(): 定时任务创建每日天气报告任务 # 这里可以读取配置获取需要每日生成报告的城市列表 cities [北京, 上海, 深圳] for city in cities: # 为每个城市创建一个后台任务 task Task( namedaily_weather_report, params{city: city, days: 1} ) task_id storage.create_task(task) # 将任务ID推送到队列等待工作进程处理 # 注意这里我们直接调用run_agent_task.delay将任务发给Celery Worker。 # 但更常见的做法是定时任务只负责创建任务记录由另一个常驻的“任务分发器”来推队列。 # 这里为了简化我们直接推。 run_agent_task.delay(task_id, f生成{city}今天的天气简报。)前端API接口会变得非常轻量# api.py (FastAPI示例) from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel from tasks import run_agent_task, storage from models import Task, TaskStatus app FastAPI() class TaskRequest(BaseModel): user_input: str app.post(/submit_task) async def submit_task(request: TaskRequest, background_tasks: BackgroundTasks): 提交一个异步Agent任务 # 1. 创建任务记录 task Task(nameuser_query, params{user_input: request.user_input}) task_id storage.create_task(task) # 2. 将任务推送到Celery队列异步执行 run_agent_task.delay(task_id, request.user_input) # 3. 立即返回任务ID return {task_id: task_id, status: submitted, message: 任务已提交到后台处理} app.get(/task_status/{task_id}) async def get_task_status(task_id: str): 查询任务状态和结果 task storage.get_task(task_id) if not task: return {error: Task not found} return task.dict()3.5 集成APScheduler实现定时唤醒我们需要在Celery Worker进程中启动一个定时调度器。这可以通过自定义Celery的启动信号worker_ready)来实现。# celery_scheduler.py from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger from celery.signals import worker_ready # 创建一个全局调度器 scheduler BackgroundScheduler() worker_ready.connect def on_worker_ready(sender, **kwargs): 当Celery Worker启动就绪时启动定时调度器 if not scheduler.running: scheduler.start() # 添加定时任务每天上午9点触发 scheduler.add_job( functrigger_daily_weather_report, # 这是上面定义的Celery任务函数 triggerCronTrigger(hour9, minute0), iddaily_weather_job, replace_existingTrue ) print(APScheduler started with daily weather report job.)关键点trigger_daily_weather_report是一个Celery任务函数用celery_app.task装饰。scheduler.add_job只是安排这个函数在特定时间被调用。当它被调用时它会在Celery的上下文中执行从而能够安全地调用run_agent_task.delay()来创建真正的后台任务。这样就实现了“定时唤醒”并创建任务。4. 部署、监控与问题排查实录系统搭建好了但要稳定运行在生产环境还有一大堆“坑”要填。4.1 部署架构与配置要点一个最小化的生产部署可能包含以下服务[Docker Compose 示例] version: 3.8 services: mongodb: image: mongo:latest volumes: - mongodb_data:/data/db redis: image: redis:alpine api: build: ./api ports: - 8000:8000 environment: - MONGO_URImongodb://mongodb:27017 - REDIS_URLredis://redis:6379/0 depends_on: - mongodb - redis celery_worker: build: ./worker command: celery -A tasks.celery_app worker --loglevelinfo --concurrency4 environment: - MONGO_URImongodb://mongodb:27017 - REDIS_URLredis://redis:6379/0 depends_on: - mongodb - redis # 注意需要挂载包含celery_scheduler.py的代码卷确保worker启动时加载调度器配置要点Celery并发数--concurrency参数需要根据你的机器CPU核心数和任务类型I/O密集型还是CPU密集型来调整。对于大量调用LLM API的Agent任务通常是I/O密集型可以设置得高一些如CPU核心数的2-4倍。Redis持久化务必配置Redis的持久化RDB或AOF防止服务器重启导致内存中的任务队列丢失。MongoDB索引在agent_sessions集合的task_id和checkpoint_id字段上创建复合索引加速会话加载。在tasks集合的status和created_at字段上创建索引方便后台管理界面查询。网络超时与重试在Agent执行器内部调用LLM API或外部工具时必须设置合理的超时和重试机制避免一个任务卡死整个工作进程。4.2 常见问题与排查技巧在实际运行中你一定会遇到下面这些问题问题1任务卡在running状态永不结束。可能原因Agent陷入死循环LLM可能在一个思考-行动循环中出不来。外部API调用失败或超时没有正确处理异常。工作进程Celery Worker崩溃但任务状态没被重置。排查与解决设置任务超时在Celery任务装饰器中设置soft_time_limit和time_limit。celery_app.task(bindTrue, soft_time_limit300, time_limit330) def run_agent_task(self, ...): ...实现心跳与看门狗在任务函数中定期更新一个“最后活跃时间戳”到Redis。另一个监控进程检查所有running状态的任务如果其“最后活跃时间戳”超过阈值如10分钟则强制将其状态标记为failed并记录原因。增强Agent的约束在Agent的提示词Prompt中明确限制最大循环次数如“最多进行5次工具调用”并在代码中强制中断。问题2定时任务没有准时执行或者重复执行。可能原因时钟不同步部署多台Worker时服务器时间不一致。APScheduler在多个Worker中重复启动每个Celery Worker进程都启动了自己的调度器导致同一个定时任务被多次触发。排查与解决使用分布式锁在worker_ready信号中在启动调度器之前先尝试获取一个Redis分布式锁。只有一个Worker能成功获取锁并启动调度器。import redis redis_client redis.from_url(os.getenv(REDIS_URL)) lock_key scheduler_lock lock redis_client.lock(lock_key, timeout60) if lock.acquire(blockingFalse): if not scheduler.running: scheduler.start() # ... 添加任务 # 注意不要释放锁让这个锁一直持有直到进程结束。使用外部分布式调度器对于更复杂的生产环境可以考虑使用Celery Beat作为独立的调度服务或者使用Kubernetes CronJob避免在应用内做调度。问题3Agent状态恢复后行为不一致或出错。可能原因序列化/反序列化丢失信息自定义的工具Tool或记忆Memory对象可能包含无法序列化的属性。代码版本不一致保存状态的代码版本和恢复状态的代码版本不同导致数据模型不兼容。排查与解决简化状态对象持久化时只保存最核心、最原始的数据如消息列表的文本内容。恢复时用这些数据重新构建运行时对象。版本化数据模型在AgentSession模型中增加一个schema_version字段。每次数据模型有重大变更时升级版本号并在加载旧版本数据时编写迁移脚本。全面测试建立完整的集成测试覆盖“执行 - 保存 - 重启 - 恢复 - 继续执行”的全流程。问题4数据库连接数暴涨或存储性能瓶颈。可能原因检查点过于频繁每轮对话都保存全量状态在高并发下对数据库压力大。没有使用连接池每次保存/加载都新建数据库连接。排查与解决优化检查点策略改为每N轮对话保存一次或者只在关键步骤如调用重要工具后保存。使用连接池确保MongoDB和Redis的客户端配置了连接池。异步写入如果状态不是需要严格实时持久化可以考虑将保存操作放入一个低优先级的队列异步执行不阻塞主任务流程。4.3 监控与运维建议要让这套系统稳定运行基本的监控必不可少任务大盘做一个简单的管理后台展示任务总数、各状态任务的数量、平均执行时间、失败率等。数据可以直接从MongoDB的tasks集合聚合查询。队列监控监控Redis中任务队列的长度。如果pending_tasks队列持续增长说明Worker处理不过来需要扩容。Agent执行质量监控记录每个任务最终消耗的Token数、工具调用次数、执行轮数。异常高的数值可能意味着Agent陷入了低效循环。日志聚合将所有服务API、Celery Worker的日志集中收集到ELK或Loki等平台方便通过task_id追踪一个任务在所有服务中的完整生命周期日志。最后再分享一个我踩过的大坑LLM上下文长度限制与长程记忆的冲突。我们的持久化方案保存了完整的对话历史但当任务步骤非常多历史消息超过LLM的上下文窗口时直接全部灌给LLM是行不通的。这时就需要引入摘要记忆或向量检索记忆。在保存检查点时不仅保存原始消息还可以用另一个LLM调用对历史对话进行摘要将摘要和最近几条原始消息一起保存。在恢复时优先加载摘要再根据需要从向量库中检索相关的历史片段。这又是另一个复杂但有趣的话题了。
1. 项目概述为什么Agent需要“持续工作”最近在跟几个做AI应用的朋友聊天发现大家踩的坑都出奇地一致好不容易用LangChain或者AutoGPT搭了个能跑起来的智能体Agent测试时对话流畅、任务完成得也挺漂亮。但一放到实际环境里问题就来了——用户聊到一半刷新了页面Agent的“记忆”就清零了想让它定时去检查一下数据或者发个报告发现它执行完一次就“下班”了更头疼的是一些需要长时间运行的后台任务动不动就因为超时或者进程中断而失败。这其实就是典型的“一次性Agent”困境。我们最初搭建的Agent大多是基于一个会话Session或者一次请求Request来设计的。它的生命周期紧紧绑定在用户的这次交互上交互结束Agent的“大脑”上下文、记忆、任务状态也就随之消散了。这种模式对于简单的问答或许够用但对于真正有价值的、复杂的自动化工作流来说是远远不够的。所以“Agent如何持续工作”这个命题就成了从玩具Demo走向生产级应用必须跨过的一道坎。它本质上是在解决三个核心问题状态不丢如何让Agent记住之前做了什么、正在做什么、以及接下来要做什么这就是任务持久化。离线也能干如何让Agent在用户关闭网页、断开连接后依然能默默地把交代的活儿干完这就是后台执行。到点就开工如何让Agent像闹钟一样在特定的时间自动醒来执行任务这就是定时唤醒。把这三点做好了你的Agent才真正具备了“7x24小时无人值守”的自动化能力。无论是做一个自动化的数据分析机器人、一个智能的客服工单处理系统还是一个个性化的资讯推送助手持续工作能力都是其核心骨架。接下来我就结合自己趟过的坑和实战经验把这套骨架的搭建过程拆解清楚。2. 核心架构设计构建一个“永动”Agent的蓝图在动手写代码之前我们先得把架构想明白。一个能持续工作的Agent其内部状态和生命周期管理远比一次性的复杂。我们不能把所有东西都塞在内存里而是需要一个清晰的分层设计。2.1 状态、任务与执行的分离这是最重要的设计思想。一个健壮的持续工作Agent应该将以下三者解耦状态StateAgent的“记忆”和“工作进度”。这包括会话历史和用户的对话记录。任务上下文当前正在执行的任务的目标、已完成的步骤、产生的中间数据。Agent的自身配置使用的工具Tools列表、系统提示词System Prompt、LLM的配置参数等。用户数据与特定用户相关的偏好、历史记录等。设计要点所有这些状态都必须设计成可序列化的。也就是说要能方便地转换成JSON、Pickle二进制或者数据库记录以便持久化存储。状态本身不应该包含任何无法序列化的对象比如一个打开的数据库连接、一个网络请求的Session对象。任务Task描述一件需要被完成的工作。一个任务应该包含唯一标识符如task_id用于在系统中追踪它。任务定义任务类型、目标描述、输入参数。元数据创建时间、创建者、优先级、超时时间等。生命周期状态pending等待中、running运行中、completed成功完成、failed失败、cancelled已取消。设计要点任务是一个轻量的描述对象。它不包含复杂的执行逻辑只定义“要做什么”和“当前做到哪一步了”。执行器Executor真正干活的部分。它负责从任务队列中领取一个处于pending状态的任务。根据任务定义加载或初始化一个Agent实例可能需要从持久化存储中恢复其状态。驱动Agent按步骤执行任务。在任务执行过程中定期保存Agent的状态回持久化存储检查点Checkpoint。在任务完成后无论成功失败更新任务的状态并可能触发后续操作如发送通知、创建新任务。通过这样的分离状态可以安全地存到数据库或文件里任务可以被排队和管理而执行器则可以随时被启动或停止甚至水平扩展。这就为实现后台执行和容错恢复打下了基础。2.2 持久化存储选型数据库 vs 向量库 vs 文件选择哪种方式存状态取决于你的数据特点和访问模式。存储类型典型代表适合存储的内容优点缺点应用场景关系型数据库PostgreSQL, MySQL结构化的任务元数据、用户信息、简单的键值对状态。ACID事务保证复杂查询能力强成熟稳定。不适合存储大块的、嵌套深的JSON状态检索非结构化内容效率低。任务调度表、用户管理、基础配置。文档数据库MongoDB, CouchDBAgent的完整状态对象大的JSON文档、任务详情、会话历史。模式灵活直接存储JSON读写速度快适合嵌套数据。事务支持相对较弱需要关注数据一致性设计。Agent状态持久化的主力存储每次检查点的快照。向量数据库Pinecone, Weaviate, Qdrant从对话历史或任务上下文中提取的嵌入向量用于语义搜索和长期记忆检索。能实现“基于含义的回忆”让Agent拥有真正的长期记忆。不能替代主状态存储主要用于增强记忆检索能力。成本可能较高。实现“记忆检索”功能例如“上周我让你分析过销售数据结论是什么”分布式缓存Redis高频访问的临时状态、任务队列、分布式锁。内存级速度支持丰富的数据结构List做队列String存状态。数据易失可配置持久化但性能下降容量有限。作为任务队列的承载存储短期会话缓存实现分布式锁防止任务重复执行。文件系统本地磁盘、对象存储S3任务产生的大型文件如生成的报告、图片、模型文件、备份。存储海量非结构化数据成本低易于扩展。管理复杂不适合频繁读写的小状态。存储任务输出产物模型文件托管。实战心得混合存储是常态在实际项目中几乎没有只用一种存储的。一个典型的架构是PostgreSQL存tasks表任务ID、状态、创建时间等和users表。MongoDB存agent_sessions集合每个文档是一个Agent在某个时间点的完整状态快照通过session_id和task_id关联。Redis用List实现一个简单的任务队列pending_tasks用String缓存当前活跃的task_id - executor_id映射用SETNX实现分布式锁。Pinecone索引所有会话历史的关键片段向量供Agent快速检索相关记忆。注意选择文档数据库如MongoDB存储Agent状态时要特别注意文档大小。如果状态对象非常庞大比如包含了很长的对话历史可能会影响读写性能并触发数据库的限制。一个优化策略是进行“状态差分”只存储相对于上一个检查点发生变化的部分而不是全量存储。2.3 后台执行与定时唤醒的引擎有了状态存储和任务定义我们需要一个“发动机”来驱动一切。后台执行的核心任务队列这是实现后台执行最经典的模式。当用户前端触发一个长时间任务时API接口并不直接执行而是在数据库中创建一条pending状态的任务记录。将该任务的task_id推入Redis的pending_tasks列表。立即向用户返回task_id和“任务已提交”的响应。 此时用户的请求立即结束无需等待。后台有一个或多个独立的工作进程Worker在持续监听pending_tasks队列。它们从队列中取出task_id加载对应的任务定义和Agent状态开始执行并更新任务状态。用户可以通过另一个API凭task_id来轮询查询任务进度和结果。定时唤醒的核心定时任务调度器对于“每天上午9点生成报告”这类需求我们需要一个调度器。同样不建议在Agent框架内自己造轮子成熟的方案有Celery Beat如果你用Celery作为任务队列它的Beat组件就是一个内置的定时调度器配置非常方便。APScheduler一个轻量级但功能强大的Python库支持固定间隔、定时cron等多种触发方式可以集成到你的工作进程中。操作系统Cron最原始但最可靠的方式。可以写一个简单的脚本用cron定时调用这个脚本负责向你的任务队列里推送一个定时任务。Kubernetes CronJob如果你的应用部署在K8s上用CronJob来启动一个Pod这个Pod的唯一工作就是触发一次定时任务触发完成后Pod自动销毁非常干净。设计要点无论是后台任务还是定时任务最终都应该被转化为一个标准的“任务”对象推入同一个任务队列由同一套工作进程来处理。这样能极大简化系统架构保证执行逻辑的一致性。3. 实战实现从零搭建一个带持久化能力的Agent系统理论说再多不如一行代码。我们用一个简化但完整的例子来实现一个具备任务持久化、后台执行和定时唤醒能力的“天气查询与报告Agent”。这个Agent能接受用户提交的“生成某城市一周天气报告”的任务在后台执行并且支持每天自动生成指定城市的天气简报。3.1 环境准备与依赖安装我们选择Python作为实现语言使用LangChain框架来构建Agent核心因为它生态成熟易于扩展。# 创建项目并安装核心依赖 pip install langchain langchain-openai # Agent核心框架和LLM pip install pymongo redis # 持久化存储和缓存 pip install celery[redis] apscheduler # 任务队列和定时调度 pip install python-dotenv # 管理环境变量这里我们选择MongoDB作为主状态存储。Redis作为Celery的消息代理Broker和结果后端Result Backend同时也用作缓存。Celery APSchedulerCelery处理异步任务队列APScheduler嵌入到Celery Worker中实现定时调度。3.2 定义数据模型与存储层首先定义我们的核心数据模型。# models.py from pydantic import BaseModel, Field from datetime import datetime from typing import Optional, Dict, Any, List from enum import Enum class TaskStatus(str, Enum): PENDING pending RUNNING running COMPLETED completed FAILED failed CANCELLED cancelled class Task(BaseModel): 任务定义 task_id: str Field(default_factorylambda: str(uuid.uuid4())) name: str # 任务名称如 generate_weather_report params: Dict[str, Any] # 任务参数如 {city: 北京, days: 7} status: TaskStatus TaskStatus.PENDING created_at: datetime Field(default_factorydatetime.utcnow) started_at: Optional[datetime] None finished_at: Optional[datetime] None result: Optional[Dict[str, Any]] None # 任务执行结果 error: Optional[str] None # 如果失败错误信息 class AgentSession(BaseModel): Agent会话状态快照 session_id: str Field(default_factorylambda: str(uuid.uuid4())) task_id: str # 关联的任务ID # 使用LangChain的BaseMessage序列化存储对话历史 message_history: List[Dict[str, Any]] # 存储序列化后的消息 # 存储Agent运行时的其他上下文如已使用的工具调用记录 context: Dict[str, Any] Field(default_factorydict) created_at: datetime Field(default_factorydatetime.utcnow) updated_at: datetime Field(default_factorydatetime.utcnow) checkpoint_id: int 0 # 检查点版本号用于增量保存然后实现一个简单的存储管理类封装对MongoDB的操作。# storage.py from pymongo import MongoClient, ReturnDocument from typing import Optional import json from models import Task, AgentSession, TaskStatus class StorageManager: def __init__(self, mongo_uri: str, db_name: str): self.client MongoClient(mongo_uri) self.db self.client[db_name] self.tasks self.db.tasks self.sessions self.db.agent_sessions # --- 任务管理 --- def create_task(self, task: Task) - str: task_dict task.dict() result self.tasks.insert_one(task_dict) return str(result.inserted_id) def get_task(self, task_id: str) - Optional[Task]: doc self.tasks.find_one({task_id: task_id}) return Task(**doc) if doc else None def update_task_status(self, task_id: str, status: TaskStatus, resultNone, errorNone): update_data {status: status, updated_at: datetime.utcnow()} if status TaskStatus.RUNNING: update_data[started_at] datetime.utcnow() elif status in [TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED]: update_data[finished_at] datetime.utcnow() if result is not None: update_data[result] result if error is not None: update_data[error] error self.tasks.update_one({task_id: task_id}, {$set: update_data}) # --- Agent会话管理 --- def save_agent_session(self, session: AgentSession) - str: 保存或更新Agent会话状态 session.updated_at datetime.utcnow() session_dict session.dict() # 使用task_id和checkpoint_id作为复合查询条件实现版本化保存或更新 result self.sessions.update_one( {task_id: session.task_id, checkpoint_id: session.checkpoint_id}, {$set: session_dict}, upsertTrue ) return session.session_id def load_latest_agent_session(self, task_id: str) - Optional[AgentSession]: 加载指定任务最新的Agent会话状态 doc self.sessions.find_one( {task_id: task_id}, sort[(checkpoint_id, -1)] # 按检查点ID降序取最新的 ) return AgentSession(**doc) if doc else None3.3 构建可持久化的Agent执行器这是最核心的部分。我们需要改造传统的LangChain Agent使其状态可保存、可恢复。# agent_executor.py from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain.memory import ConversationBufferMemory from langchain_core.messages import BaseMessage, HumanMessage, AIMessage from langchain.tools import Tool from storage import StorageManager from models import AgentSession import asyncio from typing import List, Dict, Any class PersistentAgentExecutor: def __init__(self, storage: StorageManager, llm_modelgpt-3.5-turbo): self.storage storage self.llm ChatOpenAI(modelllm_model, temperature0) # 定义工具例如一个模拟的天气查询工具 self.tools [ Tool( nameget_weather, funcself._get_weather_mock, description根据城市名和天数查询天气预报。输入应为 城市,天数例如 北京,7。 ) ] # 注意这里不立即创建LangChain的AgentExecutor因为它包含不可序列化的运行时对象。 # 我们将在每次执行时根据加载的会话状态来动态构建。 def _get_weather_mock(self, query: str) - str: 模拟天气查询工具 # 实际项目中应调用真实的天气API city, days query.split(,) return f{city}未来{days}天的天气情况第一天晴第二天多云...模拟数据 def _messages_to_dict(self, messages: List[BaseMessage]) - List[Dict]: 将LangChain消息对象序列化为字典列表 return [{type: msg.__class__.__name__, content: msg.content} for msg in messages] def _dict_to_messages(self, message_dicts: List[Dict]) - List[BaseMessage]: 将字典列表反序列化为LangChain消息对象 messages [] for msg in message_dicts: if msg[type] HumanMessage: messages.append(HumanMessage(contentmsg[content])) elif msg[type] AIMessage: messages.append(AIMessage(contentmsg[content])) # ... 处理其他消息类型 return messages def _create_agent_executor(self, message_history: List[BaseMessage]): 根据历史消息创建一个新的LangChain AgentExecutor实例 # 1. 创建Memory并注入历史 memory ConversationBufferMemory(return_messagesTrue) for msg in message_history: if isinstance(msg, HumanMessage): memory.chat_memory.add_user_message(msg.content) elif isinstance(msg, AIMessage): memory.chat_memory.add_ai_message(msg.content) # 2. 创建Agent prompt ... # 你的Agent提示词模板 agent create_openai_tools_agent(self.llm, self.tools, prompt) # 3. 创建执行器 agent_executor AgentExecutor(agentagent, toolsself.tools, memorymemory, verboseTrue) return agent_executor, memory async def execute_task(self, task_id: str, user_input: str): 执行一个任务的核心方法 # 1. 从存储加载最新的会话状态 session self.storage.load_latest_agent_session(task_id) if session: # 恢复历史消息 message_history self._dict_to_messages(session.message_history) # 恢复其他上下文如果有 restored_context session.context else: # 全新任务初始化空状态 message_history [] restored_context {} session AgentSession(task_idtask_id, message_history[], context{}) # 2. 创建新的AgentExecutor实例每次执行都新建保证干净 agent_executor, memory self._create_agent_executor(message_history) # 3. 执行这里调用Agent try: # 将用户输入添加到memory以便本次执行能“看到” memory.chat_memory.add_user_message(user_input) # 调用Agent response await agent_executor.ainvoke({input: user_input}) ai_response response[output] memory.chat_memory.add_ai_message(ai_response) # 4. 保存检查点持久化状态 # 获取当前完整的消息历史 current_messages memory.chat_memory.messages session.message_history self._messages_to_dict(current_messages) # 可以在这里保存一些运行时上下文例如本次执行中工具调用的结果摘要 # session.context.update({last_tool_used: ...}) session.checkpoint_id 1 # 版本号递增 self.storage.save_agent_session(session) return ai_response except Exception as e: # 执行失败也需要保存状态吗视情况而定。对于可重试的错误可以保存。 # 这里我们记录错误但不保存失败状态的消息历史因为可能不完整。 raise e关键点解析无状态执行器PersistentAgentExecutor本身不持有对话状态。状态全部来自外部存储session。动态创建每次execute_task被调用时都根据加载的session重新创建LangChain的AgentExecutor和Memory对象。这避免了在长时间运行的后台任务中内存中的对象状态变得复杂或难以序列化的问题。检查点机制每次Agent完成一轮思考和行动即ainvoke返回后我们立即将最新的完整消息历史保存回数据库。这就是“检查点”。如果任务执行到一半进程崩溃重启后可以从上一个检查点恢复而不是从头开始。会话关联任务通过task_id将会话与任务强关联。一个复杂的长期任务可能由多轮对话多次execute_task调用完成它们共享同一个session。3.4 集成Celery实现后台任务队列现在我们将上面的执行器包装成Celery任务。# tasks.py (Celery tasks) from celery import Celery from storage import StorageManager from agent_executor import PersistentAgentExecutor from models import Task, TaskStatus import os # 初始化Celery应用使用Redis作为消息代理 redis_url os.getenv(REDIS_URL, redis://localhost:6379/0) celery_app Celery(agent_worker, brokerredis_url, backendredis_url) # 初始化全局组件Celery Worker启动时加载 storage StorageManager(os.getenv(MONGO_URI), agent_db) agent_executor PersistentAgentExecutor(storage) celery_app.task(bindTrue, namerun_agent_task) def run_agent_task(self, task_id: str, user_input: str): Celery任务执行Agent任务 # 1. 更新任务状态为运行中 storage.update_task_status(task_id, TaskStatus.RUNNING) try: # 2. 执行Agent注意Celery任务函数本身不是async我们需要同步运行async函数 # 这里使用asyncio.run来运行异步的execute_task loop asyncio.get_event_loop() except RuntimeError: # 如果在没有事件循环的线程中就新建一个 loop asyncio.new_event_loop() asyncio.set_event_loop(loop) result loop.run_until_complete( agent_executor.execute_task(task_id, user_input) ) loop.close() else: result loop.run_until_complete( agent_executor.execute_task(task_id, user_input) ) # 3. 更新任务状态为完成并保存结果 storage.update_task_status(task_id, TaskStatus.COMPLETED, result{output: result}) return result # 一个用于定时任务的“触发器”任务 celery_app.task(nametrigger_daily_weather_report) def trigger_daily_weather_report(): 定时任务创建每日天气报告任务 # 这里可以读取配置获取需要每日生成报告的城市列表 cities [北京, 上海, 深圳] for city in cities: # 为每个城市创建一个后台任务 task Task( namedaily_weather_report, params{city: city, days: 1} ) task_id storage.create_task(task) # 将任务ID推送到队列等待工作进程处理 # 注意这里我们直接调用run_agent_task.delay将任务发给Celery Worker。 # 但更常见的做法是定时任务只负责创建任务记录由另一个常驻的“任务分发器”来推队列。 # 这里为了简化我们直接推。 run_agent_task.delay(task_id, f生成{city}今天的天气简报。)前端API接口会变得非常轻量# api.py (FastAPI示例) from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel from tasks import run_agent_task, storage from models import Task, TaskStatus app FastAPI() class TaskRequest(BaseModel): user_input: str app.post(/submit_task) async def submit_task(request: TaskRequest, background_tasks: BackgroundTasks): 提交一个异步Agent任务 # 1. 创建任务记录 task Task(nameuser_query, params{user_input: request.user_input}) task_id storage.create_task(task) # 2. 将任务推送到Celery队列异步执行 run_agent_task.delay(task_id, request.user_input) # 3. 立即返回任务ID return {task_id: task_id, status: submitted, message: 任务已提交到后台处理} app.get(/task_status/{task_id}) async def get_task_status(task_id: str): 查询任务状态和结果 task storage.get_task(task_id) if not task: return {error: Task not found} return task.dict()3.5 集成APScheduler实现定时唤醒我们需要在Celery Worker进程中启动一个定时调度器。这可以通过自定义Celery的启动信号worker_ready)来实现。# celery_scheduler.py from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger from celery.signals import worker_ready # 创建一个全局调度器 scheduler BackgroundScheduler() worker_ready.connect def on_worker_ready(sender, **kwargs): 当Celery Worker启动就绪时启动定时调度器 if not scheduler.running: scheduler.start() # 添加定时任务每天上午9点触发 scheduler.add_job( functrigger_daily_weather_report, # 这是上面定义的Celery任务函数 triggerCronTrigger(hour9, minute0), iddaily_weather_job, replace_existingTrue ) print(APScheduler started with daily weather report job.)关键点trigger_daily_weather_report是一个Celery任务函数用celery_app.task装饰。scheduler.add_job只是安排这个函数在特定时间被调用。当它被调用时它会在Celery的上下文中执行从而能够安全地调用run_agent_task.delay()来创建真正的后台任务。这样就实现了“定时唤醒”并创建任务。4. 部署、监控与问题排查实录系统搭建好了但要稳定运行在生产环境还有一大堆“坑”要填。4.1 部署架构与配置要点一个最小化的生产部署可能包含以下服务[Docker Compose 示例] version: 3.8 services: mongodb: image: mongo:latest volumes: - mongodb_data:/data/db redis: image: redis:alpine api: build: ./api ports: - 8000:8000 environment: - MONGO_URImongodb://mongodb:27017 - REDIS_URLredis://redis:6379/0 depends_on: - mongodb - redis celery_worker: build: ./worker command: celery -A tasks.celery_app worker --loglevelinfo --concurrency4 environment: - MONGO_URImongodb://mongodb:27017 - REDIS_URLredis://redis:6379/0 depends_on: - mongodb - redis # 注意需要挂载包含celery_scheduler.py的代码卷确保worker启动时加载调度器配置要点Celery并发数--concurrency参数需要根据你的机器CPU核心数和任务类型I/O密集型还是CPU密集型来调整。对于大量调用LLM API的Agent任务通常是I/O密集型可以设置得高一些如CPU核心数的2-4倍。Redis持久化务必配置Redis的持久化RDB或AOF防止服务器重启导致内存中的任务队列丢失。MongoDB索引在agent_sessions集合的task_id和checkpoint_id字段上创建复合索引加速会话加载。在tasks集合的status和created_at字段上创建索引方便后台管理界面查询。网络超时与重试在Agent执行器内部调用LLM API或外部工具时必须设置合理的超时和重试机制避免一个任务卡死整个工作进程。4.2 常见问题与排查技巧在实际运行中你一定会遇到下面这些问题问题1任务卡在running状态永不结束。可能原因Agent陷入死循环LLM可能在一个思考-行动循环中出不来。外部API调用失败或超时没有正确处理异常。工作进程Celery Worker崩溃但任务状态没被重置。排查与解决设置任务超时在Celery任务装饰器中设置soft_time_limit和time_limit。celery_app.task(bindTrue, soft_time_limit300, time_limit330) def run_agent_task(self, ...): ...实现心跳与看门狗在任务函数中定期更新一个“最后活跃时间戳”到Redis。另一个监控进程检查所有running状态的任务如果其“最后活跃时间戳”超过阈值如10分钟则强制将其状态标记为failed并记录原因。增强Agent的约束在Agent的提示词Prompt中明确限制最大循环次数如“最多进行5次工具调用”并在代码中强制中断。问题2定时任务没有准时执行或者重复执行。可能原因时钟不同步部署多台Worker时服务器时间不一致。APScheduler在多个Worker中重复启动每个Celery Worker进程都启动了自己的调度器导致同一个定时任务被多次触发。排查与解决使用分布式锁在worker_ready信号中在启动调度器之前先尝试获取一个Redis分布式锁。只有一个Worker能成功获取锁并启动调度器。import redis redis_client redis.from_url(os.getenv(REDIS_URL)) lock_key scheduler_lock lock redis_client.lock(lock_key, timeout60) if lock.acquire(blockingFalse): if not scheduler.running: scheduler.start() # ... 添加任务 # 注意不要释放锁让这个锁一直持有直到进程结束。使用外部分布式调度器对于更复杂的生产环境可以考虑使用Celery Beat作为独立的调度服务或者使用Kubernetes CronJob避免在应用内做调度。问题3Agent状态恢复后行为不一致或出错。可能原因序列化/反序列化丢失信息自定义的工具Tool或记忆Memory对象可能包含无法序列化的属性。代码版本不一致保存状态的代码版本和恢复状态的代码版本不同导致数据模型不兼容。排查与解决简化状态对象持久化时只保存最核心、最原始的数据如消息列表的文本内容。恢复时用这些数据重新构建运行时对象。版本化数据模型在AgentSession模型中增加一个schema_version字段。每次数据模型有重大变更时升级版本号并在加载旧版本数据时编写迁移脚本。全面测试建立完整的集成测试覆盖“执行 - 保存 - 重启 - 恢复 - 继续执行”的全流程。问题4数据库连接数暴涨或存储性能瓶颈。可能原因检查点过于频繁每轮对话都保存全量状态在高并发下对数据库压力大。没有使用连接池每次保存/加载都新建数据库连接。排查与解决优化检查点策略改为每N轮对话保存一次或者只在关键步骤如调用重要工具后保存。使用连接池确保MongoDB和Redis的客户端配置了连接池。异步写入如果状态不是需要严格实时持久化可以考虑将保存操作放入一个低优先级的队列异步执行不阻塞主任务流程。4.3 监控与运维建议要让这套系统稳定运行基本的监控必不可少任务大盘做一个简单的管理后台展示任务总数、各状态任务的数量、平均执行时间、失败率等。数据可以直接从MongoDB的tasks集合聚合查询。队列监控监控Redis中任务队列的长度。如果pending_tasks队列持续增长说明Worker处理不过来需要扩容。Agent执行质量监控记录每个任务最终消耗的Token数、工具调用次数、执行轮数。异常高的数值可能意味着Agent陷入了低效循环。日志聚合将所有服务API、Celery Worker的日志集中收集到ELK或Loki等平台方便通过task_id追踪一个任务在所有服务中的完整生命周期日志。最后再分享一个我踩过的大坑LLM上下文长度限制与长程记忆的冲突。我们的持久化方案保存了完整的对话历史但当任务步骤非常多历史消息超过LLM的上下文窗口时直接全部灌给LLM是行不通的。这时就需要引入摘要记忆或向量检索记忆。在保存检查点时不仅保存原始消息还可以用另一个LLM调用对历史对话进行摘要将摘要和最近几条原始消息一起保存。在恢复时优先加载摘要再根据需要从向量库中检索相关的历史片段。这又是另一个复杂但有趣的话题了。
