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

LangChain实战（二十三）：性能优化与生产环境最佳实践

news 2025/9/9 6:13:06

本文是《LangChain实战课》系列的第二十三篇，将深入探讨LangChain应用在生产环境中需要考虑的各种关键问题，包括性能优化、错误处理、成本控制、监控日志等，帮助你构建稳定高效的AI应用系统。

一、前言

在前面的文章中，我们学习了如何使用LangChain构建各种AI应用并将其部署为API服务。然而，将应用从开发环境迁移到生产环境需要考虑更多因素：性能、稳定性、成本、可观测性等。本文将分享在生产环境中运行LangChain应用的最佳实践和优化策略。

二、缓存策略优化

1. 请求级别缓存

from langchain.cache import SQLiteCache, RedisCache
import langchain
import sqlite3# 使用SQLite缓存
langchain.llm_cache = SQLiteCache(database_path=".langchain.db")# 或者使用Redis缓存（推荐用于生产环境）
from redis import Redis
langchain.llm_cache = RedisCache(redis_=Redis(host='localhost', port=6379))# 自定义缓存策略
from langchain.cache import BaseCache
from typing import Optional, Any
import hashlib
import jsonclass CustomCache(BaseCache):def __init__(self):self.cache = {}def lookup(self, prompt: str, llm_string: str) -> Optional[Any]:key = self._generate_key(prompt, llm_string)return self.cache.get(key)def update(self, prompt: str, llm_string: str, return_val: Any) -> None:key = self._generate_key(prompt, llm_string)self.cache[key] = return_valdef _generate_key(self, prompt: str, llm_string: str) -> str:return hashlib.md5(f"{prompt}{llm_string}".encode()).hexdigest()# 使用自定义缓存
langchain.llm_cache = CustomCache()

2. 语义缓存

from langchain.globals import set_llm_cache
from langchain.cache import SemanticCache
from langchain.embeddings import OpenAIEmbeddings# 基于语义相似度的缓存
set_llm_cache(SemanticCache(embedding=OpenAIEmbeddings(),similarity_threshold=0.8  # 相似度阈值)
)

3. 向量存储缓存

from langchain.storage import LocalFileStore
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.cache import SQLiteCache# 创建基于向量存储的缓存系统
store = LocalFileStore("./cache/")
embeddings = OpenAIEmbeddings()# 在链中显式使用缓存
from langchain.chains import LLMChain
from langchain.callbacks import ManagerCallbackHandlerclass CachedChain:def __init__(self, chain):self.chain = chainself.cache = {}async def arun(self, input_dict, **kwargs):cache_key = self._generate_cache_key(input_dict)if cache_key in self.cache:return self.cache[cache_key]result = await self.chain.arun(input_dict, **kwargs)self.cache[cache_key] = resultreturn resultdef _generate_cache_key(self, input_dict):import jsonreturn hashlib.md5(json.dumps(input_dict, sort_keys=True).encode()).hexdigest()

三、异步处理优化

1. 异步链设计

import asyncio
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain.chat_models import ChatOpenAIclass AsyncChainProcessor:def __init__(self, max_concurrent=10):self.semaphore = asyncio.Semaphore(max_concurrent)async def process_batch(self, inputs):"""批量处理输入，控制并发数"""tasks = []for input_data in inputs:task = self._process_single(input_data)tasks.append(task)results = await asyncio.gather(*tasks, return_exceptions=True)return resultsasync def _process_single(self, input_data):"""处理单个输入，带有信号量控制"""async with self.semaphore:try:# 模拟异步处理await asyncio.sleep(0.1