LangChain自定义LLM实战从Python函数到智能接口的魔法变形记在AI应用开发的世界里大型语言模型(LLM)正以前所未有的速度改变着技术格局。但你是否想过那些看似神秘的AI接口背后其实隐藏着一个惊人的简单本质今天我们将揭开这层神秘面纱展示如何将最普通的Python函数变身为符合LangChain规范的LLM接口——这个过程就像给灰姑娘穿上水晶鞋一样神奇。1. 重新认识LLM文本函数的华丽外衣当我们剥开大型语言模型复杂的外壳其核心本质令人惊讶地简单一个接收文本并返回文本的函数。这种极简的认知模型为我们自定义LLM提供了绝佳的切入点。def simplest_llm(prompt: str) - str: return 这是对{}的标准回复.format(prompt)这个仅有三行的Python函数已经具备了LLM最基础的特征。但在实际应用中我们需要考虑更多因素上下文管理如何处理多轮对话的历史记录参数控制温度(temperature)、top_p等生成参数停止条件指定特定标记来终止生成错误处理网络请求失败或模型加载异常LangChain的LLM基类正是围绕这些实际需求构建的抽象层。它提供了一套标准接口让各种不同的模型实现可以无缝集成到更大的应用中。2. 构建你的第一个自定义LLM类让我们从一个实用的天气查询LLM开始逐步构建符合LangChain规范的实现。这个例子将展示如何将简单的功能封装成专业接口。2.1 基础骨架继承LLM基类首先我们需要从LangChain导入基础类并建立继承关系from langchain.llms.base import LLM from typing import Optional, List, Dict, Any class WeatherLLM(LLM): property def _llm_type(self) - str: return weather def _call( self, prompt: str, stop: Optional[List[str]] None, **kwargs: Any, ) - str: # 核心逻辑将在这里实现 pass这个骨架已经包含了自定义LLM的必要元素_llm_type属性标识你的LLM类型_call方法处理实际请求的核心方法2.2 实现天气查询逻辑现在让我们填充具体的天气查询功能。我们将使用一个简单的字典来模拟天气APIdef _call(self, prompt: str, stop: Optional[List[str]] None, **kwargs) - str: # 简单解析城市名称 if 天气 in prompt: city prompt.replace(天气, ).strip() else: city prompt.strip() # 模拟天气数据库 weather_data { 北京: 晴25°C湿度40%, 上海: 多云28°C湿度65%, 广州: 雷阵雨30°C湿度80%, } # 获取天气信息 if city in weather_data: response f{city}的天气情况{weather_data[city]} else: response f抱歉找不到{city}的天气信息 # 处理停止标记 if stop is not None: for token in stop: if token in response: response response.split(token)[0] return response2.3 完整实现与测试现在我们已经有了一个功能完整的WeatherLLM类。让我们看看如何使用它weather_llm WeatherLLM() # 简单查询 print(weather_llm(北京天气)) # 输出北京的天气情况晴25°C湿度40% # 带停止标记的查询 print(weather_llm(上海天气, stop[湿度])) # 输出上海的天气情况多云28°C这个简单的例子展示了LangChain LLM封装的核心思想将特定功能适配到标准接口。虽然我们的天气查询很简单但架构与复杂模型完全相同。3. 高级技巧增强你的自定义LLM基础实现已经可以工作但要让自定义LLM真正强大我们需要添加更多专业特性。3.1 参数化配置优秀的LLM实现应该允许运行时配置。让我们为WeatherLLM添加温度参数class WeatherLLM(LLM): temperature: float 0.7 def __init__(self, temperature: float 0.7, **kwargs: Any): super().__init__(**kwargs) self.temperature temperature def _call(self, prompt: str, stop: Optional[List[str]] None, **kwargs) - str: # 根据温度参数调整回答风格 base_response self._get_base_response(prompt) if self.temperature 0.5: return base_response (今日天气不错适合外出) else: return base_response3.2 异步支持现代应用往往需要异步处理。LangChain提供了异步方法的重写支持async def _acall( self, prompt: str, stop: Optional[List[str]] None, **kwargs: Any, ) - str: # 这里可以实现真正的异步请求 # 本例中我们仅做简单模拟 import asyncio await asyncio.sleep(0.1) # 模拟网络延迟 return self._call(prompt, stop, **kwargs)3.3 流式输出对于长时间运行的生成任务流式输出能显著改善用户体验def _stream( self, prompt: str, stop: Optional[List[str]] None, **kwargs: Any, ) - Iterator[str]: # 模拟流式生成天气报告的每个单词 full_response self._call(prompt, stop, **kwargs) for word in full_response.split(): yield word time.sleep(0.1)4. 实战本地文本处理LLM让我们看一个更实用的例子——将本地文本处理脚本封装为LLM。假设我们有一个处理Markdown的Python函数def markdown_to_html(md_text: str) - str: import markdown return markdown.markdown(md_text)4.1 创建MarkdownLLM类我们可以轻松地将这个函数包装成LLMclass MarkdownLLM(LLM): property def _llm_type(self) - str: return markdown-processor def _call(self, prompt: str, stop: Optional[List[str]] None, **kwargs) - str: import markdown html markdown.markdown(prompt) if stop is not None: for token in stop: html html.split(token)[0] return html4.2 添加高级功能让我们增强这个Markdown处理器class AdvancedMarkdownLLM(MarkdownLLM): extensions: List[str] [extra, tables] css_class: Optional[str] None def _call(self, prompt: str, stop: Optional[List[str]] None, **kwargs) - str: import markdown html markdown.markdown(prompt, extensionsself.extensions) if self.css_class: html fdiv class{self.css_class}{html}/div if stop: html self._apply_stop_tokens(html, stop) return html4.3 集成到LangChain生态现在这个MarkdownLLM可以像任何其他LangChain LLM一样使用了from langchain.chains import LLMChain from langchain.prompts import PromptTemplate md_llm AdvancedMarkdownLLM( extensions[tables, fenced_code], css_classmarkdown-content ) template 将以下Markdown转换为HTML: {input} prompt PromptTemplate.from_template(template) chain LLMChain(llmmd_llm, promptprompt) result chain.run( # 标题 这是一个*示例*文本。 - 列表项1 - 列表项2 ) print(result)5. 设计模式构建可扩展的LLM封装当封装更复杂的自定义LLM时遵一些设计模式可以使代码更健壮、更易维护。5.1 适配器模式适配器模式非常适合将不同接口的模型统一到LangChain标准class ExternalModelAdapter(LLM): def __init__(self, external_model: Any): super().__init__() self.model external_model def _call(self, prompt: str, stop: Optional[List[str]] None, **kwargs) - str: # 将LangChain调用转换为外部模型期望的格式 external_input self._convert_input(prompt) external_response self.model.generate(external_input) return self._convert_output(external_response)5.2 组合模式通过组合多个简单LLM构建更强大的功能class CompositeLLM(LLM): def __init__(self, preprocessor: LLM, main_llm: LLM, postprocessor: LLM): super().__init__() self.preprocessor preprocessor self.main_llm main_llm self.postprocessor postprocessor def _call(self, prompt: str, stop: Optional[List[str]] None, **kwargs) - str: processed_input self.preprocessor(prompt) main_output self.main_llm(processed_input) final_output self.postprocessor(main_output) return final_output5.3 装饰器模式通过装饰器动态添加功能而不修改原有实现def logging_decorator(llm_cls): class LoggedLLM(llm_cls): def _call(self, prompt: str, stop: Optional[List[str]] None, **kwargs) - str: print(fRequest received: {prompt}) start_time time.time() response super()._call(prompt, stop, **kwargs) duration time.time() - start_time print(fResponse generated in {duration:.2f}s) return response return LoggedLLM logging_decorator class MyLoggedLLM(WeatherLLM): pass6. 性能优化与生产就绪当自定义LLM准备投入生产环境时需要考虑性能、可靠性和可观测性。6.1 缓存机制实现简单的请求缓存可以显著减少重复计算from functools import lru_cache class CachedLLM(LLM): lru_cache(maxsize1024) def _call(self, prompt: str, stop: Optional[List[str]] None, **kwargs) - str: # 原始实现 pass6.2 重试逻辑对于依赖外部服务的LLM自动重试机制必不可少from tenacity import retry, stop_after_attempt, wait_exponential class ResilientLLM(LLM): retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10) ) def _call(self, prompt: str, stop: Optional[List[str]] None, **kwargs) - str: # 尝试调用可能失败的外部服务 pass6.3 监控与指标添加监控指标帮助了解LLM在生产环境中的表现from prometheus_client import Counter, Histogram REQUEST_COUNT Counter(llm_requests_total, Total LLM requests) REQUEST_LATENCY Histogram(llm_request_latency_seconds, Request latency) class MonitoredLLM(LLM): def _call(self, prompt: str, stop: Optional[List[str]] None, **kwargs) - str: REQUEST_COUNT.inc() start_time time.time() try: response self._real_call(prompt, stop, **kwargs) return response finally: latency time.time() - start_time REQUEST_LATENCY.observe(latency)7. 测试与验证策略确保自定义LLM的可靠性需要全面的测试策略。7.1 单元测试示例使用pytest测试WeatherLLM的基本功能def test_weather_llm_basic(): llm WeatherLLM() response llm(北京天气) assert 北京 in response assert 晴 in response or 雨 in response or 云 in response def test_weather_llm_stop_tokens(): llm WeatherLLM() response llm(上海天气, stop[湿度]) assert 湿度 not in response7.2 集成测试验证LLM在LangChain生态中的集成情况def test_llm_in_chain(): llm MarkdownLLM() template Convert this markdown header: {input} prompt PromptTemplate.from_template(template) chain LLMChain(llmllm, promptprompt) result chain.run(input# Hello World) assert h1Hello World/h1 in result7.3 性能基准评估LLM的响应时间和吞吐量import timeit def benchmark_llm(): llm WeatherLLM() def test_call(): llm(北京天气) duration timeit.timeit(test_call, number100) print(fAverage call time: {duration/100:.4f}s)