30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度在实际 AI 应用开发中LangChain 已经从一个单纯的“链”式编排框架演变为一个包含 LangGraph、LangServe、LangSmith 等多个组件的生态系统。对于开发者而言理解 LangChain 的核心架构、掌握其最新版本如 0.1.x的用法并能够将其与 RAG、智能体Agent、微调等技术结合是构建企业级大模型应用的关键。本文将从一个资深开发者的视角带你深入 LangChain 的工程实践涵盖从核心概念、环境搭建、项目实现到生产级优化的完整链路。无论你是想快速入门 LangChain还是希望解决实际项目中遇到的上下文过长、记忆管理、智能体流程控制等问题都能在本文中找到可复现的解决方案和清晰的排查思路。1. 理解 LangChain 生态从链到图的演进LangChain 最初的核心是“链”Chain它将大语言模型LLM调用、工具使用、记忆等环节串联起来形成一个可预测的执行流程。但随着应用复杂度的提升尤其是需要循环、分支、状态管理等场景的出现单纯的链式结构显得力不从心。这正是 LangGraph 诞生的背景。1.1 LangChain 与 LangGraph 的核心区别简单来说LangChain 的Chain是线性或静态图的执行范式而 LangGraph 引入了有状态、可循环的图执行范式。LangChain Chain: 适用于顺序明确的管道式任务。例如一个典型的 RAG 链可能是检索器 - 格式化提示词 - 调用 LLM - 解析输出。这个流程是固定的输入决定输出没有内部状态循环。LangGraph: 将执行流程建模为一个有向图节点是处理单元可以是 Chain、函数、LLM 等边定义了节点间的流转条件。它维护一个共享的“状态”对象节点可以读取和修改这个状态并根据状态决定下一个执行的节点。这使得实现多轮对话、ReAct 智能体、带有工具循环调用的复杂工作流成为可能。下表对比了两者的关键差异特性LangChain ChainLangGraph执行模型线性/静态图有向图支持循环、分支状态管理隐式通过链的输入输出传递显式通过共享的State对象管理适用场景文档问答、文本总结、简单提取等确定性流程智能体、多轮对话、需要反复尝试或条件判断的复杂工作流控制流有限主要通过RunnableBranch等实现简单分支强大可基于状态任意定义节点跳转条件代码抽象LCEL(LangChain Expression Language) 声明式组合基于StateGraph定义节点和边对于大多数应用LCEL 和 Chain 已经足够。但当你需要构建一个能“思考”并决定下一步做什么的智能体时LangGraph 是更自然的选择。1.2 LangChain 最新版本架构概览以 LangChain 0.1.x 版本为例其核心架构可以理解为几个层次Schema 层: 定义了基础数据对象如Message(HumanMessage,AIMessage,SystemMessage)、Document、ChatGeneration等。这是所有组件交互的“语言”。Model I/O 层: 负责与各种 LLM、ChatModel、EmbeddingModel 交互。通过ChatOpenAI、ChatQwen等类进行抽象。Retrieval 层 (RAG核心): 包含文本分割器、向量存储接口、检索器等用于构建和查询知识库。Chains 层: 基于 LCEL 将其他组件组合成链。Runnable协议是核心一切皆可 Runnable。Agents 层: 构建智能体的高层抽象内部可能由 Chain 或 LangGraph 实现。Memory 层: 管理对话或应用的状态如ConversationBufferMemory。LangGraph: 作为独立库 (langgraph)提供基于图的编排能力。LangServe: 用于将 Chain 或 Graph 快速部署为 REST API。LangSmith: 用于调试、测试、监控和评估链与智能体的平台。理解这个分层有助于在遇到问题时快速定位是哪个环节的配置或调用出了错。2. 环境准备与核心依赖配置开始一个 LangChain 项目前环境的隔离和依赖版本的锁定至关重要不同版本间的 API 差异可能导致代码无法运行。2.1 创建并配置 Python 虚拟环境推荐使用conda或venv管理环境。# 使用 conda conda create -n langchain-demo python3.10 conda activate langchain-demo # 或使用 venv python -m venv venv # Windows venv\Scripts\activate # Linux/Mac source venv/bin/activate2.2 安装核心依赖根据项目需求选择安装。以下是一个面向 RAG 和智能体开发的常用依赖列表使用pip安装。# 核心LangChain pip install langchain langchain-community # LangGraph 用于复杂工作流 pip install langgraph # LangChain 对常见大模型的支持包 (按需安装) pip install langchain-openai # 用于OpenAI, Azure OpenAI # 假设使用通义千问可能需要对应的SDK这里以openai兼容接口为例 # pip install dashscope # 阿里云DashScope SDK # 向量数据库客户端 (例如Chroma) pip install chromadb # 文本嵌入模型 (例如使用OpenAI的text-embedding-3-small) pip install langchain-openai # Web框架用于部署API pip install fastapi uvicorn # 环境变量管理 pip install python-dotenv重要提示langchain是一个元包它会安装一系列核心子包。对于生产环境更推荐显式安装你真正需要的子包如langchain-core,langchain-openai等以减少依赖冲突和包体积。2.3 配置 API 密钥与环境变量永远不要将 API 密钥硬编码在代码中。使用.env文件和环境变量管理。在项目根目录创建.env文件# OpenAI (示例) OPENAI_API_KEYsk-... OPENAI_BASE_URLhttps://api.openai.com/v1 # 通义千问 (示例需根据实际SDK调整) DASHSCOPE_API_KEYsk-... # 其他服务的密钥...在代码中加载from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 import os openai_api_key os.getenv(OPENAI_API_KEY)3. 项目实战构建一个金融知识问答机器人我们将实现一个名为“FinBot”的机器人它结合了 RAG 和智能体技术。技术栈包括Qwen 作为 LLM、LangChain LangGraph 进行编排、Chroma 作为向量数据库、FastAPI 提供接口。3.1 项目设计与模块划分finbot-project/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用入口 │ ├── chains/ # 存放各种链 │ │ ├── __init__.py │ │ └── rag_chain.py # RAG 问答链 │ ├── graphs/ # 存放LangGraph图 │ │ ├── __init__.py │ │ └── agent_graph.py # 智能体工作流图 │ ├── memory/ # 记忆管理 │ │ ├── __init__.py │ │ └── conversation_memory.py │ ├── retriever/ # 检索器相关 │ │ ├── __init__.py │ │ └── vector_store.py # 向量库初始化与检索 │ └── config.py # 配置文件 ├── data/ # 存放原始知识文档 │ └── financial_regulations.pdf ├── scripts/ # 脚本如知识库初始化 │ └── init_vector_store.py ├── .env # 环境变量 ├── requirements.txt # 项目依赖 └── README.md3.2 核心实现步骤步骤一初始化向量知识库这是 RAG 的基石。我们编写脚本scripts/init_vector_store.py来将 PDF 文档处理并存入向量数据库。# scripts/init_vector_store.py import os from langchain_community.document_loaders import PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma def init_knowledge_base(pdf_path: str, persist_directory: str ./chroma_db): 加载PDF分割文本生成嵌入并持久化到Chroma数据库。 # 1. 加载文档 loader PyPDFLoader(pdf_path) documents loader.load() print(fLoaded {len(documents)} pages from PDF.) # 2. 分割文本 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, # 每个块的大小 chunk_overlap200, # 块之间的重叠保持上下文 separators[\n\n, \n, 。, , , , , , ] ) splits text_splitter.split_documents(documents) print(fSplit into {len(splits)} text chunks.) # 3. 初始化嵌入模型 # 这里使用OpenAI的嵌入模型实际项目中可替换为Qwen等模型的嵌入接口 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 4. 创建并持久化向量库 vectordb Chroma.from_documents( documentssplits, embeddingembeddings, persist_directorypersist_directory ) vectordb.persist() # 持久化到磁盘 print(fVector database initialized and persisted to {persist_directory}) return vectordb if __name__ __main__: # 假设你的PDF文件路径 pdf_path ../data/financial_regulations.pdf init_knowledge_base(pdf_path)运行此脚本后会在项目目录下生成chroma_db文件夹里面存储了向量数据。步骤二构建 RAG 问答链在app/chains/rag_chain.py中我们使用 LCEL 构建一个高效的 RAG 链。# app/chains/rag_chain.py from langchain_openai import ChatOpenAI from langchain.chains.combine_documents import create_stuff_documents_chain from langchain.chains import create_retrieval_chain from langchain_core.prompts import ChatPromptTemplate from app.retriever.vector_store import get_retriever def build_rag_chain(): 构建一个标准的 RAG 问答链。 # 1. 定义LLM (以Qwen为例使用与OpenAI兼容的接口) # 注意你需要一个支持Qwen且兼容OpenAI API格式的服务端点 llm ChatOpenAI( modelqwen-max, # 或具体模型名 base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, # 示例端点 api_keyos.getenv(DASHSCOPE_API_KEY), temperature0.1 # 金融问答要求精确温度调低 ) # 2. 定义系统提示词 system_prompt 你是一个专业的金融知识助手。请严格根据提供的上下文信息来回答问题。 如果上下文信息不足以回答问题请直接说“根据现有资料我无法回答这个问题”不要编造信息。 回答请简洁、准确、专业。 上下文{context} # 3. 使用LCEL定义提示词模板和链 prompt ChatPromptTemplate.from_messages([ (system, system_prompt), (human, {input}) ]) # 4. 创建“组合文档”的链 combine_docs_chain create_stuff_documents_chain(llm, prompt) # 5. 获取检索器 retriever get_retriever() # 这个函数需要你在 vector_store.py 中实现 # 6. 创建最终的检索链 rag_chain create_retrieval_chain(retriever, combine_docs_chain) return rag_chain # 在 vector_store.py 中实现 get_retriever # app/retriever/vector_store.py from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma def get_retriever(persist_directory: str ./chroma_db, k: int 4): embeddings OpenAIEmbeddings(modeltext-embedding-3-small) vectordb Chroma( persist_directorypersist_directory, embedding_functionembeddings ) # 检索器可以配置搜索类型如相似度搜索similarity_search或 MMR retriever vectordb.as_retriever( search_typesimilarity, search_kwargs{k: k} # 返回最相关的k个片段 ) return retriever这个链的工作流程是用户输入问题 - 检索器从向量库找到相关文档片段 - 将片段和问题一起填入提示词 - LLM 生成基于上下文的答案。步骤三使用 LangGraph 构建智能体工作流假设我们的机器人需要更复杂的能力先尝试用 RAG 回答如果用户不满意或问题涉及计算则调用一个“金融计算工具”。我们用 LangGraph 来实现这个有状态的工作流。# app/graphs/agent_graph.py from typing import TypedDict, Annotated, Literal import operator from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, AIMessage from app.chains.rag_chain import build_rag_chain # 1. 定义图的状态结构 class AgentState(TypedDict): messages: Annotated[list, operator.add] # 消息历史会自动追加 question: str # 原始问题 rag_answer: str | None # RAG环节的答案 final_answer: str | None # 最终答案 need_calculation: bool # 是否需要计算 # 2. 定义各个节点函数 def rag_node(state: AgentState) - dict: 调用RAG链回答问题。 print([Node] Calling RAG...) chain build_rag_chain() result chain.invoke({input: state[question]}) return {rag_answer: result[answer]} def calculation_node(state: AgentState) - dict: 模拟金融计算工具。 print([Node] Performing calculation...) # 这里可以集成真实的计算库如 numpy, pandas 或专业金融库 # 此处仅作演示 calculated_result f模拟计算根据问题{state[question]}计算结果为 XXX。 return {final_answer: calculated_result} def router(state: AgentState) - Literal[use_calculation, finalize]: 路由决策是否需要计算 # 这里可以设计更复杂的决策逻辑例如基于LLM判断或基于关键词 # 简单演示如果RAG答案包含“无法回答”或用户问题包含“计算”、“利率”等词则去计算 rag_answer state.get(rag_answer, ) question state[question].lower() if 无法回答 in rag_answer or any(word in question for word in [计算, 利率, 收益, 公式]): return use_calculation else: return finalize def finalize_node(state: AgentState) - dict: 最终节点确定答案。 print([Node] Finalizing answer...) final_answer state.get(rag_answer) or state.get(final_answer) or 未能生成答案。 # 将最终答案添加到消息历史 new_message AIMessage(contentfinal_answer) return {final_answer: final_answer, messages: [new_message]} # 3. 构建图 def create_agent_graph(): workflow StateGraph(AgentState) # 添加节点 workflow.add_node(rag, rag_node) workflow.add_node(calculation, calculation_node) workflow.add_node(finalize, finalize_node) # 设置入口点 workflow.set_entry_point(rag) # 添加条件边 workflow.add_conditional_edges( rag, router, # 路由函数决定下一个节点 { use_calculation: calculation, finalize: finalize } ) workflow.add_edge(calculation, finalize) workflow.add_edge(finalize, END) # 编译图 return workflow.compile() # 4. 使用图 if __name__ __main__: graph create_agent_graph() initial_state: AgentState { messages: [], question: 请计算一下年化利率5%投资10000元3年后的复利终值是多少, rag_answer: None, final_answer: None, need_calculation: False } result graph.invoke(initial_state) print(Final Answer:, result[final_answer])这个图定义了清晰的工作流RAG节点 - 路由判断 - (计算节点) - 最终节点。状态AgentState在整个流程中传递和更新。步骤四集成记忆与配置 RunnableConfig对于多轮对话需要管理历史消息。LangChain 提供了RunnableWithMessageHistory。关键是如何配置session_id。# app/memory/conversation_memory.py from langchain.memory import ChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory # 一个简单的内存存储生产环境应使用Redis、数据库等 store {} def get_session_history(session_id: str) - ChatMessageHistory: if session_id not in store: store[session_id] ChatMessageHistory() return store[session_id] # 假设我们有一个基础的链 base_chain (可以是上述的RAG链或智能体链) from app.chains.rag_chain import build_rag_chain base_chain build_rag_chain() # 包装成带历史记忆的链 conversational_chain RunnableWithMessageHistory( base_chain, get_session_history, # 获取历史消息的函数 input_messages_keyinput, # 输入中用户消息的键 history_messages_keychat_history, # 传递给链的历史消息键需提示词支持 ) # 使用链时需要通过 config 传入 session_id config {configurable: {session_id: user_123}} result conversational_chain.invoke( {input: 什么是货币政策}, configconfig ) print(result[answer]) # 第二轮对话历史消息会自动传入 result2 conversational_chain.invoke( {input: 它和财政政策有什么区别}, configconfig # 相同的 session_id ) print(result2[answer])关键点RunnableWithMessageHistory的config参数需要是一个字典其中configurable字段下的session_id用于标识不同的对话会话。生产环境中这个session_id可以来自 Web 请求的 Cookie 或用户 ID。步骤五使用 FastAPI 暴露服务将我们构建的链或图封装成 API。# app/main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from app.graphs.agent_graph import create_agent_graph from app.memory.conversation_memory import conversational_chain app FastAPI(titleFinBot API) class QueryRequest(BaseModel): question: str session_id: str default_session # 前端传递会话ID class QueryResponse(BaseModel): answer: str session_id: str app.post(/chat, response_modelQueryResponse) async def chat_with_agent(request: QueryRequest): 使用智能体图进行问答。 try: graph create_agent_graph() initial_state { messages: [], question: request.question, rag_answer: None, final_answer: None, need_calculation: False } result graph.invoke(initial_state) return QueryResponse(answerresult[final_answer], session_idrequest.session_id) except Exception as e: raise HTTPException(status_code500, detailfInternal server error: {str(e)}) app.post(/chat/rag, response_modelQueryResponse) async def chat_with_rag(request: QueryRequest): 使用带记忆的RAG链进行多轮对话。 try: config {configurable: {session_id: request.session_id}} result conversational_chain.invoke( {input: request.question}, configconfig ) return QueryResponse(answerresult[answer], session_idrequest.session_id) except Exception as e: raise HTTPException(status_code500, detailfInternal server error: {str(e)}) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)使用uvicorn app.main:app --reload启动服务即可通过/chat或/chat/rag端点进行交互。4. 关键问题排查与优化实践4.1 LangChain 如何处理上下文过长问题LLM 有上下文窗口限制。当检索到的文档片段总长度超过限制时直接拼接会报错。LangChain 提供了多种上下文压缩和重排策略。问题现象调用链时出现ContextLengthExceededError或类似令牌超限的错误。解决方案优化检索调整search_kwargs减少k返回的文档数量或使用MMR最大边际相关性搜索来提升多样性而非单纯数量。使用上下文压缩from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor from langchain_openai import ChatOpenAI llm ChatOpenAI(temperature0) compressor LLMChainExtractor.from_llm(llm) base_retriever get_retriever() # 你的基础检索器 compression_retriever ContextualCompressionRetriever( base_compressorcompressor, base_retrieverbase_retriever ) # 使用 compression_retriever 代替原来的 retrieverLLMChainExtractor会调用 LLM 来提取检索到的文档中与问题最相关的部分从而缩短上下文。文本分割策略优化RecursiveCharacterTextSplitter的chunk_size和chunk_overlap。对于金融、法律文本按句子或段落分割可能比按字符数更有效。Map-Reduce 方法对于极长的文档可以采用先分别总结各片段Map再总结所有摘要Reduce的方式但这会增加 LLM 调用次数和成本。4.2 记忆Memory不生效或混乱问题现象多轮对话中AI 忘记之前的对话内容或者不同用户的对话历史混在一起。排查步骤检查session_id确保每次调用RunnableWithMessageHistory时对于同一用户/会话传入的config中的session_id是稳定且唯一的。这是最常见的问题。检查提示词你的链的提示词模板中必须有一个位置用于接收历史消息。通常是chat_history变量。确保RunnableWithMessageHistory的history_messages_key参数与提示词中的变量名匹配。prompt ChatPromptTemplate.from_messages([ (system, 你是一个助手。), MessagesPlaceholder(variable_namechat_history), # 关键历史消息占位符 (human, {input}), ])检查存储后端如果使用自定义的get_session_history函数确保存储如字典、Redis是持久化的并且在多实例部署时是共享的例如使用 Redis 而不是内存字典。4.3 智能体Agent/Graph陷入循环或逻辑错误问题现象LangGraph 智能体在一个节点间无限循环或路由决策不符合预期。排查步骤启用调试在节点函数中增加print语句或使用LangSmith进行可视化跟踪查看状态 (state) 的变化和流转路径。检查条件边函数router函数的返回值必须严格匹配你定义的边条件如use_calculation,finalize。打印router函数的输入和输出进行验证。设置最大循环次数对于可能循环的图在StateGraph编译时设置checkpointer或在图定义中显式加入中断条件防止无限循环。from langgraph.graph import START, END workflow.add_edge(START, rag) # ... 其他边 workflow.add_edge(finalize, END) # 编译时可以配置检查点但更简单的是在节点逻辑中判断4.4 生产环境最佳实践清单配置外置将所有模型名称、API Base URL、温度参数、向量数据库连接信息等写入配置文件如config.py或config.yaml或环境变量不要硬编码。异常处理与降级在 API 调用层如 FastAPI 路由做好全局异常捕获。对于 LLM 或检索调用设置合理的超时和重试机制。考虑设计降级策略如 RAG 失败时返回通用回复或切换备用模型。日志与监控记录所有用户查询和 AI 响应注意脱敏。记录每次 LLM 调用的令牌使用量、耗时。使用LangSmith来跟踪链和图的执行细节它是调试 LangChain 应用的利器。性能优化嵌入缓存对频繁查询的文本片段缓存其嵌入向量避免重复计算。向量索引优化根据数据量选择合适的向量数据库Chroma, Pinecone, Weaviate 等并优化索引参数。异步调用如果业务允许使用ainvoke,abatch等异步接口提升吞吐。安全与权限输入过滤对用户输入进行基本的清理和过滤防止 Prompt 注入攻击。输出审查对 AI 生成的内容进行必要的安全审查特别是涉及金融建议时必须添加免责声明。API 密钥管理使用安全的密钥管理服务定期轮换密钥。5. 扩展方向与进阶技术结合本文的金融问答机器人是一个起点要使其更强大、更专业可以考虑以下扩展方向与 GraphRAG 结合传统的向量检索Vector RAG可能丢失文档间的复杂关系。GraphRAG 利用图数据库存储实体和关系能进行更复杂的推理查询例如“找出受政策A影响的所有公司并分析它们之间的竞争关系”。这需要将知识抽取成图结构。模型微调SFT/LoRA如果通用模型在专业金融术语、格式上表现不佳可以考虑使用领域文本对开源模型如 Qwen进行监督微调SFT或使用 LoRA 进行高效微调使其输出更符合金融文本风格。强化学习优化PPO/DPO使用人类反馈或成对偏好数据通过 PPO 或 DPO 等算法对模型进行对齐优化让它的回答更安全、更符合人类价值观。模型量化与蒸馏为了降低部署成本可以对微调后的模型进行量化如 GPTQ, AWQ或知识蒸馏在保持性能的同时减少模型体积和推理所需资源。构建技能Skill库将不同的能力如财报分析、风险计算、新闻摘要封装成独立的“技能”可以是 Chain 或 Graph由一个大智能体根据用户意图动态调度实现更模块化和可扩展的架构。通过深入理解 LangChain 和 LangGraph 的机制并妥善处理工程化中的各种细节你可以构建出既灵活又健壮的企业级大模型应用。记住清晰的架构设计、细致的错误处理和持续的性能监控与选择正确的算法同样重要。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度
30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度在实际 AI 应用开发中LangChain 已经从一个单纯的“链”式编排框架演变为一个包含 LangGraph、LangServe、LangSmith 等多个组件的生态系统。对于开发者而言理解 LangChain 的核心架构、掌握其最新版本如 0.1.x的用法并能够将其与 RAG、智能体Agent、微调等技术结合是构建企业级大模型应用的关键。本文将从一个资深开发者的视角带你深入 LangChain 的工程实践涵盖从核心概念、环境搭建、项目实现到生产级优化的完整链路。无论你是想快速入门 LangChain还是希望解决实际项目中遇到的上下文过长、记忆管理、智能体流程控制等问题都能在本文中找到可复现的解决方案和清晰的排查思路。1. 理解 LangChain 生态从链到图的演进LangChain 最初的核心是“链”Chain它将大语言模型LLM调用、工具使用、记忆等环节串联起来形成一个可预测的执行流程。但随着应用复杂度的提升尤其是需要循环、分支、状态管理等场景的出现单纯的链式结构显得力不从心。这正是 LangGraph 诞生的背景。1.1 LangChain 与 LangGraph 的核心区别简单来说LangChain 的Chain是线性或静态图的执行范式而 LangGraph 引入了有状态、可循环的图执行范式。LangChain Chain: 适用于顺序明确的管道式任务。例如一个典型的 RAG 链可能是检索器 - 格式化提示词 - 调用 LLM - 解析输出。这个流程是固定的输入决定输出没有内部状态循环。LangGraph: 将执行流程建模为一个有向图节点是处理单元可以是 Chain、函数、LLM 等边定义了节点间的流转条件。它维护一个共享的“状态”对象节点可以读取和修改这个状态并根据状态决定下一个执行的节点。这使得实现多轮对话、ReAct 智能体、带有工具循环调用的复杂工作流成为可能。下表对比了两者的关键差异特性LangChain ChainLangGraph执行模型线性/静态图有向图支持循环、分支状态管理隐式通过链的输入输出传递显式通过共享的State对象管理适用场景文档问答、文本总结、简单提取等确定性流程智能体、多轮对话、需要反复尝试或条件判断的复杂工作流控制流有限主要通过RunnableBranch等实现简单分支强大可基于状态任意定义节点跳转条件代码抽象LCEL(LangChain Expression Language) 声明式组合基于StateGraph定义节点和边对于大多数应用LCEL 和 Chain 已经足够。但当你需要构建一个能“思考”并决定下一步做什么的智能体时LangGraph 是更自然的选择。1.2 LangChain 最新版本架构概览以 LangChain 0.1.x 版本为例其核心架构可以理解为几个层次Schema 层: 定义了基础数据对象如Message(HumanMessage,AIMessage,SystemMessage)、Document、ChatGeneration等。这是所有组件交互的“语言”。Model I/O 层: 负责与各种 LLM、ChatModel、EmbeddingModel 交互。通过ChatOpenAI、ChatQwen等类进行抽象。Retrieval 层 (RAG核心): 包含文本分割器、向量存储接口、检索器等用于构建和查询知识库。Chains 层: 基于 LCEL 将其他组件组合成链。Runnable协议是核心一切皆可 Runnable。Agents 层: 构建智能体的高层抽象内部可能由 Chain 或 LangGraph 实现。Memory 层: 管理对话或应用的状态如ConversationBufferMemory。LangGraph: 作为独立库 (langgraph)提供基于图的编排能力。LangServe: 用于将 Chain 或 Graph 快速部署为 REST API。LangSmith: 用于调试、测试、监控和评估链与智能体的平台。理解这个分层有助于在遇到问题时快速定位是哪个环节的配置或调用出了错。2. 环境准备与核心依赖配置开始一个 LangChain 项目前环境的隔离和依赖版本的锁定至关重要不同版本间的 API 差异可能导致代码无法运行。2.1 创建并配置 Python 虚拟环境推荐使用conda或venv管理环境。# 使用 conda conda create -n langchain-demo python3.10 conda activate langchain-demo # 或使用 venv python -m venv venv # Windows venv\Scripts\activate # Linux/Mac source venv/bin/activate2.2 安装核心依赖根据项目需求选择安装。以下是一个面向 RAG 和智能体开发的常用依赖列表使用pip安装。# 核心LangChain pip install langchain langchain-community # LangGraph 用于复杂工作流 pip install langgraph # LangChain 对常见大模型的支持包 (按需安装) pip install langchain-openai # 用于OpenAI, Azure OpenAI # 假设使用通义千问可能需要对应的SDK这里以openai兼容接口为例 # pip install dashscope # 阿里云DashScope SDK # 向量数据库客户端 (例如Chroma) pip install chromadb # 文本嵌入模型 (例如使用OpenAI的text-embedding-3-small) pip install langchain-openai # Web框架用于部署API pip install fastapi uvicorn # 环境变量管理 pip install python-dotenv重要提示langchain是一个元包它会安装一系列核心子包。对于生产环境更推荐显式安装你真正需要的子包如langchain-core,langchain-openai等以减少依赖冲突和包体积。2.3 配置 API 密钥与环境变量永远不要将 API 密钥硬编码在代码中。使用.env文件和环境变量管理。在项目根目录创建.env文件# OpenAI (示例) OPENAI_API_KEYsk-... OPENAI_BASE_URLhttps://api.openai.com/v1 # 通义千问 (示例需根据实际SDK调整) DASHSCOPE_API_KEYsk-... # 其他服务的密钥...在代码中加载from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 import os openai_api_key os.getenv(OPENAI_API_KEY)3. 项目实战构建一个金融知识问答机器人我们将实现一个名为“FinBot”的机器人它结合了 RAG 和智能体技术。技术栈包括Qwen 作为 LLM、LangChain LangGraph 进行编排、Chroma 作为向量数据库、FastAPI 提供接口。3.1 项目设计与模块划分finbot-project/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用入口 │ ├── chains/ # 存放各种链 │ │ ├── __init__.py │ │ └── rag_chain.py # RAG 问答链 │ ├── graphs/ # 存放LangGraph图 │ │ ├── __init__.py │ │ └── agent_graph.py # 智能体工作流图 │ ├── memory/ # 记忆管理 │ │ ├── __init__.py │ │ └── conversation_memory.py │ ├── retriever/ # 检索器相关 │ │ ├── __init__.py │ │ └── vector_store.py # 向量库初始化与检索 │ └── config.py # 配置文件 ├── data/ # 存放原始知识文档 │ └── financial_regulations.pdf ├── scripts/ # 脚本如知识库初始化 │ └── init_vector_store.py ├── .env # 环境变量 ├── requirements.txt # 项目依赖 └── README.md3.2 核心实现步骤步骤一初始化向量知识库这是 RAG 的基石。我们编写脚本scripts/init_vector_store.py来将 PDF 文档处理并存入向量数据库。# scripts/init_vector_store.py import os from langchain_community.document_loaders import PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma def init_knowledge_base(pdf_path: str, persist_directory: str ./chroma_db): 加载PDF分割文本生成嵌入并持久化到Chroma数据库。 # 1. 加载文档 loader PyPDFLoader(pdf_path) documents loader.load() print(fLoaded {len(documents)} pages from PDF.) # 2. 分割文本 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, # 每个块的大小 chunk_overlap200, # 块之间的重叠保持上下文 separators[\n\n, \n, 。, , , , , , ] ) splits text_splitter.split_documents(documents) print(fSplit into {len(splits)} text chunks.) # 3. 初始化嵌入模型 # 这里使用OpenAI的嵌入模型实际项目中可替换为Qwen等模型的嵌入接口 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 4. 创建并持久化向量库 vectordb Chroma.from_documents( documentssplits, embeddingembeddings, persist_directorypersist_directory ) vectordb.persist() # 持久化到磁盘 print(fVector database initialized and persisted to {persist_directory}) return vectordb if __name__ __main__: # 假设你的PDF文件路径 pdf_path ../data/financial_regulations.pdf init_knowledge_base(pdf_path)运行此脚本后会在项目目录下生成chroma_db文件夹里面存储了向量数据。步骤二构建 RAG 问答链在app/chains/rag_chain.py中我们使用 LCEL 构建一个高效的 RAG 链。# app/chains/rag_chain.py from langchain_openai import ChatOpenAI from langchain.chains.combine_documents import create_stuff_documents_chain from langchain.chains import create_retrieval_chain from langchain_core.prompts import ChatPromptTemplate from app.retriever.vector_store import get_retriever def build_rag_chain(): 构建一个标准的 RAG 问答链。 # 1. 定义LLM (以Qwen为例使用与OpenAI兼容的接口) # 注意你需要一个支持Qwen且兼容OpenAI API格式的服务端点 llm ChatOpenAI( modelqwen-max, # 或具体模型名 base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, # 示例端点 api_keyos.getenv(DASHSCOPE_API_KEY), temperature0.1 # 金融问答要求精确温度调低 ) # 2. 定义系统提示词 system_prompt 你是一个专业的金融知识助手。请严格根据提供的上下文信息来回答问题。 如果上下文信息不足以回答问题请直接说“根据现有资料我无法回答这个问题”不要编造信息。 回答请简洁、准确、专业。 上下文{context} # 3. 使用LCEL定义提示词模板和链 prompt ChatPromptTemplate.from_messages([ (system, system_prompt), (human, {input}) ]) # 4. 创建“组合文档”的链 combine_docs_chain create_stuff_documents_chain(llm, prompt) # 5. 获取检索器 retriever get_retriever() # 这个函数需要你在 vector_store.py 中实现 # 6. 创建最终的检索链 rag_chain create_retrieval_chain(retriever, combine_docs_chain) return rag_chain # 在 vector_store.py 中实现 get_retriever # app/retriever/vector_store.py from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma def get_retriever(persist_directory: str ./chroma_db, k: int 4): embeddings OpenAIEmbeddings(modeltext-embedding-3-small) vectordb Chroma( persist_directorypersist_directory, embedding_functionembeddings ) # 检索器可以配置搜索类型如相似度搜索similarity_search或 MMR retriever vectordb.as_retriever( search_typesimilarity, search_kwargs{k: k} # 返回最相关的k个片段 ) return retriever这个链的工作流程是用户输入问题 - 检索器从向量库找到相关文档片段 - 将片段和问题一起填入提示词 - LLM 生成基于上下文的答案。步骤三使用 LangGraph 构建智能体工作流假设我们的机器人需要更复杂的能力先尝试用 RAG 回答如果用户不满意或问题涉及计算则调用一个“金融计算工具”。我们用 LangGraph 来实现这个有状态的工作流。# app/graphs/agent_graph.py from typing import TypedDict, Annotated, Literal import operator from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, AIMessage from app.chains.rag_chain import build_rag_chain # 1. 定义图的状态结构 class AgentState(TypedDict): messages: Annotated[list, operator.add] # 消息历史会自动追加 question: str # 原始问题 rag_answer: str | None # RAG环节的答案 final_answer: str | None # 最终答案 need_calculation: bool # 是否需要计算 # 2. 定义各个节点函数 def rag_node(state: AgentState) - dict: 调用RAG链回答问题。 print([Node] Calling RAG...) chain build_rag_chain() result chain.invoke({input: state[question]}) return {rag_answer: result[answer]} def calculation_node(state: AgentState) - dict: 模拟金融计算工具。 print([Node] Performing calculation...) # 这里可以集成真实的计算库如 numpy, pandas 或专业金融库 # 此处仅作演示 calculated_result f模拟计算根据问题{state[question]}计算结果为 XXX。 return {final_answer: calculated_result} def router(state: AgentState) - Literal[use_calculation, finalize]: 路由决策是否需要计算 # 这里可以设计更复杂的决策逻辑例如基于LLM判断或基于关键词 # 简单演示如果RAG答案包含“无法回答”或用户问题包含“计算”、“利率”等词则去计算 rag_answer state.get(rag_answer, ) question state[question].lower() if 无法回答 in rag_answer or any(word in question for word in [计算, 利率, 收益, 公式]): return use_calculation else: return finalize def finalize_node(state: AgentState) - dict: 最终节点确定答案。 print([Node] Finalizing answer...) final_answer state.get(rag_answer) or state.get(final_answer) or 未能生成答案。 # 将最终答案添加到消息历史 new_message AIMessage(contentfinal_answer) return {final_answer: final_answer, messages: [new_message]} # 3. 构建图 def create_agent_graph(): workflow StateGraph(AgentState) # 添加节点 workflow.add_node(rag, rag_node) workflow.add_node(calculation, calculation_node) workflow.add_node(finalize, finalize_node) # 设置入口点 workflow.set_entry_point(rag) # 添加条件边 workflow.add_conditional_edges( rag, router, # 路由函数决定下一个节点 { use_calculation: calculation, finalize: finalize } ) workflow.add_edge(calculation, finalize) workflow.add_edge(finalize, END) # 编译图 return workflow.compile() # 4. 使用图 if __name__ __main__: graph create_agent_graph() initial_state: AgentState { messages: [], question: 请计算一下年化利率5%投资10000元3年后的复利终值是多少, rag_answer: None, final_answer: None, need_calculation: False } result graph.invoke(initial_state) print(Final Answer:, result[final_answer])这个图定义了清晰的工作流RAG节点 - 路由判断 - (计算节点) - 最终节点。状态AgentState在整个流程中传递和更新。步骤四集成记忆与配置 RunnableConfig对于多轮对话需要管理历史消息。LangChain 提供了RunnableWithMessageHistory。关键是如何配置session_id。# app/memory/conversation_memory.py from langchain.memory import ChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory # 一个简单的内存存储生产环境应使用Redis、数据库等 store {} def get_session_history(session_id: str) - ChatMessageHistory: if session_id not in store: store[session_id] ChatMessageHistory() return store[session_id] # 假设我们有一个基础的链 base_chain (可以是上述的RAG链或智能体链) from app.chains.rag_chain import build_rag_chain base_chain build_rag_chain() # 包装成带历史记忆的链 conversational_chain RunnableWithMessageHistory( base_chain, get_session_history, # 获取历史消息的函数 input_messages_keyinput, # 输入中用户消息的键 history_messages_keychat_history, # 传递给链的历史消息键需提示词支持 ) # 使用链时需要通过 config 传入 session_id config {configurable: {session_id: user_123}} result conversational_chain.invoke( {input: 什么是货币政策}, configconfig ) print(result[answer]) # 第二轮对话历史消息会自动传入 result2 conversational_chain.invoke( {input: 它和财政政策有什么区别}, configconfig # 相同的 session_id ) print(result2[answer])关键点RunnableWithMessageHistory的config参数需要是一个字典其中configurable字段下的session_id用于标识不同的对话会话。生产环境中这个session_id可以来自 Web 请求的 Cookie 或用户 ID。步骤五使用 FastAPI 暴露服务将我们构建的链或图封装成 API。# app/main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from app.graphs.agent_graph import create_agent_graph from app.memory.conversation_memory import conversational_chain app FastAPI(titleFinBot API) class QueryRequest(BaseModel): question: str session_id: str default_session # 前端传递会话ID class QueryResponse(BaseModel): answer: str session_id: str app.post(/chat, response_modelQueryResponse) async def chat_with_agent(request: QueryRequest): 使用智能体图进行问答。 try: graph create_agent_graph() initial_state { messages: [], question: request.question, rag_answer: None, final_answer: None, need_calculation: False } result graph.invoke(initial_state) return QueryResponse(answerresult[final_answer], session_idrequest.session_id) except Exception as e: raise HTTPException(status_code500, detailfInternal server error: {str(e)}) app.post(/chat/rag, response_modelQueryResponse) async def chat_with_rag(request: QueryRequest): 使用带记忆的RAG链进行多轮对话。 try: config {configurable: {session_id: request.session_id}} result conversational_chain.invoke( {input: request.question}, configconfig ) return QueryResponse(answerresult[answer], session_idrequest.session_id) except Exception as e: raise HTTPException(status_code500, detailfInternal server error: {str(e)}) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)使用uvicorn app.main:app --reload启动服务即可通过/chat或/chat/rag端点进行交互。4. 关键问题排查与优化实践4.1 LangChain 如何处理上下文过长问题LLM 有上下文窗口限制。当检索到的文档片段总长度超过限制时直接拼接会报错。LangChain 提供了多种上下文压缩和重排策略。问题现象调用链时出现ContextLengthExceededError或类似令牌超限的错误。解决方案优化检索调整search_kwargs减少k返回的文档数量或使用MMR最大边际相关性搜索来提升多样性而非单纯数量。使用上下文压缩from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor from langchain_openai import ChatOpenAI llm ChatOpenAI(temperature0) compressor LLMChainExtractor.from_llm(llm) base_retriever get_retriever() # 你的基础检索器 compression_retriever ContextualCompressionRetriever( base_compressorcompressor, base_retrieverbase_retriever ) # 使用 compression_retriever 代替原来的 retrieverLLMChainExtractor会调用 LLM 来提取检索到的文档中与问题最相关的部分从而缩短上下文。文本分割策略优化RecursiveCharacterTextSplitter的chunk_size和chunk_overlap。对于金融、法律文本按句子或段落分割可能比按字符数更有效。Map-Reduce 方法对于极长的文档可以采用先分别总结各片段Map再总结所有摘要Reduce的方式但这会增加 LLM 调用次数和成本。4.2 记忆Memory不生效或混乱问题现象多轮对话中AI 忘记之前的对话内容或者不同用户的对话历史混在一起。排查步骤检查session_id确保每次调用RunnableWithMessageHistory时对于同一用户/会话传入的config中的session_id是稳定且唯一的。这是最常见的问题。检查提示词你的链的提示词模板中必须有一个位置用于接收历史消息。通常是chat_history变量。确保RunnableWithMessageHistory的history_messages_key参数与提示词中的变量名匹配。prompt ChatPromptTemplate.from_messages([ (system, 你是一个助手。), MessagesPlaceholder(variable_namechat_history), # 关键历史消息占位符 (human, {input}), ])检查存储后端如果使用自定义的get_session_history函数确保存储如字典、Redis是持久化的并且在多实例部署时是共享的例如使用 Redis 而不是内存字典。4.3 智能体Agent/Graph陷入循环或逻辑错误问题现象LangGraph 智能体在一个节点间无限循环或路由决策不符合预期。排查步骤启用调试在节点函数中增加print语句或使用LangSmith进行可视化跟踪查看状态 (state) 的变化和流转路径。检查条件边函数router函数的返回值必须严格匹配你定义的边条件如use_calculation,finalize。打印router函数的输入和输出进行验证。设置最大循环次数对于可能循环的图在StateGraph编译时设置checkpointer或在图定义中显式加入中断条件防止无限循环。from langgraph.graph import START, END workflow.add_edge(START, rag) # ... 其他边 workflow.add_edge(finalize, END) # 编译时可以配置检查点但更简单的是在节点逻辑中判断4.4 生产环境最佳实践清单配置外置将所有模型名称、API Base URL、温度参数、向量数据库连接信息等写入配置文件如config.py或config.yaml或环境变量不要硬编码。异常处理与降级在 API 调用层如 FastAPI 路由做好全局异常捕获。对于 LLM 或检索调用设置合理的超时和重试机制。考虑设计降级策略如 RAG 失败时返回通用回复或切换备用模型。日志与监控记录所有用户查询和 AI 响应注意脱敏。记录每次 LLM 调用的令牌使用量、耗时。使用LangSmith来跟踪链和图的执行细节它是调试 LangChain 应用的利器。性能优化嵌入缓存对频繁查询的文本片段缓存其嵌入向量避免重复计算。向量索引优化根据数据量选择合适的向量数据库Chroma, Pinecone, Weaviate 等并优化索引参数。异步调用如果业务允许使用ainvoke,abatch等异步接口提升吞吐。安全与权限输入过滤对用户输入进行基本的清理和过滤防止 Prompt 注入攻击。输出审查对 AI 生成的内容进行必要的安全审查特别是涉及金融建议时必须添加免责声明。API 密钥管理使用安全的密钥管理服务定期轮换密钥。5. 扩展方向与进阶技术结合本文的金融问答机器人是一个起点要使其更强大、更专业可以考虑以下扩展方向与 GraphRAG 结合传统的向量检索Vector RAG可能丢失文档间的复杂关系。GraphRAG 利用图数据库存储实体和关系能进行更复杂的推理查询例如“找出受政策A影响的所有公司并分析它们之间的竞争关系”。这需要将知识抽取成图结构。模型微调SFT/LoRA如果通用模型在专业金融术语、格式上表现不佳可以考虑使用领域文本对开源模型如 Qwen进行监督微调SFT或使用 LoRA 进行高效微调使其输出更符合金融文本风格。强化学习优化PPO/DPO使用人类反馈或成对偏好数据通过 PPO 或 DPO 等算法对模型进行对齐优化让它的回答更安全、更符合人类价值观。模型量化与蒸馏为了降低部署成本可以对微调后的模型进行量化如 GPTQ, AWQ或知识蒸馏在保持性能的同时减少模型体积和推理所需资源。构建技能Skill库将不同的能力如财报分析、风险计算、新闻摘要封装成独立的“技能”可以是 Chain 或 Graph由一个大智能体根据用户意图动态调度实现更模块化和可扩展的架构。通过深入理解 LangChain 和 LangGraph 的机制并妥善处理工程化中的各种细节你可以构建出既灵活又健壮的企业级大模型应用。记住清晰的架构设计、细致的错误处理和持续的性能监控与选择正确的算法同样重要。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度
![[C++]内存管理：串顺序存储的内存回收](http://pic.xiahunao.cn/yaotu/[C++]内存管理：串顺序存储的内存回收)
