背景： 在阅读lightRAG项目时，利用LLM辅助进行理解，当询问LLM如何自定义一个符合项目要求的大模型调用函数时，LLM给出了两个不同的版本。借此想提升一下自己的编程质量，于是让LLM对两个版本进行点评比较。
实现建议
基础版本：
from typing import Optional, List
import aiohttpasync def simple_llm_func(prompt: str,system_prompt: Optional[str] = None,history_messages: List[dict] = [],keyword_extraction: bool = False,**kwargs
) -> str:async with aiohttp.ClientSession() as session:async with session.post("你的API地址",json={"prompt": prompt,"system_prompt": system_prompt,"history": history_messages,"is_keyword_extraction": keyword_extraction},headers={"Authorization": "你的API密钥"}) as response:result = await response.json()return result["text"]
进阶版本：
from dataclasses import dataclass
from typing import Optional, List
import aiohttp
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential@dataclass
class LLMConfig:api_key: strapi_base: strmodel: strmax_retries: int = 3timeout: int = 30class CustomLLM:def __init__(self, config: LLMConfig):self.config = configself._session = Noneasync def __aenter__(self):self._session = aiohttp.ClientSession()return selfasync def __aexit__(self, exc_type, exc_val, exc_tb):if self._session:await self._session.close()@retry(stop=stop_after_attempt(3),wait=wait_exponential(multiplier=1, min=4, max=10))async def __call__(self,prompt: str,system_prompt: Optional[str] = None,history_messages: List[dict] = [],keyword_extraction: bool = False,**kwargs) -> str:# 实现 API 调用逻辑pass# 使用方式
llm_config = LLMConfig(api_key="你的API密钥",api_base="API地址",model="模型名称"
)
llm = CustomLLM(llm_config))
你想了解哪个具体部分的实现细节？
请解释你给出的进阶和基础相比的优势在哪里？
让我详细解释进阶版本相比基础版本的主要优势：
1. 会话管理优化
基础版本：
async def simple_llm_func(prompt: str, **kwargs) -> str:async with aiohttp.ClientSession() as session: # 每次调用都创建新会话async with session.post(...) as response:return (await response.json())["text"]
进阶版本：
class CustomLLM:def __init__(self, config: LLMConfig):self._session = Noneasync def __aenter__(self):self._session = aiohttp.ClientSession() # 复用会话return selfasync def __aexit__(self, exc_type, exc_val, exc_tb):if self._session:await self._session.close()
优势：
避免频繁创建/销毁 HTTP 会话
支持连接池复用
更好的性能和资源管理
2. 配置管理
基础版本：
async def simple_llm_func(prompt: str,system_prompt: Optional[str] = None,**kwargs
) -> str:# 配置直接硬编码或通过参数传递headers = {"Authorization": "你的API密钥"}
进阶版本：
@dataclass
class LLMConfig:api_key: strapi_base: strmodel: strmax_retries: int = 3timeout: int = 30class CustomLLM:def __init__(self, config: LLMConfig):self.config = config
优势：
配置集中管理
类型安全
更容易修改和维护
支持不同环境配置
3. 错误处理和重试机制
基础版本：
async def simple_llm_func(prompt: str, **kwargs) -> str:async with aiohttp.ClientSession() as session:async with session.post(...) as response:# 简单的错误处理if response.status != 200:raise Exception("API调用失败")
进阶版本：
from tenacity import retry, stop_after_attempt, wait_exponentialclass CustomLLM:@retry(stop=stop_after_attempt(3),wait=wait_exponential(multiplier=1, min=4, max=10))async def __call__(self, prompt: str, **kwargs) -> str:try:# 实现调用逻辑passexcept aiohttp.ClientError as e:logger.error(f"网络错误: {e}")raiseexcept Exception as e:logger.error(f"未知错误: {e}")raise
优势：
自动重试机制
指数退避策略
更完善的错误处理
详细的日志记录
4. 扩展性
基础版本：
async def simple_llm_func(prompt: str, **kwargs) -> str:# 功能都在一个函数里，难以扩展pass
进阶版本：
class CustomLLM:async def preprocess(self, prompt: str) -> str:# 预处理逻辑return promptasync def postprocess(self, response: dict) -> str:# 后处理逻辑return response["text"]async def __call__(self, prompt: str, **kwargs) -> str:prompt = await self.preprocess(prompt)response = await self._make_request(prompt)return await self.postprocess(response)
优势：
模块化设计
易于添加新功能
支持前处理和后处理
更好的代码组织
5. 性能监控
进阶版本可以添加性能监控：
from time import time
from dataclasses import dataclass@dataclass
class APIMetrics:total_calls: int = 0total_tokens: int = 0total_time: float = 0.0class CustomLLM:def __init__(self, config: LLMConfig):self.metrics = APIMetrics()async def __call__(self, prompt: str, **kwargs) -> str:start_time = time()try:result = await self._make_request(prompt)return resultfinally:self.metrics.total_calls += 1self.metrics.total_time += time() - start_time
优势：
调用统计
性能分析
成本追踪
问题诊断
6. 实际使用示例
# 配置初始化
config = LLMConfig(api_key="your_api_key",api_base="https://api.example.com",model="gpt-4",max_retries=3,timeout=30
)# 创建 LLM 实例
async with CustomLLM(config) as llm:try:# 单次调用response = await llm(prompt="你的问题",system_prompt="系统提示",temperature=0.7)# 批量调用tasks = [llm(prompt) for prompt in prompts]responses = await asyncio.gather(*tasks)# 查看指标print(f"总调用次数: {llm.metrics.total_calls}")print(f"平均响应时间: {llm.metrics.total_time/llm.metrics.total_calls:.2f}s")except Exception as e:logger.error(f"LLM调用失败: {e}")
这种设计更适合:
生产环境部署
大规模调用
需要监控和分析
需要高可靠性
团队协作开发
