当前位置：首页 > news >正文

3 LangChain 核心组件详解：构建企业级AI应用的基础设施

news 2025/11/5 10:07:31

LangChain 核心组件详解：构建企业级AI应用的基础设施

引言

在AI工程化实践中，单纯的大语言模型调用往往无法满足复杂业务需求。LangChain作为一个强大的框架，提供了构建复杂AI应用所需的各类组件。本节将深入解析LangChain的核心组件，并通过完整的企业级示例展示如何构建可维护、可扩展的AI应用。

LangChain 架构概览

LangChain采用分层架构设计，让我们先通过一个架构图来理解其核心组件关系：

LLMChain：基础链式操作

LLMChain是LangChain中最基础的组件，它将Prompt模板、LLM和输出解析器组合在一起。

基础LLMChain实现

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI
from langchain.output_parsers import CommaSeparatedListOutputParser
import osclass BasicChainDemo:"""基础链式操作演示"""def __init__(self, openai_api_key: str):"""初始化基础链Args:openai_api_key: OpenAI API密钥"""self.llm = OpenAI(openai_api_key=openai_api_key,temperature=0.7,max_tokens=1000)def demo_basic_chain(self):"""演示基础LLMChain"""print("=== 基础LLMChain演示 ===")# 1. 创建Prompt模板prompt_template = PromptTemplate(input_variables=["product"],template="为以下产品写3个吸引人的广告标语：{product}")# 2. 创建LLMChainchain = LLMChain(llm=self.llm, prompt=prompt_template)# 3. 运行链products = ["智能手表", "电动汽车", "智能家居系统"]for product in products:result = chain.run(product=product)print(f"产品: {product}")print(f"广告标语: {result}")print("-" * 50)def demo_chain_with_parser(self):"""演示带输出解析器的链"""print("\n=== 带输出解析器的LLMChain ===")# 输出解析器 - 将输出解析为列表output_parser = CommaSeparatedListOutputParser()# 创建包含格式指令的Prompt模板template = """请为{category}类别列出{count}个热门产品名称。只返回产品名称，用逗号分隔。{format_instructions}"""prompt = PromptTemplate(input_variables=["category", "count"],template=template,partial_variables={"format_instructions": output_parser.get_format_instructions()})chain = LLMChain(llm=self.llm, prompt=prompt, output_parser=output_parser)# 运行链categories = [{"category": "智能手机", "count": 5},{"category": "笔记本电脑", "count": 4}]for category_info in categories:result = chain.run(**category_info)print(f"类别: {category_info['category']}")print(f"产品列表: {result}")print(f"解析类型: {type(result)}")print("-" * 50)# 使用示例
def run_basic_chain_demo():"""运行基础链演示"""# 注意：需要设置OPENAI_API_KEY环境变量demo = BasicChainDemo(openai_api_key=os.getenv("OPENAI_API_KEY"))demo.demo_basic_chain()demo.demo_chain_with_parser()if __name__ == "__main__":run_basic_chain_demo()

高级链式操作：SequentialChain

SequentialChain允许我们将多个链按顺序连接，实现复杂的多步处理。

from langchain.chains import SimpleSequentialChain, SequentialChain
from langchain.prompts import ChatPromptTemplate
from langchain.chat_models import ChatOpenAIclass AdvancedChainDemo:"""高级链式操作演示"""def __init__(self, openai_api_key: str):self.llm = ChatOpenAI(openai_api_key=openai_api_key,temperature=0.7,model="gpt-3.5-turbo")def demo_simple_sequential_chain(self):"""演示简单顺序链"""print("=== 简单顺序链演示 ===")# 第一个链：生成产品创意first_prompt = ChatPromptTemplate.from_template("基于{industry}行业，生成3个创新的产品创意。只返回产品名称。")chain_one = LLMChain(llm=self.llm, prompt=first_prompt, output_key="product_ideas")# 第二个链：为产品创意写描述second_prompt = ChatPromptTemplate.from_template("为以下产品创意写一个详细的产品描述：{product_ideas}")chain_two = LLMChain(llm=self.llm, prompt=second_prompt, output_key="product_description")# 创建顺序链overall_chain = SimpleSequentialChain(chains=[chain_one, chain_two],verbose=True)# 运行链industries = ["教育科技", "医疗健康", "金融科技"]for industry in industries:print(f"\n行业: {industry}")result = overall_chain.run(industry)print(f"结果: {result}")def demo_complex_sequential_chain(self):"""演示复杂顺序链"""print("\n=== 复杂顺序链演示 ===")# 第一个链：市场分析market_analysis_template = """对{product}进行市场分析，包括：1. 目标客户群体2. 主要竞争对手3. 市场趋势产品：{product}"""market_analysis_prompt = PromptTemplate(input_variables=["product"],template=market_analysis_template)market_analysis_chain = LLMChain(llm=self.llm,prompt=market_analysis_prompt,output_key="market_analysis")# 第二个链：产品定位positioning_template = """基于以下市场分析，为{product}制定产品定位策略：{market_analysis}"""positioning_prompt = PromptTemplate(input_variables=["product", "market_analysis"],template=positioning_template)positioning_chain = LLMChain(llm=self.llm,prompt=positioning_prompt,output_key="positioning_strategy")# 第三个链：营销建议marketing_template = """基于产品定位，为{product}制定具体的营销建议：产品定位：{positioning_strategy}"""marketing_prompt = PromptTemplate(input_variables=["product", "positioning_strategy"],template=marketing_template)marketing_chain = LLMChain(llm=self.llm,prompt=marketing_prompt,output_key="marketing_recommendations")# 创建复杂顺序链overall_chain = SequentialChain(chains=[market_analysis_chain,positioning_chain,marketing_chain],input_variables=["product"],output_variables=["market_analysis", "positioning_strategy", "marketing_recommendations"],verbose=True)# 运行链products = ["智能健身镜", "AI学习助手"]for product in products:print(f"\n产品: {product}")result = overall_chain({"product": product})print("市场分析:")print(result["market_analysis"])print("\n产品定位:")print(result["positioning_strategy"])print("\n营销建议:")print(result["marketing_recommendations"])print("=" * 80)# 使用示例
def run_advanced_chain_demo():"""运行高级链演示"""demo = AdvancedChainDemo(openai_api_key=os.getenv("OPENAI_API_KEY"))demo.demo_simple_sequential_chain()demo.demo_complex_sequential_chain()

Agents：智能代理系统

Agents是LangChain中最强大的组件之一，它们可以理解用户意图、选择合适工具并执行复杂任务。

ZeroShotReactDescriptionAgent（ReAct框架）

from langchain.agents import initialize_agent, Tool, AgentType
from langchain.utilities import SerpAPIWrapper, WikipediaAPIWrapper
from langchain.tools import DuckDuckGoSearchRun
import math
import datetimeclass AdvancedAgentSystem:"""高级代理系统"""def __init__(self, openai_api_key: str, serpapi_key: str = None):"""初始化代理系统Args:openai_api_key: OpenAI API密钥serpapi_key: SerpAPI密钥（可选）"""self.llm = ChatOpenAI(openai_api_key=openai_api_key,temperature=0,model="gpt-3.5-turbo")self.setup_tools(serpapi_key)def setup_tools(self, serpapi_key: str):"""设置工具集"""# 自定义工具：计算器def calculator_tool(expression: str) -> str:"""执行数学计算"""try:# 安全地执行数学表达式allowed_names = {'abs': abs, 'round': round, 'max': max, 'min': min,'pow': pow, 'sqrt': math.sqrt, 'log': math.log}result = eval(expression, {"__builtins__": {}}, allowed_names)return f"计算结果: {expression} = {result}"except Exception as e:return f"计算错误: {str(e)}"# 自定义工具：时间查询def time_tool(timezone: str = "UTC") -> str:"""获取当前时间"""now = datetime.datetime.now()return f"当前时间（{timezone}）: {now.strftime('%Y-%m-%d %H:%M:%S')}"# 创建工具列表self.tools = [Tool(name="Calculator",func=calculator_tool,description="""用于执行数学计算。输入应该是一个数学表达式。示例：'2 + 3 * 4' 或 'sqrt(16)'"""),Tool(name="Time",func=lambda x: time_tool(),description="""用于获取当前时间。不需要输入参数。""")]# 如果提供了SerpAPI密钥，添加搜索工具if serpapi_key:search = SerpAPIWrapper(serpapi_key=serpapi_key)self.tools.append(Tool(name="Search",func=search.run,description="用于搜索最新信息。输入应该是搜索查询。"))else:# 使用DuckDuckGo作为备用搜索工具search = DuckDuckGoSearchRun()self.tools.append(Tool(name="Search",func=search.run,description="用于搜索最新信息。输入应该是搜索查询。"))# 添加Wikipedia工具wikipedia = WikipediaAPIWrapper()self.tools.append(Tool(name="Wikipedia",func=wikipedia.run,description="用于在Wikipedia上搜索信息。输入应该是搜索查询。"))def create_react_agent(self):"""创建ReAct代理"""agent = initialize_agent(tools=self.tools,llm=self.llm,agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,verbose=True,handle_parsing_errors=True,max_iterations=5)return agentdef demo_agent_reasoning(self):"""演示代理推理过程"""print("=== ReAct代理推理演示 ===")agent = self.create_react_agent()# 复杂查询示例queries = ["计算(15 + 25) * 3的结果，然后告诉我现在的时间","搜索人工智能的最新发展，然后总结主要趋势","在Wikipedia上查找机器学习的定义，并解释其重要性"]for query in queries:print(f"\n查询: {query}")print("思考过程:")try:result = agent.run(query)print(f"最终结果: {result}")except Exception as e:print(f"执行错误: {e}")print("=" * 80)# 企业级代理应用：客户服务代理
class CustomerServiceAgent:"""客户服务代理系统"""def __init__(self, openai_api_key: str):self.llm = ChatOpenAI(openai_api_key=openai_api_key,temperature=0.3,model="gpt-3.5-turbo")self.setup_customer_tools()def setup_customer_tools(self):"""设置客户服务专用工具"""# 模拟订单数据库self.orders_db = {"ORD001": {"status": "已发货", "product": "智能手机", "tracking": "TRK123456"},"ORD002": {"status": "处理中", "product": "笔记本电脑", "tracking": None},"ORD003": {"status": "已送达", "product": "耳机", "tracking": "TRK789012"}}# 订单查询工具def order_lookup(order_id: str) -> str:"""查询订单状态"""order = self.orders_db.get(order_id.upper())if order:return f"订单 {order_id}: 状态-{order['status']}, 产品-{order['product']}, 追踪号-{order['tracking'] or '暂无'}"else:return f"未找到订单 {order_id}"# 退货政策查询工具def return_policy(product_category: str) -> str:"""查询退货政策"""policies = {"电子产品": "7天内无理由退货，30天内质量问题换货","服装": "15天内无理由退货，尺码不合适可换货","食品": "不支持无理由退货，质量问题24小时内联系客服"}return policies.get(product_category, "通用政策：7天无理由退货")# 客服工具self.tools = [Tool(name="OrderLookup",func=order_lookup,description="用于查询订单状态。输入应该是订单号。"),Tool(name="ReturnPolicy",func=return_policy,description="用于查询退货政策。输入应该是产品类别，如'电子产品'、'服装'等。")]def handle_customer_query(self, query: str):"""处理客户查询"""agent = initialize_agent(tools=self.tools,llm=self.llm,agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,verbose=True,handle_parsing_errors=True)try:response = agent.run(f"""作为客服代表，请专业地回答以下客户问题：{query}请使用可用工具获取准确信息，并提供有帮助的回复。""")return responseexcept Exception as e:return f"处理查询时出现错误: {str(e)}"# 运行代理演示
def run_agent_demos():"""运行代理演示"""# 基础代理演示print("=== 基础代理系统演示 ===")agent_system = AdvancedAgentSystem(openai_api_key=os.getenv("OPENAI_API_KEY"))agent_system.demo_agent_reasoning()# 客户服务代理演示print("\n=== 客户服务代理演示 ===")cs_agent = CustomerServiceAgent(openai_api_key=os.getenv("OPENAI_API_KEY"))customer_queries = ["我的订单ORD001状态怎么样？","电子产品退货政策是什么？","请查询ORD002和ORD003的状态","服装类商品可以退货吗？"]for query in customer_queries:print(f"\n客户查询: {query}")response = cs_agent.handle_customer_query(query)print(f"客服回复: {response}")print("-" * 60)

Plan-and-Execute Agent（计划+执行模式）

Plan-and-Execute Agent采用更复杂的架构，先制定计划再执行，适合复杂任务。

from langchain.experimental.plan_and_execute import PlanAndExecute, load_agent_executor, load_chat_planner
from langchain.agents import AgentExecutorclass PlanExecuteAgentSystem:"""计划执行代理系统"""def __init__(self, openai_api_key: str):self.llm = ChatOpenAI(openai_api_key=openai_api_key,temperature=0,model="gpt-3.5-turbo")self.setup_planning_tools()def setup_planning_tools(self):"""设置计划执行工具"""# 研究工具def research_tool(topic: str) -> str:"""研究特定主题"""return f"""{topic}的研究结果：1. 当前市场趋势：快速增长2. 主要挑战：技术复杂度高3. 发展机会：AI集成4. 建议：加强研发投入"""# 分析工具def analysis_tool(data: str) -> str:"""分析数据"""return f"""数据分析结果：- 识别出3个关键模式- 发现2个潜在风险- 提出4个改进建议基于输入数据：{data[:100]}..."""# 报告生成工具def report_tool(findings: str) -> str:"""生成报告"""return f"""综合报告：{findings}总结：基于以上分析，建议采取积极的市场策略。"""self.tools = [Tool(name="Research",func=research_tool,description="用于研究特定主题，输入应该是研究主题。"),Tool(name="Analysis",func=analysis_tool,description="用于分析数据，输入应该是分析内容。"),Tool(name="Report",func=report_tool,description="用于生成综合报告，输入应该是研究发现。")]def create_plan_execute_agent(self):"""创建计划执行代理"""planner = load_chat_planner(self.llm)executor = load_agent_executor(self.llm, self.tools, verbose=True)agent = PlanAndExecute(planner=planner,executor=executor,verbose=True)return agentdef demo_complex_task(self):"""演示复杂任务处理"""print("=== 计划执行代理演示 ===")agent = self.create_plan_execute_agent()complex_tasks = ["""进行市场研究：分析电动汽车行业的现状，包括主要厂商、技术趋势和市场机会，然后生成综合报告。""","""研究人工智能在教育领域的应用，分析当前的主要挑战和未来发展方向，并提出具体的实施建议。"""]for task in complex_tasks:print(f"\n复杂任务: {task}")try:result = agent.run(task)print(f"任务完成结果: {result}")except Exception as e:print(f"任务执行失败: {e}")print("=" * 80)# 运行计划执行代理演示
def run_plan_execute_demo():"""运行计划执行代理演示"""system = PlanExecuteAgentSystem(openai_api_key=os.getenv("OPENAI_API_KEY"))system.demo_complex_task()

Tools：集成自定义工具

Tools是代理与外部世界交互的桥梁，下面展示如何创建和使用自定义工具。

from langchain.tools import BaseTool
from typing import Type, Optional
from pydantic import BaseModel, Field
import requests
import jsonclass AdvancedToolSystem:"""高级工具系统"""def __init__(self, openai_api_key: str):self.llm = ChatOpenAI(openai_api_key=openai_api_key,temperature=0,model="gpt-3.5-turbo")# 自定义工具：天气查询
class WeatherToolInput(BaseModel):"""天气查询工具输入模型"""location: str = Field(description="城市名称，如：北京、上海")class WeatherTool(BaseTool):"""自定义天气查询工具"""name = "weather_tool"description = "用于查询指定城市的天气信息"args_schema: Type[BaseModel] = WeatherToolInputdef _run(self, location: str) -> str:"""执行天气查询"""# 模拟天气数据 - 实际应用中应该调用天气APIweather_data = {"北京": {"temperature": "25°C", "condition": "晴朗", "humidity": "40%"},"上海": {"temperature": "28°C", "condition": "多云", "humidity": "65%"},"广州": {"temperature": "32°C", "condition": "阵雨", "humidity": "80%"}}if location in weather_data:data = weather_data[location]return f"{location}天气：温度{data['temperature']}，{data['condition']}，湿度{data['humidity']}"else:return f"未找到{location}的天气信息"def _arun(self, location: str):"""异步执行"""raise NotImplementedError("此工具不支持异步执行")# 自定义工具：货币转换
class CurrencyConverterInput(BaseModel):"""货币转换工具输入模型"""amount: float = Field(description="要转换的金额")from_currency: str = Field(description="源货币代码，如：USD、CNY")to_currency: str = Field(description="目标货币代码，如：USD、CNY")class CurrencyConverterTool(BaseTool):"""货币转换工具"""name = "currency_converter"description = "用于转换货币金额"args_schema: Type[BaseModel] = CurrencyConverterInputdef _run(self, amount: float, from_currency: str, to_currency: str) -> str:"""执行货币转换"""# 模拟汇率 - 实际应用中应该调用汇率APIexchange_rates = {"USD_CNY": 7.2,"CNY_USD": 0.14,"EUR_CNY": 7.8,"CNY_EUR": 0.13}rate_key = f"{from_currency.upper()}_{to_currency.upper()}"reverse_key = f"{to_currency.upper()}_{from_currency.upper()}"if rate_key in exchange_rates:rate = exchange_rates[rate_key]converted = amount * ratereturn f"{amount} {from_currency} = {converted:.2f} {to_currency} (汇率: {rate})"elif reverse_key in exchange_rates:rate = 1 / exchange_rates[reverse_key]converted = amount * ratereturn f"{amount} {from_currency} = {converted:.2f} {to_currency} (汇率: {rate})"else:return f"不支持{from_currency}到{to_currency}的转换"def _arun(self, amount: float, from_currency: str, to_currency: str):raise NotImplementedError("此工具不支持异步执行")# 工具使用演示
def demonstrate_custom_tools():"""演示自定义工具使用"""print("=== 自定义工具演示 ===")# 创建工具实例weather_tool = WeatherTool()currency_tool = CurrencyConverterTool()# 测试工具print("天气工具测试:")print(weather_tool.run("北京"))print(weather_tool.run("上海"))print("\n货币转换工具测试:")print(currency_tool.run({"amount": 100, "from_currency": "USD", "to_currency": "CNY"}))print(currency_tool.run({"amount": 500, "from_currency": "CNY", "to_currency": "USD"}))# 创建包含自定义工具的代理llm = ChatOpenAI(openai_api_key=os.getenv("OPENAI_API_KEY"), temperature=0)tools = [weather_tool, currency_tool]agent = initialize_agent(tools=tools,llm=llm,agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,verbose=True)# 使用代理处理复杂查询queries = ["查询北京的天气，然后转换100美元到人民币","今天上海的天气如何？500人民币等于多少美元？"]for query in queries:print(f"\n查询: {query}")try:result = agent.run(query)print(f"结果: {result}")except Exception as e:print(f"错误: {e}")# 数据库工具示例
class DatabaseTool(BaseTool):"""数据库查询工具"""name = "database_query"description = "用于查询数据库信息"def _run(self, query: str) -> str:"""执行数据库查询"""# 模拟数据库响应if "用户" in query or "user" in query.lower():return "用户数据：共1000条记录，最近活跃用户500人"elif "订单" in query or "order" in query.lower():return "订单数据：本月订单500个，总金额¥100,000"elif "产品" in query or "product" in query.lower():return "产品数据：在售产品50个，库存充足"else:return f"查询结果：找到与'{query}'相关的数据"def _arun(self, query: str):raise NotImplementedError("此工具不支持异步执行")if __name__ == "__main__":# 运行各种演示run_basic_chain_demo()run_advanced_chain_demo() run_agent_demos()run_plan_execute_demo()demonstrate_custom_tools()

Memory：短期与长期记忆系统

Memory组件使LangChain应用能够记住对话历史，实现真正的多轮对话。

from langchain.memory import ConversationBufferMemory, ConversationBufferWindowMemory, RedisChatMessageHistory
from langchain.schema import BaseMessage
import redisclass MemorySystemDemo:"""记忆系统演示"""def __init__(self, openai_api_key: str):self.llm = ChatOpenAI(openai_api_key=openai_api_key,temperature=0.7,model="gpt-3.5-turbo")def demo_conversation_memory(self):"""演示对话记忆"""print("=== 对话记忆系统演示 ===")# 1. 基础对话缓冲记忆memory = ConversationBufferMemory(memory_key="chat_history",return_messages=True)# 创建带记忆的链prompt = ChatPromptTemplate.from_messages([("system", "你是一个友好的客服助手。"),MessagesPlaceholder(variable_name="chat_history"),("human", "{input}")])chain = LLMChain(llm=self.llm,prompt=prompt,memory=memory,verbose=True)# 模拟多轮对话conversations = ["你好，我叫张三","我想查询订单状态","我的订单号是ORD001","谢谢你的帮助"]for i, message in enumerate(conversations):print(f"\n第{i+1}轮对话:")print(f"用户: {message}")response = chain.run(input=message)print(f"助手: {response}")# 显示当前记忆状态print(f"记忆内容: {memory.buffer}")def demo_window_memory(self):"""演示窗口记忆（只保留最近N轮对话）"""print("\n=== 窗口记忆演示 ===")# 只保留最近2轮对话memory = ConversationBufferWindowMemory(k=2,memory_key="chat_history",return_messages=True)prompt = ChatPromptTemplate.from_messages([("system", "你是客服助手。"),MessagesPlaceholder(variable_name="chat_history"),("human", "{input}")])chain = LLMChain(llm=self.llm,prompt=prompt,memory=memory)# 测试多轮对话test_messages = ["我叫李四","我想退货","订单号ORD002","什么时候能处理完？"]for msg in test_messages:response = chain.run(input=msg)print(f"用户: {msg}")print(f"助手: {response}")print(f"当前记忆({memory.k}轮): {len(memory.chat_memory.messages)}条消息")def demo_redis_memory(self):"""演示Redis长期记忆"""print("\n=== Redis长期记忆演示 ===")try:# 连接到Redisredis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)# 创建Redis消息历史message_history = RedisChatMessageHistory(url="redis://localhost:6379/0",session_id="user_session_123")# 清空历史记录（演示用）message_history.clear()# 添加一些消息from langchain.schema import HumanMessage, AIMessagemessage_history.add_user_message("你好，我是测试用户")message_history.add_ai_message("你好！我是AI助手，有什么可以帮你的？")# 显示存储的消息print("Redis中存储的消息:")for msg in message_history.messages:print(f"- {type(msg).__name__}: {msg.content}")except Exception as e:print(f"Redis连接失败: {e}")print("请确保Redis服务器正在运行")# 企业级记忆管理
class EnterpriseMemoryManager:"""企业级记忆管理器"""def __init__(self, openai_api_key: str, redis_url: str = None):self.llm = ChatOpenAI(openai_api_key=openai_api_key, temperature=0.7)self.redis_url = redis_urldef create_session_memory(self, session_id: str, memory_type: str = "buffer"):"""创建会话记忆Args:session_id: 会话IDmemory_type: 记忆类型 (buffer/window/redis)"""if memory_type == "redis" and self.redis_url:from langchain.memory import RedisChatMessageHistorymessage_history = RedisChatMessageHistory(url=self.redis_url,session_id=session_id)from langchain.memory import ConversationBufferMemorymemory = ConversationBufferMemory(chat_memory=message_history,memory_key="chat_history",return_messages=True)elif memory_type == "window":memory = ConversationBufferWindowMemory(k=5,  # 保留最近5轮对话memory_key="chat_history",return_messages=True)else:memory = ConversationBufferMemory(memory_key="chat_history",return_messages=True)return memory# 运行记忆系统演示
def run_memory_demos():"""运行记忆系统演示"""demo = MemorySystemDemo(openai_api_key=os.getenv("OPENAI_API_KEY"))demo.demo_conversation_memory()demo.demo_window_memory()demo.demo_redis_memory()if __name__ == "__main__":# 运行所有演示run_memory_demos()