网站进入沙盒后遵义交通建设网站

张

张建站

2026/4/19 10:31:29

10分钟阅读

网站进入沙盒后,遵义交通建设网站,沧州商城网站开发设计,网站开发工资多少钱LangChain AI Agent 长期记忆在企业级应用中#xff0c;AI Agent 不仅需要流畅的对话能力#xff0c;还必须具备可靠、可扩展、安全的长期记忆。本教程将带你使用企业常用的技术栈#xff0c;为 LangChain Agent 构建生产就绪的长期记忆方案。文章目录 LangChain AI Ag…LangChain AI Agent 长期记忆在企业级应用中AI Agent 不仅需要流畅的对话能力还必须具备可靠、可扩展、安全的长期记忆。本教程将带你使用企业常用的技术栈为 LangChain Agent 构建生产就绪的长期记忆方案。文章目录LangChain AI Agent 长期记忆1. 企业级技术选型2. 环境准备2.1 安装 Docker 和 Docker Compose2.2 项目结构3. 搭建基础设施Docker Compose4. 实现长期记忆模块4.1 安装依赖4.2 环境配置.env4.3 定义数据模型models.py4.4 PostgreSQL 记忆实现sql\_memory.py4.5 Pinecone 向量记忆实现vector\_memory.py4.6 Redis 缓存可选5. 构建 Agent 并集成记忆5.1 创建 Agent 核心agent.py5.2 创建 FastAPI 应用main.py6. 运行测试7. 企业级增强建议8. 总结1. 企业级技术选型上图展示了我们的记忆架构一个智能机器人将对话记忆分类存储到不同的数据库中——PostgreSQL 负责结构化存储Pinecone 处理向量检索Redis 提供快速缓存。组件技术选择作用关系型数据库PostgreSQL存储结构化对话历史支持事务、ACID、高并发缓存数据库Redis存储短期会话数据加速访问向量数据库Pinecone或 Weaviate存储记忆向量支持语义检索应用框架LangChain FastAPIAgent 核心逻辑提供 REST API容器化Docker Docker Compose简化环境部署确保一致性语言Python 3.9开发语言本教程使用 Pinecone 作为向量数据库示例因为它托管简单无需自建。你也可以替换为开源的 Weaviate 或 Milvus。2. 环境准备2.1 安装 Docker 和 Docker Compose确保你的开发环境已安装 Docker 和 Docker Compose。2.2 项目结构agent-memory-demo/├── docker-compose.yml# 定义PostgreSQL、Redis服务├──.env# 环境变量├── app/│ ├── __init__.py │ ├── main.py# FastAPI入口│ ├── agent.py# Agent核心逻辑│ ├── memory/│ │ ├── __init__.py │ │ ├── sql_memory.py# PostgreSQL记忆实现│ │ └── vector_memory.py# Pinecone记忆实现│ └── models.py# 数据模型└── requirements.txt# Python依赖3. 搭建基础设施Docker ComposeDocker 让我们的服务像火箭一样快速部署确保开发、测试、生产环境的一致性。创建docker-compose.ymlversion:3.8services:postgres:image:postgres:15container_name:agent-postgresenvironment:POSTGRES_USER:agent_userPOSTGRES_PASSWORD:agent_passPOSTGRES_DB:agent_memoryports:-5432:5432volumes:-postgres_data:/var/lib/postgresql/datanetworks:-agent-network redis:image:redis:7-alpinecontainer_name:agent-redisports:-6379:6379networks:-agent-network volumes:postgres_data:networks:agent-network:driver:bridge启动服务docker-compose up -d4. 实现长期记忆模块4.1 安装依赖创建requirements.txtlangchain0.1.0 langchain-openai0.0.2 openai1.6.1 pinecone-client3.0.2 psycopg2-binary2.9.9 redis5.0.1 sqlalchemy2.0.23 fastapi0.104.1 uvicorn0.24.0 python-dotenv1.0.0安装pip install -r requirements.txt4.2 环境配置.env创建.env文件# OpenAI OPENAI_API_KEYyour-openai-key # Pinecone PINECONE_API_KEYyour-pinecone-key PINECONE_ENVIRONMENTyour-pinecone-env PINECONE_INDEX_NAMEagent-memory # PostgreSQL DB_HOSTlocalhost DB_PORT5432 DB_NAMEagent_memory DB_USERagent_user DB_PASSWORDagent_pass # Redis REDIS_HOSTlocalhost REDIS_PORT63794.3 定义数据模型models.pyfrom sqlalchemy import create_engine, Column, String, DateTime, Text, JSON from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker import datetime import os DATABASE_URL fpostgresql://{os.getenv(DB_USER)}:{os.getenv(DB_PASSWORD)}{os.getenv(DB_HOST)}:{os.getenv(DB_PORT)}/{os.getenv(DB_NAME)} engine create_engine(DATABASE_URL) SessionLocal sessionmaker(bindengine) Base declarative_base() class ConversationHistory(Base): __tablename__ conversation_history id Column(String, primary_keyTrue, indexTrue) # 格式user_id:timestamp user_id Column(String, indexTrue) role Column(String) # human 或 ai content Column(Text) metadata Column(JSON, nullableTrue) created_at Column(DateTime, defaultdatetime.datetime.utcnow) # 创建表 Base.metadata.create_all(bindengine)4.4 PostgreSQL 记忆实现sql_memory.pyPostgreSQL 作为可靠的关系型数据库就像一个整齐的图书馆将对话历史有序地存储在书架上。实现一个自定义的BaseMemory类使用 SQLAlchemy 操作 PostgreSQL。fromlangchain.schemaimportBaseMemoryfromlangchain.schema.messagesimportBaseMessage,HumanMessage,AIMessagefromtypingimportList,Dict,Anyfromsqlalchemy.ormimportSessionfrommodelsimportSessionLocal,ConversationHistoryimportuuidimportdatetime classPostgresChatMessageHistory:管理单个用户的对话历史PostgreSQL实现def__init__(self,user_id:str,session:SessionNone):self.user_iduser_id self.sessionsessionorSessionLocal()defadd_message(self,message:BaseMessage):添加一条消息到数据库msg_idf{self.user_id}:{datetime.datetime.utcnow().isoformat()}db_msgConversationHistory(idmsg_id,user_idself.user_id,rolehumanifisinstance(message,HumanMessage)elseai,contentmessage.content,metadatamessage.additional_kwargs)self.session.add(db_msg)self.session.commit()defget_messages(self,limit:int50)-List[BaseMessage]:获取最近的消息默认最近50条recordsself.session.query(ConversationHistory)\.filter(ConversationHistory.user_idself.user_id)\.order_by(ConversationHistory.created_at.desc())\.limit(limit).all()# 按时间正序返回records.reverse()messages[]forrinrecords:ifr.rolehuman:messages.append(HumanMessage(contentr.content,additional_kwargsr.metadataor{}))else:messages.append(AIMessage(contentr.content,additional_kwargsr.metadataor{}))returnmessages classPostgresMemory(BaseMemory):LangChain Memory 适配器使用PostgreSQL存储def__init__(self,user_id:str,memory_key:strchat_history,return_messages:boolTrue):self.user_iduser_id self.memory_keymemory_key self.return_messagesreturn_messages self.historyPostgresChatMessageHistory(user_id)propertydefmemory_variables(self)-List[str]:return[self.memory_key]defload_memory_variables(self,inputs:Dict[str,Any])-Dict[str,Any]:从数据库加载记忆messagesself.history.get_messages()ifself.return_messages:return{self.memory_key:messages}else:# 拼接为字符串return{self.memory_key:\n.join([f{m.type}:{m.content}forminmessages])}defsave_context(self,inputs:Dict[str,Any],outputs:Dict[str,Any])-None:保存对话上下文到数据库input_strinputs.get(input,)output_stroutputs.get(output,)self.history.add_message(HumanMessage(contentinput_str))self.history.add_message(AIMessage(contentoutput_str))defclear(self)-None:清空记忆慎用通常保留# 这里可以添加删除逻辑根据业务需求pass4.5 Pinecone 向量记忆实现vector_memory.pyPinecone 在多维空间中连接相关记忆就像精灵在思考网络中穿梭快速找到语义相似的历史对话。使用 Pinecone 存储向量化记忆实现语义检索。importpineconeimportosfromlangchain.memoryimportVectorStoreRetrieverMemoryfromlangchain_openaiimportOpenAIEmbeddingsfromlangchain_community.vectorstoresimportPineconeasLangchainPineconefromlangchain.schemaimportDocumentclassPineconeVectorMemory:封装Pinecone向量记忆每个用户一个命名空间def__init__(self,user_id:str,namespace:strNone):self.user_iduser_id self.namespacenamespaceorfuser_{user_id}# 初始化Pineconepinecone.init(api_keyos.getenv(PINECONE_API_KEY),environmentos.getenv(PINECONE_ENVIRONMENT))index_nameos.getenv(PINECONE_INDEX_NAME)# 确保索引存在ifindex_namenotinpinecone.list_indexes():pinecone.create_index(nameindex_name,dimension1536,# OpenAI embedding维度metriccosine)self.embeddingsOpenAIEmbeddings()self.vectorstoreLangchainPinecone.from_existing_index(index_nameindex_name,embeddingself.embeddings,namespaceself.namespace)self.retrieverself.vectorstore.as_retriever(search_kwargs{k:3})defadd_memory(self,text:str,metadata:dictNone):添加一条记忆到向量库docDocument(page_contenttext,metadatametadataor{})self.vectorstore.add_documents([doc],namespaceself.namespace)defsearch_memory(self,query:str)-str:检索相关记忆返回拼接的文本docsself.retriever.get_relevant_documents(query)return\n.join([doc.page_contentfordocindocs])# 为了与LangChain集成我们也可以创建一个自定义Memory类classPineconeMemory(BaseMemory):def__init__(self,user_id:str,memory_key:strrelevant_memory):self.user_iduser_id self.memory_keymemory_key self.vector_memoryPineconeVectorMemory(user_id)propertydefmemory_variables(self)-List[str]:return[self.memory_key]defload_memory_variables(self,inputs:Dict[str,Any])-Dict[str,Any]:queryinputs.get(input,)memory_textself.vector_memory.search_memory(query)return{self.memory_key:memory_text}defsave_context(self,inputs:Dict[str,Any],outputs:Dict[str,Any])-None:# 可以选择将重要对话保存为向量记忆# 这里简单地将用户输入和AI输出合并为一条记忆textf用户:{inputs.get(input,)}\nAI:{outputs.get(output,)}self.vector_memory.add_memory(text,metadata{timestamp:str(datetime.datetime.utcnow())})defclear(self):# 清空命名空间中的所有向量危险操作一般不用pass4.6 Redis 缓存可选为了提升性能可以用 Redis 缓存最近几轮对话但非必须。这里简单示意import redis import json import os redis_client redis.Redis( hostos.getenv(REDIS_HOST), portos.getenv(REDIS_PORT), decode_responsesTrue ) def cache_recent_messages(user_id, messages, ttl3600): redis_client.setex(fchat:{user_id}, ttl, json.dumps([m.dict() for m in messages]))5. 构建 Agent 并集成记忆5.1 创建 Agent 核心agent.pyfromlangchain.agentsimportAgentExecutor,create_openai_tools_agentfromlangchain_openaiimportChatOpenAIfromlangchain.promptsimportChatPromptTemplate,MessagesPlaceholderfrommemory.sql_memoryimportPostgresMemoryfrommemory.vector_memoryimportPineconeMemoryimportosdefcreate_agent(user_id:str):llmChatOpenAI(modelgpt-4,temperature0.7,api_keyos.getenv(OPENAI_API_KEY))# 实例化两种记忆pg_memoryPostgresMemory(user_iduser_id,memory_keychat_history,return_messagesTrue)vec_memoryPineconeMemory(user_iduser_id,memory_keyrelevant_memory)# 构建提示模板包含对话历史和向量记忆promptChatPromptTemplate.from_messages([(system,你是一个智能助手。以下是可能相关的长期记忆\n{relevant_memory}\n\n请结合当前对话历史回答。),MessagesPlaceholder(variable_namechat_history),(human,{input}),MessagesPlaceholder(variable_nameagent_scratchpad),])tools[]# 可根据需要添加工具agentcreate_openai_tools_agent(llm,tools,prompt)# 组合多个记忆# AgentExecutor允许多个记忆但memory参数只接受一个。我们可以创建一个组合记忆类# 这里简单起见我们直接在invoke时手动加载记忆# 更好的做法是自定义一个CombinedMemory类继承BaseMemory内部组合两个记忆。# 为了简化教程我们创建一个wrapper来处理。classCombinedMemory(BaseMemory):def__init__(self,memories:List[BaseMemory]):self.memoriesmemoriespropertydefmemory_variables(self)-List[str]:return[varformeminself.memoriesforvarinmem.memory_variables]defload_memory_variables(self,inputs:Dict[str,Any])-Dict[str,Any]:result{}formeminself.memories:result.update(mem.load_memory_variables(inputs))returnresultdefsave_context(self,inputs:Dict[str,Any],outputs:Dict[str,Any])-None:formeminself.memories:mem.save_context(inputs,outputs)defclear(self):formeminself.memories:mem.clear()combined_memoryCombinedMemory([pg_memory,vec_memory])agent_executorAgentExecutor(agentagent,toolstools,memorycombined_memory,verboseTrue,handle_parsing_errorsTrue)returnagent_executor5.2 创建 FastAPI 应用main.py提供 HTTP 接口供外部调用。fromfastapiimportFastAPI,HTTPExceptionfrompydanticimportBaseModelfromagentimportcreate_agentimportuuid appFastAPI(titleAI Agent with Long-term Memory)classChatRequest(BaseModel):user_id:strmessage:strclassChatResponse(BaseModel):user_id:strresponse:strsession_id:strNone# 缓存agent实例简单示例生产环境需管理生命周期agent_cache{}defget_agent(user_id:str):ifuser_idnotinagent_cache:agent_cache[user_id]create_agent(user_id)returnagent_cache[user_id]app.post(/chat,response_modelChatResponse)asyncdefchat(request:ChatRequest):try:agentget_agent(request.user_id)resultagent.invoke({input:request.message})returnChatResponse(user_idrequest.user_id,responseresult[output])exceptExceptionase:raiseHTTPException(status_code500,detailstr(e))app.get(/health)asyncdefhealth():return{status:ok}if__name____main__:importuvicorn uvicorn.run(app,host0.0.0.0,port8000)6. 运行测试确保 Docker 容器已启动PostgreSQL、Redis。在项目根目录运行 FastAPI 应用python app/main.py使用 curl 或 Postman 测试curl -X POST http://localhost:8000/chat \ -H Content-Type: application/json \ -d {user_id: alice, message: 你好我叫爱丽丝}再次发送消息验证记忆curl -X POST http://localhost:8000/chat \ -H Content-Type: application/json \ -d {user_id: alice, message: 你还记得我的名字吗}Agent 应该能够从 PostgreSQL 加载对话历史并从 Pinecone 检索相关向量记忆给出正确的回答。7. 企业级增强建议连接池SQLAlchemy 已内置连接池可配置大小。异步支持使用AsyncSQLAlchemy和asyncpg实现异步数据库操作结合 FastAPI 异步特性提升并发。多租户隔离通过user_id命名空间隔离数据PostgreSQL 使用索引优化查询Pinecone 使用命名空间隔离。安全与认证在 FastAPI 层添加 API 密钥或 JWT 认证确保用户只能访问自己的记忆。监控与日志集成 Prometheus Grafana 监控 API 指标结构化日志输出。记忆压缩定期对过长的对话历史进行摘要并存入向量库删除原始细节。部署使用 Kubernetes 或 Docker Swarm 编排服务实现自动扩缩容。8. 总结本教程展示了如何用企业级技术栈为 LangChain Agent 构建长期记忆PostgreSQL存储结构化对话历史保证数据可靠性和事务性。Pinecone向量数据库支持语义级别的记忆检索。Redis可用于短期缓存加速可选。FastAPI提供 RESTful 接口便于集成到微服务架构中。Docker简化了基础设施的搭建。这个方案兼顾了性能、可扩展性和生产环境要求可直接作为企业项目的起点。你可以根据业务需求调整记忆策略例如只保存摘要、定期清理旧数据等。希望本教程对你有所帮助