企业官网建设流程全解析

如果你正在构建基于大语言模型的智能应用，可能会遇到这样的困境：传统RAG系统总是机械地检索文档，无论用户问题是简单问候还是复杂查询，都要走一遍完整的检索流程。这不仅浪费计算资源，还降低了响应速度。更糟糕的是，当检索到的文档与问题不相关时，系统仍然基于这些无关内容生成答案，导致回答质量下降。

这正是Agentic RAG要解决的核心问题——让LLM能够智能决策何时检索、何时直接回答，以及如何评估检索结果的相关性。而LangGraph作为构建复杂AI工作流的框架，为实现这种智能决策提供了强大的工具链。

本文将带你从零构建一个完整的Agentic RAG系统，重点不是简单调用API，而是深入理解每个决策环节的设计原理和实现细节。你会发现，真正的智能不在于工具本身，而在于如何设计让工具协同工作的流程。

1. 环境准备与依赖安装

在开始构建之前，我们需要准备开发环境。LangGraph生态包含多个组件，每个都有特定的用途。

# 安装核心依赖包 pip install -U langgraph langchain-anthropic langchain-text-splitters bs4 requests # 如果需要使用OpenAI模型，还需要安装 pip install langchain-openai

环境配置的关键是正确设置API密钥。不同的模型提供商需要不同的密钥：

import getpass import os def setup_environment(): """设置必要的环境变量""" # OpenAI API密钥 if "OPENAI_API_KEY" not in os.environ: os.environ["OPENAI_API_KEY"] = getpass.getpass("OpenAI API Key: ") # 可选：Anthropic Claude API密钥 if "ANTHROPIC_API_KEY" not in os.environ: os.environ["ANTHROPIC_API_KEY"] = getpass.getpass("Anthropic API Key: ") setup_environment()

重要提醒：在生产环境中，建议使用环境变量或安全的密钥管理服务，避免在代码中硬编码密钥。本文为演示目的使用getpass，实际项目中请根据安全要求调整。

2. LangGraph核心概念解析

在深入代码之前，需要理解LangGraph的几个核心概念，这些概念构成了我们构建智能工作流的基础。

2.1 状态管理（State Management）

LangGraph使用状态对象在节点间传递数据。对于聊天应用，最常用的是MessagesState：

from langgraph.graph import MessagesState from typing import TypedDict, List from langchain_core.messages import BaseMessage # MessagesState的简化定义 class ChatState(TypedDict): messages: List[BaseMessage] # 存储对话消息历史

状态对象确保每个节点都能访问到完整的对话上下文，这是实现多轮对话的基础。

2.2 节点（Nodes）与边（Edges）

节点是工作流中的处理单元，边定义了节点间的流转逻辑：

from langgraph.graph import StateGraph, END # 创建状态图 workflow = StateGraph(MessagesState) # 添加节点 workflow.add_node("process_input", process_input_function) workflow.add_node("generate_response", generate_response_function) # 添加边：定义执行顺序 workflow.add_edge("process_input", "generate_response") workflow.add_edge("generate_response", END)

2.3 条件边（Conditional Edges）

条件边让工作流具备决策能力，这是实现智能RAG的关键：

def should_retrieve(state: MessagesState) -> str: """根据当前状态决定下一步行动""" last_message = state["messages"][-1] # 如果最后一条消息包含工具调用，需要检索 if hasattr(last_message, "tool_calls") and last_message.tool_calls: return "retrieve" # 否则直接生成回答 return "generate_response" # 添加条件边 workflow.add_conditional_edges( "process_input", should_retrieve, { "retrieve": "retrieve_documents", "generate_response": "generate_response" } )

理解这些概念后，我们就能开始构建具体的RAG组件了。

3. 文档预处理与向量化

智能RAG系统的质量很大程度上取决于检索基础。我们先从文档处理开始。

3.1 文档获取与解析

import bs4 import requests from langchain_core.documents import Document from langchain_text_splitters import RecursiveCharacterTextSplitter def load_web_page(url: str) -> List[Document]: """从网页加载文档内容""" try: response = requests.get(url, timeout=30) response.raise_for_status() soup = bs4.BeautifulSoup(response.text, "html.parser") # 移除脚本和样式标签 for script in soup(["script", "style"]): script.decompose() text_content = soup.get_text() # 清理多余的空白字符 lines = (line.strip() for line in text_content.splitlines()) chunks = (phrase.strip() for line in lines for phrase in line.split(" ")) text_content = ' '.join(chunk for chunk in chunks if chunk) return [Document(page_content=text_content, metadata={"source": url})] except Exception as e: print(f"Error loading {url}: {e}") return [] # 示例：加载技术博客文章 urls = [ "https://lilianweng.github.io/posts/2024-11-28-reward-hacking/", "https://lilianweng.github.io/posts/2024-07-07-hallucination/", "https://lilianweng.github.io/posts/2024-04-12-diffusion-video/", ] # 并行加载多个文档 docs = [] for url in urls: doc_chunks = load_web_page(url) docs.extend(doc_chunks) print(f"Loaded {len(docs)} documents")

3.2 文本分割与向量化

文档分割策略直接影响检索质量。过大的块会包含无关信息，过小的块可能丢失上下文。

def prepare_documents(documents: List[Document]) -> List[Document]: """预处理文档：分割和清理""" # 创建文本分割器 text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每个块约500字符 chunk_overlap=50, # 块间重叠50字符 length_function=len, separators=["\n\n", "\n", ". ", "! ", "? ", " ", ""] ) # 分割文档 all_splits = [] for doc in documents: splits = text_splitter.split_documents([doc]) # 为每个分割块添加源信息 for split in splits: split.metadata["source"] = doc.metadata.get("source", "unknown") split.metadata["chunk_id"] = len(all_splits) all_splits.extend(splits) return all_splits # 处理文档 doc_splits = prepare_documents(docs) print(f"Created {len(doc_splits)} document chunks")

3.3 创建向量检索器

from langchain_core.vectorstores import InMemoryVectorStore from langchain_openai import OpenAIEmbeddings from functools import lru_cache @lru_cache(maxsize=1) def create_retriever(): """创建向量检索器（使用缓存避免重复创建）""" # 初始化嵌入模型 embeddings = OpenAIEmbeddings(model="text-embedding-3-small") # 创建内存向量存储 vectorstore = InMemoryVectorStore.from_documents( documents=doc_splits, embedding=embeddings, ) # 配置检索器：返回最相关的3个文档 retriever = vectorstore.as_retriever( search_type="similarity", search_kwargs={"k": 3} ) return retriever # 测试检索器 retriever = create_retriever() test_results = retriever.invoke("什么是reward hacking？") print(f"检索到 {len(test_results)} 个相关文档")

4. 构建智能检索工具

现在创建核心的检索工具，让LLM能够按需调用。

from langchain.tools import tool from typing import Dict, Any @tool def retrieve_blog_posts(query: str) -> str: """ 根据查询检索相关的博客文章内容。 Args: query: 搜索查询字符串 Returns: 检索到的相关文档内容，用换行符分隔 """ try: retriever = create_retriever() retrieved_docs = retriever.invoke(query) # 格式化返回结果 results = [] for i, doc in enumerate(retrieved_docs): source = doc.metadata.get("source", "未知来源") content_preview = doc.page_content[:200] + "..." if len(doc.page_content) > 200 else doc.page_content results.append(f"[文档 {i+1} - {source}]: {content_preview}") return "\n\n".join(results) except Exception as e: return f"检索过程中出现错误: {str(e)}" # 测试工具 retriever_tool = retrieve_blog_posts test_query = "reward hacking的类型" result = retriever_tool.invoke({"query": test_query}) print("检索工具测试结果:") print(result)

5. 实现智能决策节点

这是Agentic RAG的核心部分，我们创建多个决策节点来实现智能工作流。

5.1 查询生成与响应决策节点

from langchain.chat_models import init_chat_model from langgraph.graph import MessagesState # 初始化响应模型 response_model = init_chat_model("openai:gpt-4o-mini", temperature=0) def generate_query_or_respond(state: MessagesState): """ 决策节点：根据用户输入决定是检索文档还是直接响应 这个节点让LLM自主判断： - 如果问题需要外部知识：调用检索工具 - 如果是一般对话：直接响应 """ # 绑定工具，让模型知道可以调用检索功能 model_with_tools = response_model.bind_tools([retriever_tool]) # 基于当前对话状态生成响应 response = model_with_tools.invoke(state["messages"]) return {"messages": [response]} # 测试决策逻辑 test_cases = [ {"role": "user", "content": "你好！"}, # 应该直接响应 {"role": "user", "content": "Lilian Weng关于reward hacking的观点是什么？"} # 应该调用检索 ] for i, test_case in enumerate(test_cases): state = {"messages": [test_case]} result = generate_query_or_respond(state) last_msg = result["messages"][-1] print(f"测试用例 {i+1}: {test_case['content']}") if hasattr(last_msg, 'tool_calls') and last_msg.tool_calls: print("决策：需要检索文档") print(f"检索查询: {last_msg.tool_calls[0]['args']['query']}") else: print("决策：直接响应") print("---")

5.2 文档相关性评估节点

检索到文档后，需要评估其与问题的相关性，避免基于无关内容生成答案。

from pydantic import BaseModel, Field from typing import Literal class GradeDocuments(BaseModel): """文档相关性评估结果""" binary_score: str = Field( description="相关性评分：'yes'表示相关，'no'表示不相关", pattern="^(yes|no)$" ) # 相关性评估提示词 GRADE_PROMPT = """你是一个评估检索文档与用户问题相关性的评分器。 检索到的文档内容： <context> {context} </context> 用户问题：{question} 请严格评估文档内容是否与用户问题相关。如果文档包含与问题相关的关键词或语义含义，评为相关('yes')，否则评为不相关('no')。 只返回二进制评分，不要解释。""" def grade_documents(state: MessagesState) -> Literal["generate_answer", "rewrite_question"]: """ 评估检索到的文档与用户问题的相关性 Returns: "generate_answer": 文档相关，直接生成答案 "rewrite_question": 文档不相关，需要重写问题 """ # 初始化评估模型 grader_model = init_chat_model("openai:gpt-4o-mini", temperature=0) # 提取问题和上下文 question = state["messages"][0].content context = state["messages"][-1].content # 构建评估提示 prompt = GRADE_PROMPT.format(question=question, context=context) # 获取结构化评估结果 response = grader_model.with_structured_output(GradeDocuments).invoke( [{"role": "user", "content": prompt}] ) # 根据评估结果路由到不同节点 if response.binary_score == "yes": return "generate_answer" return "rewrite_question" # 测试评估逻辑 test_contexts = [ "reward hacking可以分为环境错误规范和奖励篡改两种类型", # 相关 "深度学习模型在图像生成中的应用", # 不相关 ] for context in test_contexts: test_state = { "messages": [ {"role": "user", "content": "reward hacking有哪些类型？"}, {"role": "assistant", "content": "", "tool_calls": [{"id": "1", "name": "retrieve_blog_posts", "args": {"query": "reward hacking类型"}}]}, {"role": "tool", "content": context, "tool_call_id": "1"} ] } decision = grade_documents(test_state) relevance = "相关" if decision == "generate_answer" else "不相关" print(f"文档内容: {context[:50]}... → 评估结果: {relevance}")

5.3 问题重写节点

当检索结果不相关时，可能是原始问题表述不够清晰，需要重写问题。

from langchain.messages import HumanMessage REWRITE_PROMPT = """请分析用户的原始问题，并推理其背后的语义意图。 原始问题： ------- {question} ------- 当前检索到的文档与问题不相关，这可能是因为： 1. 问题表述不够具体 2. 使用了不准确的关键词 3. 问题范围太宽泛或太狭窄 请重新表述一个更可能检索到相关信息的问题：""" def rewrite_question(state: MessagesState): """重写用户问题以提高检索效果""" question = state["messages"][0].content prompt = REWRITE_PROMPT.format(question=question) response = response_model.invoke([{"role": "user", "content": prompt}]) # 创建重写后的问题消息 rewritten_question = HumanMessage(content=response.content) return {"messages": [rewritten_question]} # 测试问题重写 test_question = "关于AI安全的一些想法" rewritten = rewrite_question({"messages": [{"role": "user", "content": test_question}]}) print(f"原始问题: {test_question}") print(f"重写后: {rewritten['messages'][-1].content}")

5.4 答案生成节点

当文档相关时，基于检索内容生成最终答案。

GENERATE_PROMPT = """你是一个问答助手，请基于检索到的上下文回答问题。 要求： - 严格基于提供的上下文内容回答 - 如果上下文不包含答案，明确说明不知道 - 保持回答简洁，最多3句话 - 不要编造上下文之外的信息 问题：{question} 上下文： <context> {context} </context> 请基于以上上下文回答问题：""" def generate_answer(state: MessagesState): """基于相关文档生成答案""" question = state["messages"][0].content context = state["messages"][-1].content prompt = GENERATE_PROMPT.format(question=question, context=context) response = response_model.invoke([{"role": "user", "content": prompt}]) return {"messages": [response]} # 测试答案生成 test_qa_state = { "messages": [ {"role": "user", "content": "reward hacking有哪些类型？"}, {"role": "assistant", "content": "", "tool_calls": [{"id": "1", "name": "retrieve_blog_posts", "args": {"query": "reward hacking类型"}}]}, {"role": "tool", "content": "reward hacking主要包括两种类型：环境错误规范和奖励篡改。环境错误规范指智能体利用环境漏洞获得奖励，奖励篡改指直接修改奖励函数。", "tool_call_id": "1"} ] } answer_result = generate_answer(test_qa_state) print("生成的答案:") print(answer_result["messages"][-1].content)

6. 组装完整的工作流图

现在我们将所有节点组装成完整的智能RAG工作流。

from langgraph.graph import END, START, StateGraph from langgraph.prebuilt import ToolNode def create_agentic_rag_graph(): """创建完整的Agentic RAG工作流图""" workflow = StateGraph(MessagesState) # 添加所有节点 workflow.add_node("generate_query_or_respond", generate_query_or_respond) workflow.add_node("retrieve", ToolNode([retriever_tool])) workflow.add_node("rewrite_question", rewrite_question) workflow.add_node("generate_answer", generate_answer) # 设置起始节点 workflow.add_edge(START, "generate_query_or_respond") def route_on_tool_calls(state: MessagesState): """根据工具调用情况路由""" last_message = state["messages"][-1] if hasattr(last_message, "tool_calls") and last_message.tool_calls: return "retrieve" return END # 决策：是否需要检索 workflow.add_conditional_edges( "generate_query_or_respond", route_on_tool_calls, { "retrieve": "retrieve", END: END } ) # 决策：文档是否相关 workflow.add_conditional_edges( "retrieve", grade_documents, { "generate_answer": "generate_answer", "rewrite_question": "rewrite_question" } ) # 设置结束边 workflow.add_edge("generate_answer", END) # 重写问题后重新决策 workflow.add_edge("rewrite_question", "generate_query_or_respond") # 编译图 graph = workflow.compile() return graph # 创建并测试完整工作流 rag_graph = create_agentic_rag_graph() print("Agentic RAG工作流创建成功！") print("工作流节点:", list(rag_graph.nodes))

7. 运行与测试智能RAG系统

现在让我们测试完整的系统，观察其智能决策过程。

def test_agentic_rag_system(): """全面测试Agentic RAG系统""" test_cases = [ { "name": "简单问候", "input": "你好，请介绍下你自己", "expected": "直接响应" # 应该不需要检索 }, { "name": "具体技术问题", "input": "Lilian Weng如何分类reward hacking的类型？", "expected": "检索后回答" # 应该检索相关文档 }, { "name": "模糊问题", "input": "AI安全相关的内容", "expected": "可能重写问题" # 初始检索可能不相关 } ] for i, test_case in enumerate(test_cases): print(f"\n=== 测试用例 {i+1}: {test_case['name']} ===") print(f"输入: {test_case['input']}") # 运行工作流 result = rag_graph.invoke({ "messages": [{"role": "user", "content": test_case["input"]}] }) # 分析执行路径 final_message = result["messages"][-1] has_tool_calls = hasattr(final_message, 'tool_calls') and final_message.tool_calls if has_tool_calls: print("决策路径: 需要检索文档") print(f"检索查询: {final_message.tool_calls[0]['args']['query']}") else: print("决策路径: 直接生成回答") print(f"回答: {final_message.content}") print(f"预期: {test_case['expected']}") print("=" * 50) # 运行测试 test_agentic_rag_system()

7.1 流式输出实现

对于实际应用，流式输出能提供更好的用户体验。

def stream_agentic_rag_response(question: str): """流式输出Agentic RAG的响应过程""" print(f"用户问题: {question}") print("处理过程:") print("-" * 30) events = rag_graph.stream( {"messages": [{"role": "user", "content": question}]}, config={"recursion_limit": 50} ) for event in events: if "generate_query_or_respond" in event: print("✓ 决策节点: 分析问题意图") elif "retrieve" in event: print("✓ 检索节点: 搜索相关文档") elif "grade_documents" in event: print("✓ 评估节点: 检查文档相关性") elif "rewrite_question" in event: print("✓ 重写节点: 优化问题表述") elif "generate_answer" in event: print("✓ 生成节点: 创建最终答案") print("-" * 30) final_result = rag_graph.invoke( {"messages": [{"role": "user", "content": question}]} ) print("最终回答:", final_result["messages"][-1].content) # 测试流式输出 stream_agentic_rag_response("请解释reward hacking的概念和类型")

8. 高级功能与优化

8.1 添加对话记忆

让系统能够处理多轮对话，记住之前的上下文。

from langgraph.checkpoint.memory import MemorySaver def create_rag_with_memory(): """创建带记忆功能的RAG系统""" # 添加记忆存储 memory = MemorySaver() # 创建带记忆的工作流 graph_with_memory = create_agentic_rag_graph() graph_with_memory = graph_with_memory.compile(checkpointer=memory) return graph_with_memory # 测试多轮对话 memory_graph = create_rag_with_memory() def test_multi_turn_conversation(): """测试多轮对话能力""" config = {"configurable": {"thread_id": "test_conversation_1"}} # 第一轮 print("用户: 什么是reward hacking？") result1 = memory_graph.invoke( {"messages": [{"role": "user", "content": "什么是reward hacking？"}]}, config=config ) print("助手:", result1["messages"][-1].content) # 第二轮（基于上文） print("\n用户: 能详细说说它的类型吗？") result2 = memory_graph.invoke( {"messages": [{"role": "user", "content": "能详细说说它的类型吗？"}]}, config=config ) print("助手:", result2["messages"][-1].content) test_multi_turn_conversation()

8.2 性能优化建议

在实际项目中，这些优化能显著提升系统性能：

import time from functools import wraps def timing_decorator(func): """性能计时装饰器""" @wraps(func) def wrapper(*args, **kwargs): start_time = time.time() result = func(*args, **kwargs) end_time = time.time() print(f"{func.__name__} 执行时间: {end_time - start_time:.2f}秒") return result return wrapper # 为关键函数添加性能监控 @timing_decorator def optimized_retrieve(query: str) -> str: """优化版本的检索函数""" # 添加缓存逻辑 # 添加批量处理 # 添加超时控制 return retrieve_blog_posts(query) # 性能优化配置建议 optimization_tips = """ Agentic RAG性能优化建议： 1. 检索优化： - 使用更高效的向量数据库（Chroma、Pinecone） - 实现检索结果缓存 - 调整chunk大小和重叠比例 2. 模型优化： - 根据场景选择合适大小的模型 - 实现响应流式输出 - 设置合理的超时时间 3. 系统优化： - 实现异步处理 - 添加速率限制 - 监控关键指标（延迟、准确率） """ print(optimization_tips)

9. 常见问题与解决方案

在实际部署中，你可能会遇到这些问题：

9.1 检索相关性问题

问题：检索到的文档与问题不相关

解决方案：

def improve_retrieval_quality(): """提升检索质量的实用技巧""" improvements = """ 1. 优化文本分割策略： - 尝试不同的chunk_size（300-1000字符） - 调整chunk_overlap（10%-20%） - 基于语义分割而非固定长度 2. 改进检索参数： - 调整返回文档数量（k值） - 尝试混合检索（语义+关键词） - 使用重排序（re-ranking）技术 3. 增强查询理解： - 实现查询扩展 - 添加同义词处理 - 考虑多语言支持 """ return improvements print(improve_retrieval_quality())

9.2 错误处理与容错

def robust_rag_invocation(question: str, max_retries: int = 3): """带错误处理的RAG调用""" for attempt in range(max_retries): try: result = rag_graph.invoke( {"messages": [{"role": "user", "content": question}]} ) return result except Exception as e: print(f"第{attempt + 1}次尝试失败: {str(e)}") if attempt == max_retries - 1: # 最后一次尝试失败，返回降级响应 return {"messages": [{ "role": "assistant", "content": "抱歉，系统暂时无法处理您的请求。请稍后再试或简化您的问题。" }]} time.sleep(1) # 重试前等待 # 测试容错能力 result = robust_rag_invocation("test question") print("容错测试结果:", result["messages"][-1].content)

10. 生产环境部署建议

当准备将Agentic RAG系统部署到生产环境时，考虑以下关键因素：

10.1 安全考虑

def security_considerations(): """生产环境安全注意事项""" security_checklist = """ 🔒 安全清单： 1. API密钥管理： - 使用环境变量或密钥管理服务 - 定期轮换密钥 - 设置最小权限原则 2. 输入验证： - 验证用户输入长度和内容 - 防止提示词注入攻击 - 实现速率限制 3. 输出过滤： - 检查模型输出是否包含敏感信息 - 实现内容审核机制 - 添加滥用检测 4. 数据隐私： - 匿名化用户数据 - 遵守数据保护法规 - 定期清理日志 """ return security_checklist print(security_considerations())

10.2 监控与日志

def setup_monitoring(): """设置系统监控""" monitoring_config = """ 📊 监控配置建议： 1. 关键指标监控： - 请求延迟（P50、P95、P99） - 错误率和成功率 - Token使用量统计 2. 业务指标： - 检索相关性评分 - 用户满意度反馈 - 对话完成率 3. 日志策略： - 结构化日志记录 - 敏感信息脱敏 - 日志保留策略 """ return monitoring_config print(setup_monitoring())

通过本文的完整实现，你已经构建了一个真正智能的Agentic RAG系统。关键在于理解每个决策节点的作用，以及如何将它们组合成有效的工作流。这种架构不仅提升了回答质量，还显著优化了资源使用效率。

实际项目中，建议从简单版本开始，逐步添加复杂功能，并持续基于用户反馈进行优化。智能RAG系统的价值在于它能随着使用不断改进，真正理解用户需求并提供精准的知识服务。