ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

LangChain 1.3实战指南:从Prompt工程到RAG系统完整开发流程

LangChain 1.3实战指南:从Prompt工程到RAG系统完整开发流程 最近在AI应用开发中很多开发者反馈LangChain框架版本更新快从0.x到1.x变化较大网上资料分散不成体系。本文基于最新LangChain 1.3版本完整梳理从基础概念到项目落地的全流程包含Prompt工程、Agent开发、RAG系统构建等核心模块提供可运行的代码示例和避坑指南。无论你是刚接触AI应用开发的新手还是有一定经验想要系统掌握LangChain的开发者都能从本文获得实用价值。我们将通过具体的项目案例演示如何将大模型能力整合到实际业务场景中。1. LangChain核心概念与架构解析1.1 什么是LangChain及其发展历程LangChain是一个用于开发由语言模型驱动的应用程序的框架。它简化了AI应用的开发流程让开发者能够更轻松地构建复杂的AI系统。从最初的0.x版本到现在的1.x版本LangChain在架构设计和API稳定性方面都有了显著提升。LangChain 1.3版本的主要改进包括更清晰的模块化设计更好的类型提示和错误处理增强的链式操作支持改进的Agent开发体验更稳定的RAG实现1.2 LangChain核心组件架构LangChain的核心架构包含以下几个关键组件模型层Models支持多种大语言模型包括OpenAI、通义千问、文心一言等主流模型。提示模板Prompt Templates用于管理和优化提示词提高与大模型交互的效果。链Chains将多个组件组合成完整的工作流实现复杂的AI应用逻辑。代理Agents让大模型能够使用工具和执行动作实现更智能的交互。内存Memory在多次交互中保持状态实现对话的连续性。索引Indexes用于文档加载、分割和检索是RAG系统的核心。1.3 LangChain与其他框架的区别很多开发者困惑于LangChain与LangGraph等框架的区别。LangChain主要专注于构建基于大模型的应用程序而LangGraph更侧重于构建有状态的、多步骤的工作流。在实际项目中两者可以结合使用LangChain提供基础组件LangGraph处理复杂的工作流逻辑。2. 环境准备与版本配置2.1 系统环境要求在开始LangChain项目前需要确保开发环境满足以下要求Python 3.8或更高版本pip包管理工具虚拟环境推荐使用venv或conda2.2 LangChain版本安装与兼容性LangChain 1.3版本的安装命令如下# 创建虚拟环境 python -m venv langchain_env source langchain_env/bin/activate # Linux/Mac # 或 langchain_env\Scripts\activate # Windows # 安装LangChain核心包 pip install langchain1.3.11 # 安装社区扩展包 pip install langchain-community0.3.9 # 安装文本分割工具 pip install langchain-text-splitters0.3.9 # 安装其他可能需要的依赖 pip install openai tiktoken版本兼容性说明LangChain 1.3.11版本建议搭配langchain-community 0.3.9版本这两个版本在API兼容性和功能完整性方面有较好的配合。如果遇到版本冲突可以尝试使用pip的依赖解析功能自动解决。2.3 开发工具配置推荐使用VS Code或PyCharm作为开发环境安装Python扩展和必要的调试工具。对于大型项目建议配置代码格式化工具如black和类型检查工具如mypy。3. Prompt提示词工程实战3.1 Prompt设计基础原则Prompt工程是与大模型有效交互的关键。一个好的Prompt应该包含明确的指令清楚地告诉模型需要做什么。充足的上下文提供完成任务所需的背景信息。适当的格式要求指定输出的格式和结构。示例演示通过few-shot learning提供参考示例。3.2 LangChain Prompt模板使用LangChain提供了强大的Prompt模板功能下面是一个完整的示例from langchain.prompts import PromptTemplate from langchain_openai import ChatOpenAI import os # 设置API密钥 os.environ[OPENAI_API_KEY] your-api-key # 创建提示词模板 template 你是一个专业的{role}请根据以下要求回答问题 问题{question} 背景信息{context} 要求 1. 回答要专业、准确 2. 如果信息不足请明确说明 3. 使用{language}语言回答 4. 格式要求{format} 请开始回答 prompt_template PromptTemplate( input_variables[role, question, context, language, format], templatetemplate ) # 使用模板 filled_prompt prompt_template.format( role技术专家, questionLangChain的主要功能是什么, contextLangChain是一个用于开发由语言模型驱动的应用程序的框架, language中文, format分点说明 ) print(filled_prompt) # 调用模型 model ChatOpenAI(modelgpt-3.5-turbo) response model.invoke(filled_prompt) print(response.content)3.3 高级Prompt技巧思维链Chain-of-ThoughtPromptingcot_template 请逐步推理解决以下问题 问题{question} 思考步骤 1. 首先分析问题的核心要求 2. 然后拆解问题的关键要素 3. 接着制定解决方案 4. 最后给出完整答案 请开始 少样本学习Few-Shot Learningfew_shot_template 示例1 输入如何学习Python编程 输出建议从基础语法开始然后学习常用库最后做项目实践。 示例2 输入{question} 输出 prompt PromptTemplate.from_template(few_shot_template)3.4 常见Prompt错误与调试在实际使用中经常会遇到Prompt相关的问题。以下是一些常见错误和解决方案错误1提示词过于模糊# 错误示例 bad_prompt 告诉我关于AI的事情 # 正确示例 good_prompt 请以技术专家的身份用通俗易懂的方式解释人工智能的基本概念 包括机器学习、深度学习和神经网络的区别字数控制在500字以内。 错误2格式要求不明确# 改进后的模板 structured_template 请按照以下格式回答问题 【问题分析】 {analysis} 【核心要点】 1. {point1} 2. {point2} 3. {point3} 【总结】 {summary} 问题{question} 4. Agent代理系统开发实战4.1 Agent核心概念与类型Agent是LangChain中最强大的功能之一它让大模型能够使用工具和执行动作。主要的Agent类型包括Zero-shot ReAct Agent不需要示例就能根据工具描述决定行动。Conversational Agent专为对话场景设计支持多轮交互。Self-ask with Search Agent能够自主提出问题并搜索答案。Structured Chat Agent支持结构化输出的对话Agent。4.2 基础Agent开发示例下面是一个完整的Agent开发示例from langchain.agents import AgentType, initialize_agent from langchain.tools import Tool from langchain_openai import ChatOpenAI import math import datetime # 定义自定义工具 def calculate_square_root(n): 计算数字的平方根 return math.sqrt(n) def get_current_time(): 获取当前时间 return datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) # 创建工具列表 tools [ Tool( nameSquareRootCalculator, funccalculate_square_root, description用于计算数字的平方根。输入应该是一个数字。 ), Tool( nameTimeGetter, funcget_current_time, description用于获取当前时间。不需要输入参数。 ) ] # 初始化模型和Agent llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) agent initialize_agent( toolstools, llmllm, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verboseTrue ) # 运行Agent result agent.run(请计算16的平方根并告诉我当前时间) print(result)4.3 高级Agent功能多工具协同工作from langchain.agents import Tool, AgentExecutor from langchain.agents.format_scratchpad import format_log_to_str from langchain.agents.output_parsers import ReActSingleInputOutputParser from langchain.tools.render import render_text_description # 定义更复杂的工具集 def weather_check(city): 模拟天气查询功能 # 实际项目中这里会调用天气API return f{city}的天气晴25℃ def currency_converter(amount, from_currency, to_currency): 货币转换工具 # 简化示例实际需要调用汇率API rates {USD: 1, CNY: 7.2, EUR: 0.92} result amount * rates[to_currency] / rates[from_currency] return f{amount} {from_currency} {result:.2f} {to_currency} tools [ Tool(nameWeather, funcweather_check, description查询城市天气。输入格式城市名称), Tool(nameCurrencyConverter, funccurrency_converter, description货币转换。输入格式金额 原货币 目标货币如100 USD CNY) ] # 构建高级Agent prompt 请使用可用工具回答用户问题。你可以使用的工具有 {tools} 问题{input} 请按照以下格式思考 Thought: 我需要思考如何解决问题 Action: 要使用的工具名称 Action Input: 工具的输入 Observation: 工具返回的结果 ...这个循环可以重复 最终答案最终回答 agent_executor AgentExecutor.from_agent_and_tools( agentagent, toolstools, verboseTrue, max_iterations5 )4.4 Agent错误处理与优化在实际开发中Agent可能会遇到各种问题需要做好错误处理from langchain.agents import AgentType from langchain.callbacks import BaseCallbackHandler class AgentCallbackHandler(BaseCallbackHandler): def on_agent_action(self, action, **kwargs): print(fAgent执行动作: {action}) def on_agent_finish(self, finish, **kwargs): print(fAgent完成: {finish}) def on_tool_error(self, error, **kwargs): print(f工具执行错误: {error}) # 带有错误处理的Agent配置 safe_agent initialize_agent( toolstools, llmllm, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verboseTrue, handle_parsing_errorsTrue, max_iterations3, early_stopping_methodgenerate, callbacks[AgentCallbackHandler()] )5. RAG知识库系统构建5.1 RAG技术原理与架构RAGRetrieval-Augmented Generation检索增强生成技术通过结合检索器和生成器让大模型能够访问外部知识库。其核心流程包括文档加载与处理从各种来源加载文档文本分割将大文档分割成小块向量化将文本转换为向量表示检索根据问题检索相关文档片段生成基于检索结果生成答案5.2 完整RAG系统实现下面是一个企业知识库RAG系统的完整实现from langchain_community.document_loaders import TextLoader, PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_community.vectorstores import Chroma from langchain.chains import RetrievalQA import os class RAGSystem: def __init__(self, persist_directory./chroma_db): self.persist_directory persist_directory self.embeddings OpenAIEmbeddings() self.llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) self.vectorstore None def load_documents(self, file_paths): 加载文档 documents [] for file_path in file_paths: if file_path.endswith(.pdf): loader PyPDFLoader(file_path) else: loader TextLoader(file_path, encodingutf-8) documents.extend(loader.load()) return documents def split_documents(self, documents, chunk_size1000, chunk_overlap200): 分割文档 text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, length_functionlen ) return text_splitter.split_documents(documents) def create_vectorstore(self, documents): 创建向量数据库 self.vectorstore Chroma.from_documents( documentsdocuments, embeddingself.embeddings, persist_directoryself.persist_directory ) return self.vectorstore def setup_qa_chain(self): 设置QA链 if not self.vectorstore: raise ValueError(请先创建向量数据库) retriever self.vectorstore.as_retriever( search_typesimilarity, search_kwargs{k: 3} ) self.qa_chain RetrievalQA.from_chain_type( llmself.llm, chain_typestuff, retrieverretriever, return_source_documentsTrue ) return self.qa_chain def query(self, question): 查询知识库 if not hasattr(self, qa_chain): self.setup_qa_chain() result self.qa_chain.invoke({query: question}) return result # 使用示例 if __name__ __main__: rag_system RAGSystem() # 加载文档示例文件路径实际使用时需要替换 documents rag_system.load_documents([ company_handbook.txt, technical_docs.pdf ]) # 分割文档 split_docs rag_system.split_documents(documents) # 创建向量数据库 rag_system.create_vectorstore(split_docs) # 查询 result rag_system.query(公司的请假政策是什么) print(答案:, result[result]) print(来源文档:, result[source_documents])5.3 RAG系统优化策略多路检索优化from langchain.retrievers import BM25Retriever, EnsembleRetriever from langchain_community.vectorstores import FAISS def create_ensemble_retriever(documents): 创建混合检索器 # 向量检索器 vectorstore FAISS.from_documents(documents, OpenAIEmbeddings()) vector_retriever vectorstore.as_retriever(search_kwargs{k: 3}) # 关键词检索器 bm25_retriever BM25Retriever.from_documents(documents) bm25_retriever.k 3 # 混合检索器 ensemble_retriever EnsembleRetriever( retrievers[vector_retriever, bm25_retriever], weights[0.5, 0.5] ) return ensemble_retriever查询重写与扩展from langchain.chains import LLMChain from langchain.prompts import PromptTemplate query_expansion_template 原始问题{original_query} 请将这个问题重写为3个不同角度的相关问题以便更好地检索相关信息 1. 角度一具体化 2. 角度二扩展背景 3. 角度三相关概念 请直接输出重写后的问题每个问题一行。 def expand_query(original_query, llm): 查询扩展 prompt PromptTemplate.from_template(query_expansion_template) chain LLMChain(llmllm, promptprompt) expanded_queries chain.run(original_queryoriginal_query) return expanded_queries.split(\n)6. 综合项目实战智能技术客服系统6.1 项目需求分析我们将构建一个智能技术客服系统集成以下功能基于企业知识库的智能问答多轮对话支持工具调用能力如查询订单、技术文档检索对话历史管理6.2 系统架构设计from langchain.memory import ConversationBufferWindowMemory from langchain.chains import ConversationalRetrievalChain from langchain.agents import AgentType, initialize_agent from datetime import datetime import json class TechnicalSupportSystem: def __init__(self, knowledge_base_path): self.knowledge_base_path knowledge_base_path self.memory ConversationBufferWindowMemory( memory_keychat_history, k5, return_messagesTrue ) self.setup_system() def setup_system(self): 初始化系统组件 self.llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.2) # 知识库检索链 self.qa_chain self.setup_knowledge_qa() # 工具定义 self.tools self.setup_tools() # 对话Agent self.conversational_agent initialize_agent( toolsself.tools, llmself.llm, agentAgentType.CONVERSATIONAL_REACT_DESCRIPTION, memoryself.memory, verboseTrue, handle_parsing_errorsTrue ) def setup_knowledge_qa(self): 设置知识库QA系统 # 这里简化实现实际项目中需要加载真实知识库 from langchain.schema import Document sample_docs [ Document(page_content产品A的技术规格支持Python 3.8内存要求8GB以上), Document(page_content常见问题1安装失败请检查网络连接和权限设置), Document(page_contentAPI使用指南先调用初始化接口再执行主要功能) ] vectorstore Chroma.from_documents(sample_docs, OpenAIEmbeddings()) retriever vectorstore.as_retriever() return RetrievalQA.from_chain_type( llmself.llm, chain_typestuff, retrieverretriever ) def setup_tools(self): 设置工具集 def search_knowledge_base(query): 知识库搜索工具 result self.qa_chain.run(query) return f知识库检索结果{result} def get_system_status(): 系统状态检查工具 return f系统状态正常当前时间{datetime.now()} return [ Tool( nameKnowledgeSearch, funcsearch_knowledge_base, description搜索技术知识库输入技术问题 ), Tool( nameSystemStatus, funcget_system_status, description检查系统状态无需输入 ) ] def process_query(self, user_input): 处理用户查询 try: # 首先尝试用知识库回答技术问题 if any(keyword in user_input.lower() for keyword in [如何, 怎么, 问题, 错误, 安装, 配置]): knowledge_result self.qa_chain.run(user_input) return knowledge_result else: # 其他问题使用对话Agent result self.conversational_agent.run(user_input) return result except Exception as e: return f处理查询时出现错误{str(e)}请重新表述您的问题。 # 使用示例 support_system TechnicalSupportSystem(knowledge_base) response support_system.process_query(产品A的技术要求是什么) print(response)6.3 系统测试与优化测试用例设计def test_support_system(): 测试客服系统 test_cases [ 产品A的技术规格要求是什么, 系统当前状态如何, 安装过程中出现网络错误怎么办, API应该如何调用 ] system TechnicalSupportSystem(test_kb) for i, test_case in enumerate(test_cases, 1): print(f\n测试用例 {i}: {test_case}) response system.process_query(test_case) print(f响应: {response}) print(- * 50) # 运行测试 test_support_system()7. 常见问题与解决方案7.1 LangChain版本兼容性问题问题1版本冲突导致导入错误错误信息ImportError: cannot import name XXX from langchain 解决方案检查版本兼容性使用推荐的版本组合# 解决方案 pip uninstall langchain langchain-community pip install langchain1.3.11 langchain-community0.3.9问题2API变更导致的代码错误# 旧版本代码已过时 from langchain.llms import OpenAI # 新版本正确写法 from langchain_openai import OpenAI7.2 Prompt工程常见错误问题Prompt输出验证失败错误prompt outputs failed validation: checkpointloadersimple: - value not in list 解决方案检查Prompt模板的输入变量是否匹配确保所有变量都有值# 错误示例 template Hello {name}, your score is {score} prompt PromptTemplate(templatetemplate, input_variables[name]) # 缺少score变量 # 正确示例 prompt PromptTemplate(templatetemplate, input_variables[name, score])7.3 Agent开发中的典型问题问题Agent陷入循环或无法完成任务# 解决方案设置合理的迭代限制和停止条件 agent initialize_agent( toolstools, llmllm, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, max_iterations5, # 限制最大迭代次数 early_stopping_methodgenerate, # 设置早停策略 handle_parsing_errorsTrue # 处理解析错误 )7.4 RAG系统性能优化问题检索结果不准确或响应慢# 优化方案调整检索参数和分块策略 text_splitter RecursiveCharacterTextSplitter( chunk_size800, # 调整分块大小 chunk_overlap150, # 调整重叠大小 separators[\n\n, \n, 。, , , , .] ) retriever vectorstore.as_retriever( search_typemmr, # 使用最大边际相关性搜索 search_kwargs{k: 5, fetch_k: 10} # 调整检索参数 )8. 最佳实践与工程建议8.1 项目结构组织推荐的项目结构ai-support-system/ ├── src/ │ ├── agents/ # Agent相关代码 │ ├── chains/ # 自定义链 │ ├── tools/ # 工具定义 │ ├── memory/ # 内存管理 │ └── utils/ # 工具函数 ├── data/ # 数据文件 ├── tests/ # 测试代码 ├── config/ # 配置文件 └── docs/ # 项目文档8.2 配置管理最佳实践# config.py import os from dataclasses import dataclass from typing import Optional dataclass class ModelConfig: model_name: str gpt-3.5-turbo temperature: float 0.2 max_tokens: Optional[int] None dataclass class VectorStoreConfig: persist_directory: str ./chroma_db chunk_size: int 1000 chunk_overlap: int 200 dataclass class AgentConfig: max_iterations: int 5 early_stopping: bool True # 使用配置 model_config ModelConfig() vector_config VectorStoreConfig() agent_config AgentConfig()8.3 错误处理与日志记录import logging from functools import wraps def setup_logging(): 设置日志配置 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(ai_system.log), logging.StreamHandler() ] ) def error_handler(func): 错误处理装饰器 wraps(func) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: logging.error(fError in {func.__name__}: {str(e)}) # 根据错误类型返回适当的响应 return f处理过程中出现错误{str(e)} return wrapper8.4 性能监控与优化import time from contextlib import contextmanager contextmanager def timer(operation_name): 性能计时器 start_time time.time() try: yield finally: end_time time.time() duration end_time - start_time logging.info(f{operation_name} completed in {duration:.2f} seconds) # 使用示例 with timer(知识库检索): result qa_chain.invoke({query: question})通过本文的完整学习你应该已经掌握了LangChain 1.3版本的核心概念和实战技能。从基础的Prompt工程到复杂的Agent和RAG系统开发这些知识能够帮助你在实际项目中构建强大的AI应用。建议按照本文的示例代码逐步实践先从简单的Prompt模板开始逐步扩展到完整的RAG系统。在实际项目中记得做好错误处理、日志记录和性能监控确保系统的稳定性和可维护性。
返回列表