从API到Agent：万字长文洞悉LangChain工程化设计

引言：为什么LangChain值得深入理解？在现代AI应用开发中，LangChain已成为连接LLM（大语言模型）与实际业务场景的核心框架。从最初的API调用到复杂的Agent系统，LangChain提供了一套工程化的设计范式，让开发者能够快速构建可扩展的智能应用。本文将通过实战代码，深入剖析LangChain的核心设计理念、组件化架构以及从简单API到复杂Agent的演进路径。## 一、API调用的基础：从OpenAI到LangChain### 1.1 原生API调用 vs LangChain封装首先，我们看看直接调用OpenAI API与使用LangChain的对比：python# 原生OpenAI API调用import openairesponse = openai.ChatCompletion.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": "你是一个AI助手"}, {"role": "user", "content": "解释什么是LangChain?"} ], temperature=0.7)print(response.choices[0].message.content)# LangChain封装版本from langchain_openai import ChatOpenAIfrom langchain.schema import HumanMessage, SystemMessagellm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)messages = [ SystemMessage(content="你是一个AI助手"), HumanMessage(content="解释什么是LangChain?")]response = llm.invoke(messages)print(response.content)核心差异：- LangChain提供了统一的接口抽象，支持多种LLM模型切换- 消息结构使用了类型化的SystemMessage、HumanMessage等对象- 接口命名更加语义化（invoke代替create）### 1.2 构建可复用Prompt模板工程化设计的核心之一是模板化。LangChain的PromptTemplate让Prompt管理变得优雅：pythonfrom langchain.prompts import PromptTemplatefrom langchain_openai import ChatOpenAI# 定义可复用的Prompt模板template = """你是一个专业的{role}。请根据以下问题提供详细解答：问题：{question}要求：- 回答要专业且易于理解- 使用中文回答- 字数控制在{max_words}字以内"""prompt = PromptTemplate( input_variables=["role", "question", "max_words"], template=template)# 使用模板生成最终Promptfinal_prompt = prompt.format( role="AI工程师", question="如何优化大模型的推理性能？", max_words=200)# 调用LLMllm = ChatOpenAI(model="gpt-3.5-turbo")response = llm.invoke(final_prompt)print(response.content)设计亮点：- 通过input_variables动态注入变量，避免硬编码- 支持复杂逻辑的模板组合- 可版本化管理不同场景的Prompt## 二、Chain：构建可组合的处理流水线### 2.1 LLMChain的实战应用Chain是LangChain的核心组件，它定义了处理逻辑的流水线。下面是一个完整的问答链：pythonfrom langchain.chains import LLMChainfrom langchain.prompts import ChatPromptTemplatefrom langchain_openai import ChatOpenAI# 1. 定义Prompt模板prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个{field}领域的专家，回答要简洁准确。"), ("human", "{query}")])# 2. 初始化LLMllm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.3)# 3. 构建Chainchain = LLMChain( llm=llm, prompt=prompt, verbose=True # 开启日志，方便调试)# 4. 执行Chainresult = chain.invoke({ "field": "机器学习", "query": "解释过拟合的解决方法"})print(f"输出：{result['text']}")Chain的设计哲学：- 单一职责：每个Chain只完成一个特定任务- 可组合性：通过SequentialChain、SimpleSequentialChain等组合多个Chain- 参数传递：输入/输出通过字典映射，支持复杂的变量流动### 2.2 构建多步骤处理流水线pythonfrom langchain.chains import SequentialChainfrom langchain_openai import ChatOpenAI# 第一步：生成代码code_prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个Python专家，生成高质量的代码。"), ("human", "请用Python写一个函数：{task}")])code_chain = LLMChain(llm=ChatOpenAI(), prompt=code_prompt, output_key="code")# 第二步：解释代码explain_prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个技术文档专家。"), ("human", "请用中文解释以下代码的功能和关键点：\n{code}")])explain_chain = LLMChain(llm=ChatOpenAI(), prompt=explain_prompt, output_key="explanation")# 组合成流水线pipeline = SequentialChain( chains=[code_chain, explain_chain], input_variables=["task"], output_variables=["code", "explanation"], verbose=True)# 执行流水线result = pipeline.invoke({"task": "实现一个快速排序算法"})print(f"生成的代码：\n{result['code']}")print(f"\n解释：\n{result['explanation']}")工程化优势：- 每个步骤可以独立测试和优化- 通过output_key控制变量传递- 支持条件分支和循环逻辑（通过RouterChain）## 三、Memory：让AI记住对话上下文### 3.1 基于Buffer的记忆管理在对话应用中，上下文管理至关重要。LangChain提供了多种Memory实现：pythonfrom langchain.memory import ConversationBufferMemoryfrom langchain.chains import ConversationChainfrom langchain_openai import ChatOpenAI# 初始化带记忆的对话链memory = ConversationBufferMemory( return_messages=True, memory_key="history", input_key="input")llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)conversation = ConversationChain( llm=llm, memory=memory, verbose=True)# 连续对话测试print(conversation.invoke({"input": "你好！我叫小明。"})['response'])print(conversation.invoke({"input": "你知道我的名字吗？"})['response'])print(conversation.invoke({"input": "我刚刚说了什么？"})['response'])Memory的核心功能：- 自动保存历史对话- 支持多种存储策略（Buffer、Summary、VectorStore等）- 可自定义窗口大小和摘要策略### 3.2 自定义Memory实现pythonfrom langchain.memory import BaseMemoryfrom typing import Dict, Any, Listclass CustomMemory(BaseMemory): """自定义记忆：只保留最近3轮对话""" def __init__(self, max_turns=3): super().__init__() self.max_turns = max_turns self.buffer: List[Dict] = [] self.memory_key = "history" self.input_key = "input" @property def memory_variables(self) -> List[str]: return [self.memory_key] def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]: # 格式化历史记录 formatted = "\n".join([ f"用户: {msg['user']}\nAI: {msg['ai']}" for msg in self.buffer ]) return {self.memory_key: formatted} def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, Any]) -> None: # 保存新对话，并控制长度 self.buffer.append({ "user": inputs.get(self.input_key, ""), "ai": outputs.get("response", "") }) if len(self.buffer) > self.max_turns: self.buffer.pop(0) def clear(self) -> None: self.buffer = []# 使用自定义记忆custom_convo = ConversationChain( llm=ChatOpenAI(), memory=CustomMemory(max_turns=2), verbose=True)print(custom_convo.invoke({"input": "天气如何？"})['response'])print(custom_convo.invoke({"input": "明天呢？"})['response'])设计模式：- 继承BaseMemory基类，实现load_memory_variables和save_context- 控制记忆的存储策略和生命周期- 与Chain无缝集成，无需修改业务逻辑## 四、Agent：从被动响应到主动推理### 4.1 构建第一个AgentAgent是LangChain最强大的特性之一，它让LLM能够自主决策、调用工具。下面是一个数学计算Agent：pythonfrom langchain.agents import AgentExecutor, create_react_agentfrom langchain.tools import Toolfrom langchain_openai import ChatOpenAIfrom langchain.prompts import PromptTemplateimport math# 定义工具函数def calculate(expression: str) -> str: """计算数学表达式""" try: result = eval(expression, {"__builtins__": {}}, math.__dict__) return str(result) except Exception as e: return f"计算错误：{str(e)}"def get_current_time(query: str) -> str: """获取当前时间""" from datetime import datetime return datetime.now().strftime("%Y-%m-%d %H:%M:%S")# 创建工具列表tools = [ Tool( name="计算器", func=calculate, description="用于执行数学运算，输入数学表达式" ), Tool( name="当前时间", func=get_current_time, description="获取当前的日期和时间" )]# 初始化LLMllm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)# 创建Agentprompt = PromptTemplate.from_template("""你是一个智能助手，可以调用工具来完成任务。可用的工具：{tools}目标：{input}工作流程：1. 分析问题2. 决定是否需要调用工具3. 调用工具获取结果4. 给出最终答案{agent_scratchpad}""")agent = create_react_agent( llm=llm, tools=tools, prompt=prompt)agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, max_iterations=3 # 防止无限循环)# 测试Agentresponse = agent_executor.invoke({ "input": "计算 2的10次方 加上 3的5次方，然后告诉我当前时间"})print(f"\n最终答案：{response['output']}")Agent的核心机制：1. ReAct模式：思考-行动-观察的循环2. 工具注册：通过Tool对象注册可调用的功能3. 决策逻辑：LLM自主决定何时调用哪个工具### 4.2 多工具协作Agent实战pythonfrom langchain.agents import create_structured_chat_agentfrom langchain.agents.agent import AgentExecutorfrom langchain.tools import toolimport jsonimport requests# 使用装饰器定义工具@tooldef search_weather(city: str) -> str: """查询指定城市的天气情况""" # 模拟API调用 weather_data = { "北京": "晴，温度25°C", "上海": "多云，温度28°C", "深圳": "小雨，温度30°C" } return weather_data.get(city, f"未找到{city}的天气信息")@tooldef translate_text(text: str, target_lang: str = "中文") -> str: """翻译文本到目标语言""" translations = { ("hello", "中文"): "你好", ("world", "中文"): "世界", ("good morning", "中文"): "早上好" } key = (text.lower(), target_lang) return translations.get(key, f"翻译失败：{text} -> {target_lang}")@tooldef generate_code(task_description: str) -> str: """根据描述生成Python代码""" code_templates = { "排序": "def sort_list(lst):\n return sorted(lst)", "计算平均值": "def average(nums):\n return sum(nums)/len(nums)" } for keyword, code in code_templates.items(): if keyword in task_description: return code return "# 无法生成匹配的代码"# 构建高级Agenttools = [search_weather, translate_text, generate_code]system_message = """你是一个多功能AI助手，可以：- 查询天气- 翻译文本- 生成代码请根据用户需求，选择合适的工具完成任务。"""agent = create_structured_chat_agent( llm=ChatOpenAI(model="gpt-4", temperature=0), tools=tools, system_message=system_message)agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, handle_parsing_errors=True, max_iterations=5)# 复杂任务测试tasks = [ "北京的天气怎么样？顺便把'hello world'翻译成中文", "帮我写一个排序算法，然后查询上海的天气"]for task in tasks: print(f"\n{'='*50}") print(f"任务：{task}") result = agent_executor.invoke({"input": task}) print(f"结果：{result['output']}")工程化特性：- 工具装饰器@tool简化定义- 结构化Agent支持多步骤推理- 错误处理机制（handle_parsing_errors）- 迭代次数控制防止死循环## 五、高级主题：自定义组件与优化### 5.1 自定义Callback监控pythonfrom langchain.callbacks.base import BaseCallbackHandlerfrom langchain.schema import LLMResultfrom typing import Any, Dict, Listclass CustomCallbackHandler(BaseCallbackHandler): """自定义回调：记录所有LLM调用""" def __init__(self): self.llm_calls = [] self.total_tokens = 0 def on_llm_start( self, serialized: Dict[str, Any], prompts: List[str], **kwargs ) -> None: print(f"LLM调用开始：{prompts[:50]}...") def on_llm_end(self, response: LLMResult, **kwargs) -> None: # 记录token使用情况 for gen in response.generations: for g in gen: if hasattr(g, 'generation_info') and g.generation_info: tokens = g.generation_info.get('token_usage', {}) self.total_tokens += tokens.get('total_tokens', 0) self.llm_calls.append({ 'response': response, 'tokens': self.total_tokens }) print(f"LLM调用结束，累计Token：{self.total_tokens}") def get_stats(self) -> Dict: return { "total_calls": len(self.llm_calls), "total_tokens": self.total_tokens }# 使用自定义Callbackcallback = CustomCallbackHandler()llm = ChatOpenAI( model="gpt-3.5-turbo", callbacks=[callback])chain = LLMChain(llm=llm, prompt=ChatPromptTemplate.from_template("回答：{q}"))chain.invoke({"q": "LangChain有哪些核心组件？"})print(f"统计信息：{callback.get_stats()}")### 5.2 多模型切换与负载均衡pythonfrom langchain.llms import BaseLLMfrom typing import Optional, Listimport randomclass LoadBalancedLLM(BaseLLM): """负载均衡的LLM包装器""" def __init__(self, models: List[BaseLLM]): super().__init__() self.models = models self.current_index = 0 def _generate(self, prompts: List[str], **kwargs): # 轮询选择模型 model = self.models[self.current_index] self.current_index = (self.current_index + 1) % len(self.models) return model._generate(prompts, **kwargs) @property def _llm_type(self) -> str: return "load_balanced_llm"# 创建多个模型实例model1 = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.1)model2 = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.5)model3 = ChatOpenAI(model="gpt-4", temperature=0.3)# 构建负载均衡器balanced_llm = LoadBalancedLLM([model1, model2, model3])# 测试负载均衡for i in range(5): response = balanced_llm.invoke(f"测试请求{i}") print(f"请求{i}：{response[:50]}...")## 六、性能优化与工程实践### 6.1 缓存策略pythonfrom langchain.cache import InMemoryCachefrom langchain.globals import set_llm_cache# 启用LLM缓存set_llm_cache(InMemoryCache())llm = ChatOpenAI(model="gpt-3.5-turbo")# 第一次调用import timestart = time.time()result1 = llm.invoke("解释机器学习")print(f"第一次调用耗时：{time.time() - start:.2f}s")# 第二次调用（命中缓存）start = time.time()result2 = llm.invoke("解释机器学习")print(f"第二次调用耗时：{time.time() - start:.2f}s")print(f"结果相同：{result1 == result2}")### 6.2