在实际邮件处理场景中无论是个人邮箱还是客服工单系统都面临一个核心矛盾如何让自动化回复既精准又具备连续性。简单的关键词匹配或固定模板回复无法理解用户的历史诉求和上下文导致每次交互都像是“初次见面”用户体验割裂问题解决效率低下。而具备“记忆”能力的邮件智能体正是为了解决这一痛点而生。它能够记住与特定联系人或特定会话的历史交互内容并基于这些记忆在后续回复中做出更连贯、更个性化的决策。本文将围绕“Lindy 邮件智能体”这一概念深入探讨如何为邮件处理系统构建记忆能力并实现基于记忆的自动回复。我们将从核心概念入手逐步构建一个具备基础记忆与回复功能的原型系统。整个过程将涵盖记忆的存储与检索机制、自动回复的决策逻辑以及如何将大模型、Agent、工具调用等前沿概念融入一个可运行的工程实践中。无论你是希望为现有邮件系统增加智能还是想深入理解AI Agent中“记忆”模块的实现本文都将提供一条清晰的路径。1. 理解邮件智能体的核心记忆与上下文在讨论具体实现之前我们必须先厘清几个关键概念以及它们在邮件处理场景下的具体含义。1.1 什么是邮件智能体的“记忆”在AI Agent的语境中“记忆”并非指生物记忆而是一种数据持久化机制用于存储和检索与特定实体如用户、会话、任务相关的历史信息。对于邮件智能体而言记忆主要分为两类会话记忆针对单次邮件往来线程的记忆。它记录了当前会话中已交换的所有邮件内容、提取的关键信息如订单号、问题描述、解决方案步骤以及智能体已执行的操作如已查询的数据库、已回复的内容。这确保了智能体在回复一封长线程邮件时不会遗忘线程开头的信息。长期记忆针对特定发件人或联系人的记忆。它超越了单次会话存储了与该联系人的所有历史交互中总结出的偏好、历史问题、身份信息等。例如记住某位用户是VIP客户、偏好电话沟通、或曾报告过某个特定设备的故障。没有记忆的智能体每次处理邮件都像一张白纸只能基于当前邮件内容做出反应无法提供连贯的服务。1.2 记忆、RAG与Agent的关系当前热门的几个技术概念在此交汇大模型作为智能体的“大脑”负责理解邮件内容、生成回复文本、进行逻辑推理。它需要高质量的输入上下文。Prompt是引导大模型完成特定任务的指令和上下文模板。一个设计良好的Prompt需要嵌入相关的记忆信息。RAG为智能体提供了从外部知识库如产品手册、FAQ、历史工单中检索相关信息的能力。这可以看作是一种“外部知识记忆”。Agent是具备自主决策能力的程序。它利用大模型进行思考根据记忆和当前状态决定调用哪个工具如查询数据库、发送回复、创建待办事项。记忆是Agent的“个人经历库”为决策提供历史上下文。它与RAG检索的外部知识共同构成Agent的认知基础。工具调用是Agent执行动作的方式例如调用邮件发送API、查询CRM系统。Workflow定义了Agent处理邮件的标准化流程例如“接收邮件 - 解析内容 - 检索记忆 - 决策 - 调用工具 - 更新记忆”。我们的邮件智能体就是一个集成了记忆模块、能够进行工具调用、并按照一定Workflow运行的Agent其核心决策依赖于大模型和精心设计的Prompt。1.3 为什么“新开会话丢失上下文”是个关键问题许多初代的AI编程助手或聊天机器人都有一个通病每次开启新会话之前聊过的内容全部清零。这在邮件场景下是灾难性的。用户可能在第三封邮件中说“还是上次那个问题”如果智能体忘记了“上次那个问题”是什么就无法有效协助。因此我们的系统设计必须优先解决记忆的持久化和关联检索问题。2. 系统设计与环境准备我们将构建一个简化但核心功能完整的邮件智能体原型。它不直接连接真实邮件服务器而是通过模拟的邮件接收接口来演示整个工作流程。2.1 系统架构概览整个系统包含以下核心组件模拟邮件接收器一个HTTP端点接收模拟的邮件数据发件人、主题、正文。记忆存储库使用数据库存储会话记忆和长期记忆。这里为了简化使用SQLite。记忆检索器根据当前邮件信息如发件人、会话ID从记忆存储库中检索相关历史记录。大模型集成层调用大模型API如OpenAI GPT、国内合规大模型API将邮件内容、检索到的记忆组合成Prompt请求生成回复和决策。工作流引擎控制处理流程协调各组件工作。动作执行器执行模型决策出的动作如“发送回复”、“转人工”、“标记待办”。2.2 技术栈与依赖配置我们选择Python作为实现语言因为它有丰富的大模型和AI相关库。项目环境准备创建项目目录并初始化虚拟环境mkdir lindy-mail-agent cd lindy-mail-agent python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate安装核心依赖创建requirements.txt文件内容如下fastapi0.104.1 uvicorn0.24.0 sqlalchemy2.0.23 pydantic2.5.0 openai1.3.0 # 或其他兼容大模型SDK如 dashscope, zhipuai python-dotenv1.0.0执行安装pip install -r requirements.txt准备大模型API密钥创建.env文件存放你的API密钥请使用合规的大模型服务# .env 文件示例 OPENAI_API_KEYyour_openai_api_key_here # 或者使用国内模型 # DASHSCOPE_API_KEYyour_dashscope_api_key_here MODEL_NAMEgpt-3.5-turbo # 或 qwen-max, glm-4等2.3 项目结构设计一个清晰的项目结构有助于维护和扩展。lindy-mail-agent/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用入口 │ ├── models.py # SQLAlchemy 数据模型 Pydantic 请求/响应模型 │ ├── memory.py # 记忆存储与检索核心逻辑 │ ├── agent.py # 智能体决策与工作流逻辑 │ ├── prompts.py # 存放各类Prompt模板 │ └── config.py # 配置文件 ├── requirements.txt ├── .env └── README.md3. 构建记忆存储与检索模块记忆模块是智能体的基石。我们设计一个简单的数据库模型来存储记忆。3.1 定义数据模型在app/models.py中我们定义两个核心表ConversationMemory会话记忆和LongTermMemory长期记忆。# app/models.py from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship, sessionmaker from datetime import datetime import os from dotenv import load_dotenv load_dotenv() Base declarative_base() class ConversationMemory(Base): 存储单次邮件会话的记忆 __tablename__ conversation_memories id Column(Integer, primary_keyTrue) # 会话唯一标识可以用邮件线程ID或自定义UUID session_id Column(String(255), nullableFalse, indexTrue) # 发件人邮箱用于关联长期记忆 sender_email Column(String(255), nullableFalse, indexTrue) # 记忆内容可以是原始邮件片段或提取的摘要 content Column(Text, nullableFalse) # 记忆类型如 user_message, agent_response, extracted_info memory_type Column(String(50)) # 时间戳 created_at Column(DateTime, defaultdatetime.utcnow) def __repr__(self): return fConversationMemory(session_id{self.session_id}, type{self.memory_type}) class LongTermMemory(Base): 存储与特定发件人相关的长期记忆 __tablename__ long_term_memories id Column(Integer, primary_keyTrue) # 关联的发件人邮箱 sender_email Column(String(255), nullableFalse, indexTrue, uniqueFalse) # 记忆键用于分类如 preference, past_issue, customer_tier memory_key Column(String(100), nullableFalse) # 记忆值 memory_value Column(Text, nullableFalse) # 更新时间 updated_at Column(DateTime, defaultdatetime.utcnow, onupdatedatetime.utcnow) # 复合索引确保同一用户的同类型记忆只有一条或可多条 __table_args__ (Index(idx_email_key, sender_email, memory_key),) # 数据库连接使用SQLite作为示例 DATABASE_URL os.getenv(DATABASE_URL, sqlite:///./mail_agent.db) engine create_engine(DATABASE_URL, connect_args{check_same_thread: False} if DATABASE_URL.startswith(sqlite) else {}) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) # 创建表 Base.metadata.create_all(bindengine)3.2 实现记忆的存储与检索类在app/memory.py中我们创建MemoryManager类来封装所有记忆操作。# app/memory.py from sqlalchemy.orm import Session from .models import SessionLocal, ConversationMemory, LongTermMemory from typing import List, Optional, Dict, Any import json class MemoryManager: def __init__(self): self.db: Session SessionLocal() def add_conversation_memory(self, session_id: str, sender_email: str, content: str, memory_type: str user_message): 添加一条会话记忆 memory ConversationMemory( session_idsession_id, sender_emailsender_email, contentcontent, memory_typememory_type ) self.db.add(memory) self.db.commit() return memory def get_conversation_history(self, session_id: str, limit: int 10) - List[Dict[str, Any]]: 获取指定会话的历史记录按时间倒序 memories self.db.query(ConversationMemory).filter( ConversationMemory.session_id session_id ).order_by(ConversationMemory.created_at.desc()).limit(limit).all() # 返回格式化的历史便于放入Prompt history [] for mem in reversed(memories): # 反转回时间顺序 role user if mem.memory_type user_message else assistant history.append({role: role, content: mem.content}) return history def upsert_long_term_memory(self, sender_email: str, key: str, value: str): 更新或插入一条长期记忆 existing self.db.query(LongTermMemory).filter( LongTermMemory.sender_email sender_email, LongTermMemory.memory_key key ).first() if existing: existing.memory_value value else: memory LongTermMemory(sender_emailsender_email, memory_keykey, memory_valuevalue) self.db.add(memory) self.db.commit() def get_long_term_memories(self, sender_email: str) - Dict[str, str]: 获取指定发件人的所有长期记忆 memories self.db.query(LongTermMemory).filter( LongTermMemory.sender_email sender_email ).all() return {mem.memory_key: mem.memory_value for mem in memories} def close(self): self.db.close()关键解释add_conversation_memory: 每次处理邮件无论是用户来信还是智能体回复都调用此方法存储形成会话流水账。get_conversation_history: 在生成回复前调用获取最近的对话历史作为上下文提供给大模型。这里限制limit是为了防止上下文过长。upsert_long_term_memory: 当从对话中提取出重要用户特征如“偏好英文回复”时调用此方法更新长期记忆。使用SQLite是为了演示简便生产环境应更换为PostgreSQL或MySQL并考虑记忆内容的向量化存储和语义检索以支持更复杂的“相关记忆”查找而非仅按会话ID或键查询。4. 实现智能体工作流与自动回复智能体是系统的指挥中心。它按既定工作流运行接收邮件 - 检索记忆 - 决策 - 执行 - 更新记忆。4.1 设计Prompt模板Prompt是与大模型沟通的桥梁。在app/prompts.py中定义。# app/prompts.py from typing import Dict, List def build_agent_prompt( current_email: Dict, # 包含 subject, body, sender conversation_history: List[Dict], long_term_memories: Dict[str, str] ) - str: 构建驱动智能体决策的Prompt。 返回一个结构化的字符串包含系统指令、记忆上下文和当前请求。 # 格式化长期记忆 lt_memory_str if long_term_memories: lt_memory_str \n关于这位用户的长期记忆\n for k, v in long_term_memories.items(): lt_memory_str f- {k}: {v}\n # 格式化对话历史 history_str if conversation_history: history_str \n本次会话历史最近几次交流\n for msg in conversation_history: role 用户 if msg[role] user else 助理 history_str f{role}: {msg[content]}\n prompt f 你是一个专业的邮件助理智能体Lindy。你的任务是分析收到的邮件结合记忆决定如何回复或处理。 {lt_memory_str} {history_str} 当前收到一封新邮件 发件人{current_email[sender]} 主题{current_email[subject]} 正文 {current_email[body]} 请根据以上信息按以下JSON格式输出你的决策 {{ thought_process: 你的思考过程分析用户意图、相关记忆等, action: 需执行的动作可选值reply_directly直接回复, need_more_info需要更多信息, escalate_to_human转人工, create_todo创建待办, reply_content: 如果action是reply_directly这里填写回复的邮件正文用礼貌、专业的口吻。否则留空或写N/A。, extracted_info_for_memory: {{ key1: 从本次交互中提取的值得长期记忆的信息如用户偏好、问题类型等, key2: ... }} }} 请确保你的回复是纯JSON格式便于程序解析。 return prompt4.2 构建智能体核心逻辑在app/agent.py中我们创建MailAgent类。# app/agent.py from .memory import MemoryManager from .prompts import build_agent_prompt import openai # 或 from openai import OpenAI import json import os from dotenv import load_dotenv from typing import Dict, Any load_dotenv() class MailAgent: def __init__(self): self.memory_manager MemoryManager() # 初始化大模型客户端这里以OpenAI格式为例 self.client openai.OpenAI(api_keyos.getenv(OPENAI_API_KEY)) self.model os.getenv(MODEL_NAME, gpt-3.5-turbo) def process_email(self, session_id: str, sender: str, subject: str, body: str) - Dict[str, Any]: 处理一封邮件的核心工作流。 返回处理结果。 # 1. 存储当前用户邮件到会话记忆 self.memory_manager.add_conversation_memory(session_id, sender, body, user_message) # 2. 检索记忆 conv_history self.memory_manager.get_conversation_history(session_id, limit5) long_term_mem self.memory_manager.get_long_term_memories(sender) # 3. 构建Prompt并调用大模型 current_email {sender: sender, subject: subject, body: body} prompt build_agent_prompt(current_email, conv_history, long_term_mem) try: response self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], temperature0.2, # 低温度使输出更稳定 response_format{type: json_object} # 要求返回JSON ) decision_str response.choices[0].message.content decision json.loads(decision_str) except Exception as e: # 模型调用失败降级处理 decision { thought_process: f模型调用失败: {e}, action: escalate_to_human, reply_content: N/A, extracted_info_for_memory: {} } # 4. 执行动作这里主要演示回复 result {decision: decision} if decision.get(action) reply_directly and decision.get(reply_content): reply_content decision[reply_content] # 模拟发送回复实际应调用邮件发送API print(f[模拟] 发送回复给 {sender}: {reply_content[:100]}...) # 将助理回复也存入会话记忆 self.memory_manager.add_conversation_memory(session_id, sender, reply_content, agent_response) result[reply_sent] True result[reply_preview] reply_content[:200] else: result[reply_sent] False result[action_needed] decision.get(action) # 5. 更新长期记忆 extracted_info decision.get(extracted_info_for_memory, {}) if extracted_info and isinstance(extracted_info, dict): for key, value in extracted_info.items(): if value and value ! N/A: self.memory_manager.upsert_long_term_memory(sender, key, value) # 6. 返回处理结果 return result def close(self): self.memory_manager.close()工作流详解记忆存储收到邮件先存入会话记忆表。记忆检索获取该会话的历史和该发件人的长期记忆。模型决策将记忆和当前邮件组合成Prompt请求大模型分析并输出结构化决策JSON。动作执行根据决策的action字段执行相应操作。本例主要实现reply_directly。记忆更新将模型提取的extracted_info_for_memory更新到长期记忆表中。资源清理处理完成后关闭数据库连接。5. 创建API接口并运行验证我们将使用FastAPI创建一个简单的HTTP接口来模拟接收邮件并触发智能体处理。5.1 定义API端点在app/main.py中创建FastAPI应用。# app/main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from .agent import MailAgent import uuid app FastAPI(titleLindy Mail Agent API) # 请求数据模型 class EmailRequest(BaseModel): sender: str subject: str body: str # 可选如果不提供则自动生成一个会话ID模拟新会话 session_id: str None # 响应数据模型 class EmailResponse(BaseModel): session_id: str processed: bool decision: dict reply_sent: bool reply_preview: str None action_needed: str None app.post(/process_email/, response_modelEmailResponse) async def process_email(request: EmailRequest): 接收一封邮件交由智能体处理 agent MailAgent() try: # 如果未提供session_id则生成一个模拟新邮件线程 session_id request.session_id if request.session_id else fsess_{uuid.uuid4().hex[:8]} result agent.process_email( session_idsession_id, senderrequest.sender, subjectrequest.subject, bodyrequest.body ) response_data { session_id: session_id, processed: True, decision: result.get(decision, {}), reply_sent: result.get(reply_sent, False), reply_preview: result.get(reply_preview), action_needed: result.get(action_needed) } return EmailResponse(**response_data) except Exception as e: raise HTTPException(status_code500, detailf处理邮件时出错: {str(e)}) finally: agent.close() app.get(/) async def root(): return {message: Lindy Mail Agent is running.}5.2 运行与测试启动服务uvicorn app.main:app --reload --host 0.0.0.0 --port 8000模拟发送邮件进行测试使用curl或 Postman 等工具调用API。测试用例1新用户首次咨询curl -X POST http://localhost:8000/process_email/ \ -H Content-Type: application/json \ -d { sender: customer_aexample.com, subject: 产品价格咨询, body: 你好我想了解一下你们旗舰版产品的价格和功能区别。 }预期结果智能体没有该用户的长期记忆但会根据邮件内容生成一个专业的回复并可能将customer_aexample.com的inquiry_type提取为price_and_feature存入长期记忆。测试用例2同一用户再次询问测试记忆curl -X POST http://localhost:8000/process_email/ \ -H Content-Type: application/json \ -d { sender: customer_aexample.com, subject: Re: 产品价格咨询, body: 谢谢回复关于旗舰版它的存储空间具体是多少, session_id: sess_abc123 # 使用第一次回复的session_id模拟同一会话 }预期结果智能体会检索到会话历史第一次的问答和长期记忆知道用户咨询过价格功能。它的回复会更具连续性例如“正如我上次提到的旗舰版包含...关于存储空间具体是...”检查数据库使用SQLite浏览器或命令行查看mail_agent.db文件确认conversation_memories和long_term_memories表中已正确存入数据。6. 常见问题排查与优化在实际部署和运行中你可能会遇到以下问题。6.1 模型调用失败或返回非JSON现象API返回500错误日志显示json.decoder.JSONDecodeError。可能原因大模型没有严格遵守response_format参数返回了非JSON内容。API密钥错误、网络问题或模型服务不可用。排查与解决检查API配置确认.env文件中的OPENAI_API_KEY和MODEL_NAME正确且账户有余额。增强JSON解析健壮性在agent.py的process_email方法中用try-except包裹json.loads并在解析失败时提供降级策略。try: decision json.loads(decision_str) except json.JSONDecodeError: # 尝试提取可能的JSON部分或直接降级 print(f模型返回非标准JSON: {decision_str[:200]}) decision { thought_process: 模型返回格式异常转为人工处理。, action: escalate_to_human, reply_content: N/A, extracted_info_for_memory: {} }调整Prompt在Prompt中更强烈地要求返回JSON并给出更严格的格式示例。6.2 记忆检索不相关或效率低现象随着数据量增长按session_id和sender_email的简单检索可能无法找到最相关的历史信息或者检索速度变慢。解决方案引入向量化记忆将会话记忆的内容通过嵌入模型转换为向量存储在向量数据库如Chroma, Weaviate, Pinecone中。检索时使用当前邮件内容的向量进行语义相似度搜索找到最相关的历史片段而非仅仅按会话ID。记忆摘要对于长会话不要存储所有原始消息。可以定期或在会话结束时让大模型生成一个“会话摘要”只存储摘要大幅减少存储和检索的负担。数据库索引优化确保session_id和sender_email字段已建立索引。6.3 上下文长度限制与Token超限现象当会话历史很长时拼接的Prompt可能超过大模型的上下文窗口限制导致调用失败。解决方案限制历史条数如代码中的limit5只取最近N条消息。动态摘要在Prompt中不直接放入全部历史而是先让模型根据当前问题从历史中提取最关键的信息形成一个简短的“上下文摘要”再将摘要放入主Prompt。使用支持长上下文模型选择上下文窗口更大的模型。6.4 长期记忆的更新策略冲突现象模型提取的extracted_info_for_memory可能不准确或相互矛盾导致长期记忆被错误覆盖。解决方案置信度过滤让模型在提取信息时附带一个置信度分数只存储高置信度的信息。人工审核对于关键信息如客户等级、合同金额的更新可以先存入“待审核”表由人工确认后再更新到正式长期记忆。版本化或追加对于某些记忆如“历史问题”可以采用追加模式而非覆盖记录该用户所有报告过的问题列表。7. 生产环境最佳实践与扩展方向将原型发展为生产可用的系统需要考虑更多因素。7.1 生产环境检查清单维度学习/开发环境生产环境建议数据库SQLitePostgreSQL/MySQL考虑读写分离、备份策略。记忆向量化推荐专用向量库。大模型API直接调用配置重试、熔断、降级、多个API Key轮询。设置合理的超时和限流。安全性本地运行API接口需增加认证API Key/JWT。用户邮箱等PII信息需脱敏或加密存储。可观测性打印日志集成结构化日志如JSON Log记录每次处理的session_id、模型调用耗时、决策结果。接入监控和告警。性能顺序处理引入消息队列如RabbitMQ, Redis Queue异步处理邮件避免HTTP请求阻塞。记忆管理永不过期设计记忆过期和归档策略。定期清理过于陈旧的会话记忆或将长期记忆迁移到冷存储。7.2 扩展功能方向集成真实邮件服务使用imaplib/smtplib或第三方库如exchangelibfor Outlook, Gmail API替换模拟接口实现真正的邮箱监听和发送。工具调用增强让Agent能执行更丰富的动作如query_knowledge_base: 从公司内部知识库通过RAG检索答案。check_order_status: 调用内部订单系统API查询状态。schedule_meeting: 调用日历API安排会议。 这需要扩展Prompt中的动作定义并实现相应的工具函数。工作流引擎复杂化使用LangGraph或类似框架来定义更复杂、带状态循环的决策流程。例如当动作为need_more_info时自动进入一个“追问”子流程。多模态处理支持处理邮件中的图片、附件提取其中的文字信息使用OCR或分析附件内容。评估与持续学习建立反馈机制收集人工对自动回复的评分好/差用于微调模型或优化Prompt。7.3 关键配置参数说明在app/config.py中集中管理配置以下是一些关键参数# app/config.py import os from dotenv import load_dotenv load_dotenv() class Config: # 大模型相关 LLM_API_KEY os.getenv(LLM_API_KEY) LLM_BASE_URL os.getenv(LLM_BASE_URL, https://api.openai.com/v1) # 适配国内模型 LLM_MODEL os.getenv(LLM_MODEL, gpt-3.5-turbo) LLM_TEMPERATURE float(os.getenv(LLM_TEMPERATURE, 0.2)) LLM_MAX_TOKENS int(os.getenv(LLM_MAX_TOKENS, 2000)) # 记忆相关 CONVERSATION_HISTORY_LIMIT int(os.getenv(CONVERSATION_HISTORY_LIMIT, 5)) # 是否启用长期记忆 ENABLE_LONG_TERM_MEMORY os.getenv(ENABLE_LONG_TERM_MEMORY, true).lower() true # 数据库 DATABASE_URL os.getenv(DATABASE_URL, sqlite:///./mail_agent.db) # 代理与超时 HTTP_PROXY os.getenv(HTTP_PROXY) REQUEST_TIMEOUT int(os.getenv(REQUEST_TIMEOUT, 30))构建一个具备记忆能力的邮件智能体核心在于将离散的邮件交互转化为连续的、有状态的对话。通过本文的实践你不仅实现了一个原型更重要的是理解了记忆存储、检索、更新与大模型协同工作的完整链路。在实际项目中应从简单的规则和关键词匹配入手逐步引入语义记忆和复杂决策并始终将系统的可靠性、可解释性和安全性置于首位。下一步你可以尝试为你的智能体添加第一个真正的工具调用例如连接一个FAQ数据库让它从“能记忆”进化到“能查资料、能行动”。 构建具备记忆能力的邮件智能体：从RAG到Agent的工程实践
在实际邮件处理场景中无论是个人邮箱还是客服工单系统都面临一个核心矛盾如何让自动化回复既精准又具备连续性。简单的关键词匹配或固定模板回复无法理解用户的历史诉求和上下文导致每次交互都像是“初次见面”用户体验割裂问题解决效率低下。而具备“记忆”能力的邮件智能体正是为了解决这一痛点而生。它能够记住与特定联系人或特定会话的历史交互内容并基于这些记忆在后续回复中做出更连贯、更个性化的决策。本文将围绕“Lindy 邮件智能体”这一概念深入探讨如何为邮件处理系统构建记忆能力并实现基于记忆的自动回复。我们将从核心概念入手逐步构建一个具备基础记忆与回复功能的原型系统。整个过程将涵盖记忆的存储与检索机制、自动回复的决策逻辑以及如何将大模型、Agent、工具调用等前沿概念融入一个可运行的工程实践中。无论你是希望为现有邮件系统增加智能还是想深入理解AI Agent中“记忆”模块的实现本文都将提供一条清晰的路径。1. 理解邮件智能体的核心记忆与上下文在讨论具体实现之前我们必须先厘清几个关键概念以及它们在邮件处理场景下的具体含义。1.1 什么是邮件智能体的“记忆”在AI Agent的语境中“记忆”并非指生物记忆而是一种数据持久化机制用于存储和检索与特定实体如用户、会话、任务相关的历史信息。对于邮件智能体而言记忆主要分为两类会话记忆针对单次邮件往来线程的记忆。它记录了当前会话中已交换的所有邮件内容、提取的关键信息如订单号、问题描述、解决方案步骤以及智能体已执行的操作如已查询的数据库、已回复的内容。这确保了智能体在回复一封长线程邮件时不会遗忘线程开头的信息。长期记忆针对特定发件人或联系人的记忆。它超越了单次会话存储了与该联系人的所有历史交互中总结出的偏好、历史问题、身份信息等。例如记住某位用户是VIP客户、偏好电话沟通、或曾报告过某个特定设备的故障。没有记忆的智能体每次处理邮件都像一张白纸只能基于当前邮件内容做出反应无法提供连贯的服务。1.2 记忆、RAG与Agent的关系当前热门的几个技术概念在此交汇大模型作为智能体的“大脑”负责理解邮件内容、生成回复文本、进行逻辑推理。它需要高质量的输入上下文。Prompt是引导大模型完成特定任务的指令和上下文模板。一个设计良好的Prompt需要嵌入相关的记忆信息。RAG为智能体提供了从外部知识库如产品手册、FAQ、历史工单中检索相关信息的能力。这可以看作是一种“外部知识记忆”。Agent是具备自主决策能力的程序。它利用大模型进行思考根据记忆和当前状态决定调用哪个工具如查询数据库、发送回复、创建待办事项。记忆是Agent的“个人经历库”为决策提供历史上下文。它与RAG检索的外部知识共同构成Agent的认知基础。工具调用是Agent执行动作的方式例如调用邮件发送API、查询CRM系统。Workflow定义了Agent处理邮件的标准化流程例如“接收邮件 - 解析内容 - 检索记忆 - 决策 - 调用工具 - 更新记忆”。我们的邮件智能体就是一个集成了记忆模块、能够进行工具调用、并按照一定Workflow运行的Agent其核心决策依赖于大模型和精心设计的Prompt。1.3 为什么“新开会话丢失上下文”是个关键问题许多初代的AI编程助手或聊天机器人都有一个通病每次开启新会话之前聊过的内容全部清零。这在邮件场景下是灾难性的。用户可能在第三封邮件中说“还是上次那个问题”如果智能体忘记了“上次那个问题”是什么就无法有效协助。因此我们的系统设计必须优先解决记忆的持久化和关联检索问题。2. 系统设计与环境准备我们将构建一个简化但核心功能完整的邮件智能体原型。它不直接连接真实邮件服务器而是通过模拟的邮件接收接口来演示整个工作流程。2.1 系统架构概览整个系统包含以下核心组件模拟邮件接收器一个HTTP端点接收模拟的邮件数据发件人、主题、正文。记忆存储库使用数据库存储会话记忆和长期记忆。这里为了简化使用SQLite。记忆检索器根据当前邮件信息如发件人、会话ID从记忆存储库中检索相关历史记录。大模型集成层调用大模型API如OpenAI GPT、国内合规大模型API将邮件内容、检索到的记忆组合成Prompt请求生成回复和决策。工作流引擎控制处理流程协调各组件工作。动作执行器执行模型决策出的动作如“发送回复”、“转人工”、“标记待办”。2.2 技术栈与依赖配置我们选择Python作为实现语言因为它有丰富的大模型和AI相关库。项目环境准备创建项目目录并初始化虚拟环境mkdir lindy-mail-agent cd lindy-mail-agent python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate安装核心依赖创建requirements.txt文件内容如下fastapi0.104.1 uvicorn0.24.0 sqlalchemy2.0.23 pydantic2.5.0 openai1.3.0 # 或其他兼容大模型SDK如 dashscope, zhipuai python-dotenv1.0.0执行安装pip install -r requirements.txt准备大模型API密钥创建.env文件存放你的API密钥请使用合规的大模型服务# .env 文件示例 OPENAI_API_KEYyour_openai_api_key_here # 或者使用国内模型 # DASHSCOPE_API_KEYyour_dashscope_api_key_here MODEL_NAMEgpt-3.5-turbo # 或 qwen-max, glm-4等2.3 项目结构设计一个清晰的项目结构有助于维护和扩展。lindy-mail-agent/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用入口 │ ├── models.py # SQLAlchemy 数据模型 Pydantic 请求/响应模型 │ ├── memory.py # 记忆存储与检索核心逻辑 │ ├── agent.py # 智能体决策与工作流逻辑 │ ├── prompts.py # 存放各类Prompt模板 │ └── config.py # 配置文件 ├── requirements.txt ├── .env └── README.md3. 构建记忆存储与检索模块记忆模块是智能体的基石。我们设计一个简单的数据库模型来存储记忆。3.1 定义数据模型在app/models.py中我们定义两个核心表ConversationMemory会话记忆和LongTermMemory长期记忆。# app/models.py from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship, sessionmaker from datetime import datetime import os from dotenv import load_dotenv load_dotenv() Base declarative_base() class ConversationMemory(Base): 存储单次邮件会话的记忆 __tablename__ conversation_memories id Column(Integer, primary_keyTrue) # 会话唯一标识可以用邮件线程ID或自定义UUID session_id Column(String(255), nullableFalse, indexTrue) # 发件人邮箱用于关联长期记忆 sender_email Column(String(255), nullableFalse, indexTrue) # 记忆内容可以是原始邮件片段或提取的摘要 content Column(Text, nullableFalse) # 记忆类型如 user_message, agent_response, extracted_info memory_type Column(String(50)) # 时间戳 created_at Column(DateTime, defaultdatetime.utcnow) def __repr__(self): return fConversationMemory(session_id{self.session_id}, type{self.memory_type}) class LongTermMemory(Base): 存储与特定发件人相关的长期记忆 __tablename__ long_term_memories id Column(Integer, primary_keyTrue) # 关联的发件人邮箱 sender_email Column(String(255), nullableFalse, indexTrue, uniqueFalse) # 记忆键用于分类如 preference, past_issue, customer_tier memory_key Column(String(100), nullableFalse) # 记忆值 memory_value Column(Text, nullableFalse) # 更新时间 updated_at Column(DateTime, defaultdatetime.utcnow, onupdatedatetime.utcnow) # 复合索引确保同一用户的同类型记忆只有一条或可多条 __table_args__ (Index(idx_email_key, sender_email, memory_key),) # 数据库连接使用SQLite作为示例 DATABASE_URL os.getenv(DATABASE_URL, sqlite:///./mail_agent.db) engine create_engine(DATABASE_URL, connect_args{check_same_thread: False} if DATABASE_URL.startswith(sqlite) else {}) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) # 创建表 Base.metadata.create_all(bindengine)3.2 实现记忆的存储与检索类在app/memory.py中我们创建MemoryManager类来封装所有记忆操作。# app/memory.py from sqlalchemy.orm import Session from .models import SessionLocal, ConversationMemory, LongTermMemory from typing import List, Optional, Dict, Any import json class MemoryManager: def __init__(self): self.db: Session SessionLocal() def add_conversation_memory(self, session_id: str, sender_email: str, content: str, memory_type: str user_message): 添加一条会话记忆 memory ConversationMemory( session_idsession_id, sender_emailsender_email, contentcontent, memory_typememory_type ) self.db.add(memory) self.db.commit() return memory def get_conversation_history(self, session_id: str, limit: int 10) - List[Dict[str, Any]]: 获取指定会话的历史记录按时间倒序 memories self.db.query(ConversationMemory).filter( ConversationMemory.session_id session_id ).order_by(ConversationMemory.created_at.desc()).limit(limit).all() # 返回格式化的历史便于放入Prompt history [] for mem in reversed(memories): # 反转回时间顺序 role user if mem.memory_type user_message else assistant history.append({role: role, content: mem.content}) return history def upsert_long_term_memory(self, sender_email: str, key: str, value: str): 更新或插入一条长期记忆 existing self.db.query(LongTermMemory).filter( LongTermMemory.sender_email sender_email, LongTermMemory.memory_key key ).first() if existing: existing.memory_value value else: memory LongTermMemory(sender_emailsender_email, memory_keykey, memory_valuevalue) self.db.add(memory) self.db.commit() def get_long_term_memories(self, sender_email: str) - Dict[str, str]: 获取指定发件人的所有长期记忆 memories self.db.query(LongTermMemory).filter( LongTermMemory.sender_email sender_email ).all() return {mem.memory_key: mem.memory_value for mem in memories} def close(self): self.db.close()关键解释add_conversation_memory: 每次处理邮件无论是用户来信还是智能体回复都调用此方法存储形成会话流水账。get_conversation_history: 在生成回复前调用获取最近的对话历史作为上下文提供给大模型。这里限制limit是为了防止上下文过长。upsert_long_term_memory: 当从对话中提取出重要用户特征如“偏好英文回复”时调用此方法更新长期记忆。使用SQLite是为了演示简便生产环境应更换为PostgreSQL或MySQL并考虑记忆内容的向量化存储和语义检索以支持更复杂的“相关记忆”查找而非仅按会话ID或键查询。4. 实现智能体工作流与自动回复智能体是系统的指挥中心。它按既定工作流运行接收邮件 - 检索记忆 - 决策 - 执行 - 更新记忆。4.1 设计Prompt模板Prompt是与大模型沟通的桥梁。在app/prompts.py中定义。# app/prompts.py from typing import Dict, List def build_agent_prompt( current_email: Dict, # 包含 subject, body, sender conversation_history: List[Dict], long_term_memories: Dict[str, str] ) - str: 构建驱动智能体决策的Prompt。 返回一个结构化的字符串包含系统指令、记忆上下文和当前请求。 # 格式化长期记忆 lt_memory_str if long_term_memories: lt_memory_str \n关于这位用户的长期记忆\n for k, v in long_term_memories.items(): lt_memory_str f- {k}: {v}\n # 格式化对话历史 history_str if conversation_history: history_str \n本次会话历史最近几次交流\n for msg in conversation_history: role 用户 if msg[role] user else 助理 history_str f{role}: {msg[content]}\n prompt f 你是一个专业的邮件助理智能体Lindy。你的任务是分析收到的邮件结合记忆决定如何回复或处理。 {lt_memory_str} {history_str} 当前收到一封新邮件 发件人{current_email[sender]} 主题{current_email[subject]} 正文 {current_email[body]} 请根据以上信息按以下JSON格式输出你的决策 {{ thought_process: 你的思考过程分析用户意图、相关记忆等, action: 需执行的动作可选值reply_directly直接回复, need_more_info需要更多信息, escalate_to_human转人工, create_todo创建待办, reply_content: 如果action是reply_directly这里填写回复的邮件正文用礼貌、专业的口吻。否则留空或写N/A。, extracted_info_for_memory: {{ key1: 从本次交互中提取的值得长期记忆的信息如用户偏好、问题类型等, key2: ... }} }} 请确保你的回复是纯JSON格式便于程序解析。 return prompt4.2 构建智能体核心逻辑在app/agent.py中我们创建MailAgent类。# app/agent.py from .memory import MemoryManager from .prompts import build_agent_prompt import openai # 或 from openai import OpenAI import json import os from dotenv import load_dotenv from typing import Dict, Any load_dotenv() class MailAgent: def __init__(self): self.memory_manager MemoryManager() # 初始化大模型客户端这里以OpenAI格式为例 self.client openai.OpenAI(api_keyos.getenv(OPENAI_API_KEY)) self.model os.getenv(MODEL_NAME, gpt-3.5-turbo) def process_email(self, session_id: str, sender: str, subject: str, body: str) - Dict[str, Any]: 处理一封邮件的核心工作流。 返回处理结果。 # 1. 存储当前用户邮件到会话记忆 self.memory_manager.add_conversation_memory(session_id, sender, body, user_message) # 2. 检索记忆 conv_history self.memory_manager.get_conversation_history(session_id, limit5) long_term_mem self.memory_manager.get_long_term_memories(sender) # 3. 构建Prompt并调用大模型 current_email {sender: sender, subject: subject, body: body} prompt build_agent_prompt(current_email, conv_history, long_term_mem) try: response self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], temperature0.2, # 低温度使输出更稳定 response_format{type: json_object} # 要求返回JSON ) decision_str response.choices[0].message.content decision json.loads(decision_str) except Exception as e: # 模型调用失败降级处理 decision { thought_process: f模型调用失败: {e}, action: escalate_to_human, reply_content: N/A, extracted_info_for_memory: {} } # 4. 执行动作这里主要演示回复 result {decision: decision} if decision.get(action) reply_directly and decision.get(reply_content): reply_content decision[reply_content] # 模拟发送回复实际应调用邮件发送API print(f[模拟] 发送回复给 {sender}: {reply_content[:100]}...) # 将助理回复也存入会话记忆 self.memory_manager.add_conversation_memory(session_id, sender, reply_content, agent_response) result[reply_sent] True result[reply_preview] reply_content[:200] else: result[reply_sent] False result[action_needed] decision.get(action) # 5. 更新长期记忆 extracted_info decision.get(extracted_info_for_memory, {}) if extracted_info and isinstance(extracted_info, dict): for key, value in extracted_info.items(): if value and value ! N/A: self.memory_manager.upsert_long_term_memory(sender, key, value) # 6. 返回处理结果 return result def close(self): self.memory_manager.close()工作流详解记忆存储收到邮件先存入会话记忆表。记忆检索获取该会话的历史和该发件人的长期记忆。模型决策将记忆和当前邮件组合成Prompt请求大模型分析并输出结构化决策JSON。动作执行根据决策的action字段执行相应操作。本例主要实现reply_directly。记忆更新将模型提取的extracted_info_for_memory更新到长期记忆表中。资源清理处理完成后关闭数据库连接。5. 创建API接口并运行验证我们将使用FastAPI创建一个简单的HTTP接口来模拟接收邮件并触发智能体处理。5.1 定义API端点在app/main.py中创建FastAPI应用。# app/main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from .agent import MailAgent import uuid app FastAPI(titleLindy Mail Agent API) # 请求数据模型 class EmailRequest(BaseModel): sender: str subject: str body: str # 可选如果不提供则自动生成一个会话ID模拟新会话 session_id: str None # 响应数据模型 class EmailResponse(BaseModel): session_id: str processed: bool decision: dict reply_sent: bool reply_preview: str None action_needed: str None app.post(/process_email/, response_modelEmailResponse) async def process_email(request: EmailRequest): 接收一封邮件交由智能体处理 agent MailAgent() try: # 如果未提供session_id则生成一个模拟新邮件线程 session_id request.session_id if request.session_id else fsess_{uuid.uuid4().hex[:8]} result agent.process_email( session_idsession_id, senderrequest.sender, subjectrequest.subject, bodyrequest.body ) response_data { session_id: session_id, processed: True, decision: result.get(decision, {}), reply_sent: result.get(reply_sent, False), reply_preview: result.get(reply_preview), action_needed: result.get(action_needed) } return EmailResponse(**response_data) except Exception as e: raise HTTPException(status_code500, detailf处理邮件时出错: {str(e)}) finally: agent.close() app.get(/) async def root(): return {message: Lindy Mail Agent is running.}5.2 运行与测试启动服务uvicorn app.main:app --reload --host 0.0.0.0 --port 8000模拟发送邮件进行测试使用curl或 Postman 等工具调用API。测试用例1新用户首次咨询curl -X POST http://localhost:8000/process_email/ \ -H Content-Type: application/json \ -d { sender: customer_aexample.com, subject: 产品价格咨询, body: 你好我想了解一下你们旗舰版产品的价格和功能区别。 }预期结果智能体没有该用户的长期记忆但会根据邮件内容生成一个专业的回复并可能将customer_aexample.com的inquiry_type提取为price_and_feature存入长期记忆。测试用例2同一用户再次询问测试记忆curl -X POST http://localhost:8000/process_email/ \ -H Content-Type: application/json \ -d { sender: customer_aexample.com, subject: Re: 产品价格咨询, body: 谢谢回复关于旗舰版它的存储空间具体是多少, session_id: sess_abc123 # 使用第一次回复的session_id模拟同一会话 }预期结果智能体会检索到会话历史第一次的问答和长期记忆知道用户咨询过价格功能。它的回复会更具连续性例如“正如我上次提到的旗舰版包含...关于存储空间具体是...”检查数据库使用SQLite浏览器或命令行查看mail_agent.db文件确认conversation_memories和long_term_memories表中已正确存入数据。6. 常见问题排查与优化在实际部署和运行中你可能会遇到以下问题。6.1 模型调用失败或返回非JSON现象API返回500错误日志显示json.decoder.JSONDecodeError。可能原因大模型没有严格遵守response_format参数返回了非JSON内容。API密钥错误、网络问题或模型服务不可用。排查与解决检查API配置确认.env文件中的OPENAI_API_KEY和MODEL_NAME正确且账户有余额。增强JSON解析健壮性在agent.py的process_email方法中用try-except包裹json.loads并在解析失败时提供降级策略。try: decision json.loads(decision_str) except json.JSONDecodeError: # 尝试提取可能的JSON部分或直接降级 print(f模型返回非标准JSON: {decision_str[:200]}) decision { thought_process: 模型返回格式异常转为人工处理。, action: escalate_to_human, reply_content: N/A, extracted_info_for_memory: {} }调整Prompt在Prompt中更强烈地要求返回JSON并给出更严格的格式示例。6.2 记忆检索不相关或效率低现象随着数据量增长按session_id和sender_email的简单检索可能无法找到最相关的历史信息或者检索速度变慢。解决方案引入向量化记忆将会话记忆的内容通过嵌入模型转换为向量存储在向量数据库如Chroma, Weaviate, Pinecone中。检索时使用当前邮件内容的向量进行语义相似度搜索找到最相关的历史片段而非仅仅按会话ID。记忆摘要对于长会话不要存储所有原始消息。可以定期或在会话结束时让大模型生成一个“会话摘要”只存储摘要大幅减少存储和检索的负担。数据库索引优化确保session_id和sender_email字段已建立索引。6.3 上下文长度限制与Token超限现象当会话历史很长时拼接的Prompt可能超过大模型的上下文窗口限制导致调用失败。解决方案限制历史条数如代码中的limit5只取最近N条消息。动态摘要在Prompt中不直接放入全部历史而是先让模型根据当前问题从历史中提取最关键的信息形成一个简短的“上下文摘要”再将摘要放入主Prompt。使用支持长上下文模型选择上下文窗口更大的模型。6.4 长期记忆的更新策略冲突现象模型提取的extracted_info_for_memory可能不准确或相互矛盾导致长期记忆被错误覆盖。解决方案置信度过滤让模型在提取信息时附带一个置信度分数只存储高置信度的信息。人工审核对于关键信息如客户等级、合同金额的更新可以先存入“待审核”表由人工确认后再更新到正式长期记忆。版本化或追加对于某些记忆如“历史问题”可以采用追加模式而非覆盖记录该用户所有报告过的问题列表。7. 生产环境最佳实践与扩展方向将原型发展为生产可用的系统需要考虑更多因素。7.1 生产环境检查清单维度学习/开发环境生产环境建议数据库SQLitePostgreSQL/MySQL考虑读写分离、备份策略。记忆向量化推荐专用向量库。大模型API直接调用配置重试、熔断、降级、多个API Key轮询。设置合理的超时和限流。安全性本地运行API接口需增加认证API Key/JWT。用户邮箱等PII信息需脱敏或加密存储。可观测性打印日志集成结构化日志如JSON Log记录每次处理的session_id、模型调用耗时、决策结果。接入监控和告警。性能顺序处理引入消息队列如RabbitMQ, Redis Queue异步处理邮件避免HTTP请求阻塞。记忆管理永不过期设计记忆过期和归档策略。定期清理过于陈旧的会话记忆或将长期记忆迁移到冷存储。7.2 扩展功能方向集成真实邮件服务使用imaplib/smtplib或第三方库如exchangelibfor Outlook, Gmail API替换模拟接口实现真正的邮箱监听和发送。工具调用增强让Agent能执行更丰富的动作如query_knowledge_base: 从公司内部知识库通过RAG检索答案。check_order_status: 调用内部订单系统API查询状态。schedule_meeting: 调用日历API安排会议。 这需要扩展Prompt中的动作定义并实现相应的工具函数。工作流引擎复杂化使用LangGraph或类似框架来定义更复杂、带状态循环的决策流程。例如当动作为need_more_info时自动进入一个“追问”子流程。多模态处理支持处理邮件中的图片、附件提取其中的文字信息使用OCR或分析附件内容。评估与持续学习建立反馈机制收集人工对自动回复的评分好/差用于微调模型或优化Prompt。7.3 关键配置参数说明在app/config.py中集中管理配置以下是一些关键参数# app/config.py import os from dotenv import load_dotenv load_dotenv() class Config: # 大模型相关 LLM_API_KEY os.getenv(LLM_API_KEY) LLM_BASE_URL os.getenv(LLM_BASE_URL, https://api.openai.com/v1) # 适配国内模型 LLM_MODEL os.getenv(LLM_MODEL, gpt-3.5-turbo) LLM_TEMPERATURE float(os.getenv(LLM_TEMPERATURE, 0.2)) LLM_MAX_TOKENS int(os.getenv(LLM_MAX_TOKENS, 2000)) # 记忆相关 CONVERSATION_HISTORY_LIMIT int(os.getenv(CONVERSATION_HISTORY_LIMIT, 5)) # 是否启用长期记忆 ENABLE_LONG_TERM_MEMORY os.getenv(ENABLE_LONG_TERM_MEMORY, true).lower() true # 数据库 DATABASE_URL os.getenv(DATABASE_URL, sqlite:///./mail_agent.db) # 代理与超时 HTTP_PROXY os.getenv(HTTP_PROXY) REQUEST_TIMEOUT int(os.getenv(REQUEST_TIMEOUT, 30))构建一个具备记忆能力的邮件智能体核心在于将离散的邮件交互转化为连续的、有状态的对话。通过本文的实践你不仅实现了一个原型更重要的是理解了记忆存储、检索、更新与大模型协同工作的完整链路。在实际项目中应从简单的规则和关键词匹配入手逐步引入语义记忆和复杂决策并始终将系统的可靠性、可解释性和安全性置于首位。下一步你可以尝试为你的智能体添加第一个真正的工具调用例如连接一个FAQ数据库让它从“能记忆”进化到“能查资料、能行动”。