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

大模型做导师之方案版本比较

article 2025/2/8 22:08:39

背景：在阅读lightRAG项目时，利用LLM辅助进行理解，当询问LLM如何自定义一个符合项目要求的大模型调用函数时，LLM给出了两个不同的版本。借此想提升一下自己的编程质量，于是让LLM对两个版本进行点评比较。

实现建议
基础版本：

from typing import Optional, List
import aiohttp

async 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: str
    api_base: str
    model: str
    max_retries: int = 3
    timeout: int = 30

class CustomLLM:
    def __init__(self, config: LLMConfig):
        self.config = config
        self._session = None
        
    async def __aenter__(self):
        self._session = aiohttp.ClientSession()
        return self
        
    async 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 = None
    
    async def __aenter__(self):
        self._session = aiohttp.ClientSession()  # 复用会话
        return self
        
    async 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: str
    api_base: str
    model: str
    max_retries: int = 3
    timeout: int = 30

class 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_exponential

class 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:
            # 实现调用逻辑
            pass
        except aiohttp.ClientError as e:
            logger.error(f"网络错误: {e}")
            raise
        except 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 prompt
        
    async 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 = 0
    total_tokens: int = 0
    total_time: float = 0.0

class 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 result
        finally:
            self.metrics.total_calls += 1
            self.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}")