企业官网建设流程全解析

1. 项目概述

LangGraph 是一个基于图结构的 AI 开发框架，它允许开发者通过连接不同的功能模块来构建复杂的 AI 应用。最近我在一个客户项目中尝试使用 LangGraph 实现了一个具备 RAG（检索增强生成）和长期记忆能力的 AI Agent，效果出乎意料地好。这个 Agent 不仅能回答专业领域的问题，还能记住对话历史并在后续交流中主动引用，大大提升了用户体验。

对于需要处理复杂任务、长期交互的 AI 系统来说，RAG 和长期记忆是两大核心能力。RAG 让 AI 能够从外部知识库获取最新信息，而长期记忆则使 AI 能够建立上下文感知和个性化交互。本文将详细介绍如何利用 LangGraph 的强大功能实现这两项关键技术。

2. 核心架构设计

2.1 LangGraph 基础概念

LangGraph 的核心思想是将 AI 应用建模为有向图，其中节点代表处理单元，边代表数据流向。与传统的线性链式处理不同，图结构允许更复杂的逻辑流，包括：

• 条件分支
• 并行处理
• 循环反馈
• 状态保持

在 LangGraph 中，每个节点都是一个独立的函数或 LangChain 组件，可以执行特定任务，如：

• 调用 LLM
• 执行检索
• 处理数据
• 管理记忆

2.2 系统整体架构

我们设计的 Agent 架构包含以下核心组件：

[用户输入] → [输入解析节点] → [记忆检索节点] → [RAG检索节点] → [响应生成节点] → [记忆更新节点] → [输出响应]

这个流程看似线性，但实际上每个节点都可能触发子图或循环。例如，当记忆检索节点发现相关信息不足时，可以跳过直接进入 RAG 检索；当响应生成节点需要多轮思考时，可以形成内部循环。

3. RAG 实现细节

3.1 知识库构建

有效的 RAG 系统始于高质量的知识库构建。我们采用了以下流程：

1. 数据收集：从客户提供的 PDF、网页和内部文档中提取文本
2. 分块处理：使用递归字符分割器，设置 chunk_size=1000，chunk_overlap=200
3. 向量化：采用 OpenAI 的 text-embedding-3-large 模型生成嵌入
4. 索引存储：使用 FAISS 实现高效的相似度搜索

提示：分块大小对检索质量影响很大。经过测试，技术文档适合 800-1200 字符的分块，而对话记录更适合 400-600 字符的分块。

3.2 检索节点实现

在 LangGraph 中，我们创建了一个专用的检索节点：

from langchain_community.vectorstores import FAISS from langchain_core.retrievers import BaseRetriever class CustomRetriever(BaseRetriever): def __init__(self, vectorstore): self.vectorstore = vectorstore def get_relevant_documents(self, query): # 添加查询重写逻辑 expanded_query = self._expand_query(query) return self.vectorstore.similarity_search(expanded_query, k=3) def _expand_query(self, query): # 使用小模型生成查询扩展 return query + " " + generate_query_expansion(query)

这个检索器不仅执行简单的相似度搜索，还加入了查询扩展功能，显著提高了召回率。

3.3 检索结果融合

检索到的文档需要与对话上下文融合后才能送入生成环节。我们实现了以下策略：

1. 相关性过滤：只保留相似度分数 > 0.7 的结果
2. 多样性采样：确保结果来自不同文档源
3. 上下文压缩：使用 LLM 提取最相关的片段

def format_retrieved_docs(docs, query): context = "\n\n".join([f"来源：{doc.metadata['source']}\n内容：{doc.page_content}" for doc in docs]) compressed = llm.invoke(f"基于问题'{query}'，从以下文本中提取最相关信息：\n{context}") return compressed

4. 长期记忆实现

4.1 记忆存储设计

长期记忆系统需要解决三个核心问题：

1. 存储什么：对话历史、用户偏好、事实知识
2. 如何存储：向量存储 + 传统数据库
3. 何时更新：每次交互后增量更新

我们采用了分层存储架构：

• 短期记忆：保存在对话状态中，容量有限但访问快
• 长期记忆：存储在外部数据库中，容量大但检索慢
• 摘要记忆：定期生成的对话摘要，平衡细节与效率

4.2 记忆检索优化

简单的向量检索在记忆系统中效果不佳，因为用户可能以不同方式提及相同概念。我们实现了多路检索策略：

1. 直接向量检索：标准的相似度搜索
2. 时间加权检索：最近记忆获得更高权重
3. 关联检索：通过知识图谱查找相关概念
4. 元数据过滤：按记忆类型（事实/偏好/事件）筛选

def retrieve_memories(query, user_id): # 并行执行多种检索 vector_results = vector_store.search(query) temporal_results = temporal_index.search(query, recency_weight=0.3) related_results = kg.find_related_concepts(query) # 结果融合与去重 all_results = merge_results(vector_results, temporal_results, related_results) return rank_results(all_results)

4.3 记忆更新策略

记忆系统最容易犯的错误是存储过多无关信息。我们制定了严格的更新规则：

1. 重要性评估：使用 LLM 判断信息是否值得长期记忆
2. 去重检查：与现有记忆比较，避免重复
3. 摘要生成：定期将多个相关记忆合并为更高层次的摘要

def should_remember(text, user_id): response = llm.invoke( f"评估以下文本是否值得作为长期记忆存储：\n{text}\n" "仅回答'是'或'否'" ) return response.strip() == "是"

5. LangGraph 实现技巧

5.1 状态管理

LangGraph 的核心是状态对象，它在节点间传递。我们设计了专门的状态类：

class AgentState(TypedDict): user_input: str context: List[dict] memories: List[dict] retrieved_docs: List[Document] response: Optional[str]

5.2 条件边与循环

通过条件边实现复杂逻辑流：

from langgraph.graph import Graph workflow = Graph() # 定义节点 workflow.add_node("parse_input", parse_input) workflow.add_node("retrieve_memories", retrieve_memories) # ... 其他节点 # 定义边 workflow.add_edge("parse_input", "retrieve_memories") workflow.add_conditional_edge( "retrieve_memories", decide_next_step, { "enough_info": "generate_response", "need_more": "retrieve_docs" } )

5.3 调试技巧

LangGraph 应用调试比较困难，我们总结了以下方法：

1. 可视化图结构：使用workflow.get_graph().draw_mermaid()生成流程图
2. 状态检查点：在每个节点后记录状态快照
3. 单元测试节点：单独测试每个节点函数
4. 追踪执行路径：添加日志记录决策过程

6. 性能优化

6.1 缓存策略

重复计算是性能杀手，我们实现了多级缓存：

1. LLM 响应缓存：相同输入直接返回缓存结果
2. 检索结果缓存：相似查询返回近似结果
3. 记忆缓存：高频访问记忆保留在内存中

from langchain.cache import SQLiteCache from langchain.globals import set_llm_cache set_llm_cache(SQLiteCache(database_path=".langchain.db"))

6.2 异步执行

利用 LangGraph 的异步支持提升吞吐量：

async def aprocess_node(state): # 并行执行独立任务 mem_future = retrieve_memories(state["user_input"]) docs_future = retrieve_docs(state["user_input"]) memories, docs = await asyncio.gather(mem_future, docs_future) return {**state, "memories": memories, "retrieved_docs": docs}

6.3 负载测试

使用 locust 模拟不同负载下的表现：

from locust import HttpUser, task class AgentUser(HttpUser): @task def chat(self): self.client.post("/chat", json={"message": "测试消息"})

测试结果指导我们优化批处理大小和并发设置。

7. 实际应用案例

7.1 技术支持助手

为客户实现的 IT 支持助手表现：

• 平均响应时间：1.2 秒
• 首次解决率：78%
• 用户满意度：4.6/5.0

关键成功因素：

• 整合了产品文档和社区讨论的 RAG 知识库
• 能记住用户的设备配置和历史问题
• 复杂问题自动转人工的平滑交接

7.2 教育辅导助手

一个数学辅导 Agent 的特殊设计：

• 分步解题而非直接给答案
• 记住学生的薄弱知识点
• 根据学习进度调整难度

def generate_hint(problem, student_id): mistakes = get_common_mistakes(student_id) return llm.invoke( f"为这个问题生成提示，避免学生常犯的这些错误：{mistakes}\n" f"问题：{problem}" )

8. 常见问题与解决

8.1 记忆检索不准确

症状：Agent 频繁回忆无关信息解决：

1. 调整向量相似度阈值
2. 添加元数据过滤
3. 实现记忆衰减机制

def decay_memory_weights(user_id): # 每隔一段时间降低旧记忆的权重 for memory in get_memories(user_id): age = now() - memory["timestamp"] memory["weight"] *= 0.9 ** (age.days / 30)

8.2 RAG 结果不相关

症状：检索到的文档不能回答问题解决：

1. 优化分块策略
2. 添加查询扩展
3. 实现重排序模型

8.3 响应速度慢

症状：简单查询也耗时较长解决：

1. 实现缓存层
2. 优化图结构，减少不必要节点
3. 设置超时和回退机制

9. 进阶技巧

9.1 记忆压缩

长期积累的记忆会变得臃肿，我们定期运行压缩流程：

1. 聚类相似记忆：使用嵌入向量聚类
2. 生成摘要：用 LLM 合并相关记忆
3. 淘汰低频记忆：删除长时间未访问的内容

9.2 个性化适配

通过少量样本学习用户风格：

few_shot_examples = get_user_examples(user_id) prompt = PromptTemplate.from_examples( examples=few_shot_examples, suffix="请以类似风格回答用户问题：{input}" )

9.3 安全防护

防止记忆系统被恶意利用：

1. 输入过滤：检测并阻止有害内容
2. 记忆审查：定期扫描存储的记忆
3. 权限控制：敏感记忆需要额外授权

def is_safe_content(text): return not llm.invoke( f"判断以下文本是否包含有害内容：\n{text}\n" "仅回答'是'或'否'" ).strip() == "是"

在实际部署中，我发现 LangGraph 的最大优势是其灵活性。当客户突然要求添加一个新功能时，通常只需要添加一个新节点或调整图结构，而不需要重构整个系统。例如，当需要增加多模态支持时，我们简单地添加了一个图像处理分支节点，整个集成过程只用了不到一天时间。