LangChain 提供了丰富的预构建组件集,用于与大语言模型(LLMs)交互、构建提示词和解析输出,但生产环境常有独特需求,要求扩展或替换这些标准实现。该框架的模块化设计基于明确定义的接口构建,允许开发者在需要的地方精确注入自定义逻辑。定制核心构建块,如LLM包装器、提示词模板和输出解析器,可以调整LangChain的行为以适应特定的模型交互、复杂的输入格式和非标准输出结构。掌握这些定制对于构建专门且高效的LLM应用很有用。定制 LLM 和 ChatModel 包装器LangChain 中的标准 LLM 和 ChatModel 包装器处理与各类模型提供商(OpenAI、Anthropic、Cohere、Hugging Face 模型等)的通信。但您可能因以下几个原因需要自定义包装器:不支持的模型: 与 LangChain 不直接支持的 LLM 提供商或自托管模型进行连接。特定API参数: 需要传递自定义头部信息、使用标准包装器不支持的特定推理参数,或实现独特的身份验证方案。自定义日志/监控: 添加针对您的基础设施的详细请求、响应、token计数或延迟日志。模型特定错误处理: 根据特定模型的行为实现定制的重试逻辑或错误处理。请求/响应修改: 在提示词传递回 LangChain 框架之前进行预处理,或在模型响应传递回 LangChain 框架之前进行后处理(例如,注入安全前缀、过滤输出)。创建自定义 LLM 包装器,您通常继承自 langchain_core.language_models.llms.LLM 并实现 _call 方法用于同步执行,以及 _acall 用于异步执行。对于聊天模型,继承自 langchain_core.language_models.chat_models.BaseChatModel 并实现 _generate 和 _agenerate。一个很有用的方法是 _generate,它接受一个消息列表和可选的 stop 序列,对它们进行处理,并返回一个包含 ChatGeneration 对象的 ChatResult。您还需要实现 _llm_type 属性来识别您的自定义模型。我们通过一个例子来说明自定义聊天模型包装器,它为每个用户消息添加一个简单前缀并记录交互时长。import time from typing import Any, List, Optional from langchain_core.callbacks.manager import CallbackManagerForLLMRun from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.messages import BaseMessage, AIMessage, HumanMessage from langchain_core.outputs import ChatGeneration, ChatResult # 假设我们有一个要包装的现有聊天模型实例 # from langchain_openai import ChatOpenAI # underlying_chat_model = ChatOpenAI() class CustomLoggedChatWrapper(BaseChatModel): """ 一个为用户消息添加前缀并记录交互时间的包装器。 """ underlying_model: BaseChatModel # 实际要调用的模型 @property def _llm_type(self) -> str: return "custom_logged_chat_wrapper" def _generate( self, messages: List[BaseMessage], stop: Optional[List[str]] = None, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any, ) -> ChatResult: start_time = time.time() # 修改消息(示例:为 HumanMessage 添加前缀) processed_messages = [] for msg in messages: if isinstance(msg, HumanMessage): processed_messages.append( HumanMessage(content=f"[User Inquiry] {msg.content}", additional_kwargs=msg.additional_kwargs) ) else: processed_messages.append(msg) # 调用底层模型 # 在实际场景中,您会适当地传递回调和关键字参数 result = self.underlying_model._generate(processed_messages, stop=stop, run_manager=run_manager, **kwargs) end_time = time.time() duration = end_time - start_time print(f"Custom Wrapper: Interaction took {duration:.2f} seconds.") # 返回前可能修改结果 # 例如,如果 result.llm_output 中有 token 使用情况,则记录下来 return result # 类似地实现 _agenerate 以支持异步 async def _agenerate( self, messages: List[BaseMessage], stop: Optional[List[str]] = None, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any, ) -> ChatResult: # 异步实现将使用 underlying_model._agenerate # 以及异步时间跟踪/日志记录 # (为简洁起见,省略实现) pass # 使用示例(假设 underlying_chat_model 已定义) # custom_llm = CustomLoggedChatWrapper(underlying_model=underlying_chat_model) # response = custom_llm.invoke([HumanMessage(content="Tell me about LangChain.")]) # print(response)这个例子展示了如何包装现有模型、修改输入(processed_messages)、调用底层模型的生成方法并添加自定义逻辑(计时)。对于完全不支持的模型的包装器,将涉及在 _generate 中进行直接 API 调用并手动构建 ChatResult。请记得实现异步对应方法(_agenerate、_acall),以便与 ainvoke 等异步 LangChain 功能兼容。定制提示词模板LangChain 的 PromptTemplate 和 ChatPromptTemplate 为输入格式化提供了灵活性,但有时您需要更精细的逻辑:条件格式化: 根据输入变量包含或排除提示词的部分内容。动态指令: 根据当前上下文或任务复杂度即时生成指令或示例。复杂输入结构: 处理非简单键值对的输入,可能涉及需要特定格式的嵌套对象或列表。集成外部状态: 将来自外部源(如数据库或配置文件)的信息直接纳入提示词逻辑。您可以通过继承 langchain_core.prompts.BasePromptTemplate 或 langchain_core.prompts.BaseChatPromptTemplate 来创建自定义提示词模板。主要实现的方法是 format_prompt(应返回 PromptValue),或对于更简单的模板,只需 format(返回格式化字符串)。对于聊天模板,您将实现 format_messages。考虑一个场景,我们想要一个聊天提示词模板,它添加一个系统消息来概述预期的输出格式,仅当用户的查询表明需要结构化数据时(例如,包含“list”或“table”字样)。from typing import Any, Dict, List from langchain_core.prompts import BaseChatPromptTemplate from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage from pydantic import BaseModel, Field class DynamicStructureChatPrompt(BaseChatPromptTemplate, BaseModel): """ 一个聊天提示词模板,它根据条件添加关于结构化输出的系统消息。 """ # 如果需要,使用 Pydantic 字段定义输入变量 input_variables: List[str] = Field(default=["user_query"]) def format_messages(self, **kwargs: Any) -> List[BaseMessage]: user_query = kwargs["user_query"] messages: List[BaseMessage] = [] # 系统消息的条件逻辑 if "list" in user_query.lower() or "table" in user_query.lower(): messages.append( SystemMessage(content="Please provide the output as a structured list or table.") ) else: messages.append( SystemMessage(content="You are a helpful assistant.") ) messages.append(HumanMessage(content=user_query)) # 可能根据其他关键字参数或逻辑添加更多消息 return messages def _prompt_type(self) -> str: return "dynamic-structure-chat-prompt" # 使用示例: dynamic_prompt = DynamicStructureChatPrompt() query1 = "Summarize the benefits of LCEL." messages1 = dynamic_prompt.format_messages(user_query=query1) # messages1 将是 [SystemMessage(...有用的助手...), HumanMessage(...总结...)] query2 = "Give me a list of vector stores supported by LangChain." messages2 = dynamic_prompt.format_messages(user_query=query2) # messages2 将是 [SystemMessage(...结构化列表/表格...), HumanMessage(...给我一个列表...)] # print(messages1) # print(messages2)这个 DynamicStructureChatPrompt 检查 user_query 输入变量以决定包含哪个系统消息。这使得基于输入性质的自适应提示成为可能,可能更有效地引导LLM。定制输出解析器输出解析器负责将LLM的原始字符串或消息输出进行转换,转换为更结构化的格式(如字典、Pydantic对象或自定义领域对象)。尽管 LangChain 提供了常见格式(JSON、CSV、Pydantic 模型、结构化函数/工具调用)的解析器,但在以下情况需要自定义解析器:处理不规则格式: LLM 生成的输出不严格遵循 JSON 或其他标准结构,需要灵活的解析逻辑(例如,使用正则表达式、宽松解析)。提取特定信息: 您需要从一个更大的非结构化或半结构化文本块中提取特定信息,该文本块由 LLM 生成。实现复杂验证: 输出需要比标准 Pydantic 模型更高级的验证逻辑。尝试纠正/重试: 解析器需要识别解析失败,并可能触发纠正循环(例如,用错误信息重新提示LLM)。专有格式: 处理您的应用程序定义的领域特定或非标准输出结构。创建自定义输出解析器,继承自 langchain_core.output_parsers.BaseOutputParser。主要实现的方法是 parse,它接受LLM的输出字符串(或通过 parse_result 接受 Generation 对象)并返回所需的结构化数据。您也可以实现 get_format_instructions 为LLM提供关于如何格式化其输出的指导。设想一个场景,LLM 有时返回一个人的姓名和年龄,但格式各异(例如,“Name: Alice, Age: 30”、“Bob is 25 years old”、“Age: 40 Name: Carol”)。我们可以使用正则表达式创建一个解析器。import re from typing import Dict, Any from langchain_core.output_parsers import BaseOutputParser from langchain_core.exceptions import OutputParserException class RegexNameAgeParser(BaseOutputParser[Dict[str, Any]]): """ 使用正则表达式解析文本以提取姓名和年龄, 处理各种不同的格式。 """ def parse(self, text: str) -> Dict[str, Any]: """解析输出文本以提取姓名和年龄。""" name_match = re.search(r"(?:Name:\s*|)(\b[A-Z][a-z]+)\b", text, re.IGNORECASE) age_match = re.search(r"(?:Age:\s*|is\s*|)(\d+)\s*(?:years old|)", text, re.IGNORECASE) if name_match and age_match: name = name_match.group(1) age = int(age_match.group(1)) return {"name": name, "age": age} else: # 处理解析失败的情况 # 选项 1:抛出异常 raise OutputParserException( f"Could not parse Name and Age from output: {text}" ) # 选项 2:返回默认/错误结构 # return {"name": None, "age": None, "error": "Parsing failed"} def get_format_instructions(self) -> str: """为LLM提供关于所需格式的指令。""" # 虽然这个解析器很灵活,但提供指令仍然有所帮助。 return "Please provide the name and age. For example: 'Name: John, Age: 35'" @property def _type(self) -> str: return "regex_name_age_parser" # 使用示例: parser = RegexNameAgeParser() output1 = "The user is Alice, Age: 30." output2 = "Bob is 25 years old." output3 = "Age: 40 Name: Carol" output4 = "David's information is not available." try: parsed1 = parser.parse(output1) # {"name": "Alice", "age": 30} parsed2 = parser.parse(output2) # {"name": "Bob", "age": 25} parsed3 = parser.parse(output3) # {"name": "Carol", "age": 40} # print(parsed1, parsed2, parsed3) parsed4 = parser.parse(output4) except OutputParserException as e: print(e) # 打印 output4 的异常消息 # print(parser.get_format_instructions())这个解析器使用 re.search 来查找指示姓名和年龄的模式,展示了对LLM输出变化的适应能力。它还通过 OutputParserException 包含错误处理,并提供格式指令。集成自定义组件LangChain 设计的灵活性,尤其是 LangChain 表达式语言 (LCEL),在于符合基本接口的自定义组件可以与标准组件集成。您可以像内置组件一样将它们连接起来。# 假设 custom_prompt、custom_llm 和 custom_parser 已实例化 # custom_prompt: DynamicStructureChatPrompt 的一个实例 # custom_llm: CustomLoggedChatWrapper 的一个实例(包装了一个真实模型) # custom_parser: RegexNameAgeParser 的一个实例 # 您可以将自定义组件链接在一起: # chain = custom_prompt | custom_llm | custom_parser # 或者混合标准组件和自定义组件: from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate # standard_llm = ChatOpenAI(model="gpt-3.5-turbo") # standard_parser = StrOutputParser() # 混合示例:自定义提示词、标准 LLM、自定义解析器 # mixed_chain_1 = custom_prompt | standard_llm | custom_parser # 混合示例:标准提示词、自定义 LLM、标准解析器 # standard_prompt = ChatPromptTemplate.from_messages([ # ("system", "您是一个有用的助手。"), # ("human", "{问题}") # ]) # mixed_chain_2 = standard_prompt | custom_llm | standard_parser # result = mixed_chain_1.invoke({"user_query": "Provide details for Eve, Age: 22."}) # print(result) # 可能输出: {'name': 'Eve', 'age': 22} 这种组合性让您能精确地针对LLM交互逻辑中需要定制的部分,而无需重写整个链。从标准组件开始,并随着具体需求的出现逐步引入自定义组件。请记住,充分的测试是必不可少的,特别是对于处理可能不可预测的LLM输出的自定义输出解析器。通过利用自定义组件,您可以获得构建精密、可靠和生产就绪的 LangChain 应用程序所需的精细控制。