2.1 LLM集成与调用

2.1 LLM集成与调用 — LangChain框架精通 大语言模型集成

本节导读：学习如何在LangChain中集成和调用不同供应商的大语言模型，掌握API配置、参数调优和性能优化的完整技能

学习目标

• 掌握LangChain中多种LLM的集成方法
• 理解不同模型提供商的API配置和限制
• 学会优化模型参数以提高输出质量
• 掌握批量处理和异步调用技巧
• 了解成本控制和性能监控方法

2.1.1 LLM集成基础

🤖 什么是LLM集成

LLM（Large Language Model）集成是指在LangChain框架中使用各种大语言模型的过程。LangChain提供了统一的接口来支持多个模型提供商，包括OpenAI、Anthropic、Google等。

集成的核心价值

🔧 集成架构

LangChain的LLM集成采用分层架构：

# 基础接口层
from langchain.llms.base import BaseLLM
from langchain.chat_models.base import BaseChatModel

# 具体实现层
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google import ChatGoogle

# 统一调用方式
class UnifiedLLM:
    def __init__(self, provider: str = "openai"):
        self.provider = provider
        self.llm = self._create_llm()
    
    def _create_llm(self):
        providers = {
            "openai": ChatOpenAI(),
            "anthropic": ChatAnthropic(),
            "google": ChatGoogle()
        }
        return providers.get(self.provider)
    
    def invoke(self, prompt: str):
        return self.llm.invoke(prompt)
​

2.1.2 OpenAI集成详解

🔑 API配置

环境变量配置

# 设置OpenAI API密钥
export OPENAI_API_KEY="sk-your-api-key-here"

# 设置代理（如果需要）
export HTTP_PROXY="http://proxy-server:port"
export HTTPS_PROXY="https://proxy-server:port"

# 设置超时参数
export OPENAI_TIMEOUT=60
export OPENAI_MAX_RETRIES=3
​

Python代码配置

import os
from langchain_openai import ChatOpenAI

# 方法1：环境变量（推荐）
os.environ["OPENAI_API_KEY"] = "sk-your-api-key-here"
llm = ChatOpenAI()

# 方法2：直接传入
llm = ChatOpenAI(
    api_key="sk-your-api-key-here",
    model="gpt-4"
)

# 方法3：配置文件
config = {
    "api_key": "sk-your-api-key-here",
    "model": "gpt-4",
    "temperature": 0.7
}
llm = ChatOpenAI(**config)
​

🎯 模型选择

OpenAI模型对比

模型选择策略

def select_openai_model(task_type: str, budget: float = None):
    """根据任务类型选择合适的OpenAI模型"""
    
    model_config = {
        "complex_reasoning": {
            "model": "gpt-4",
            "temperature": 0.3,
            "max_tokens": 2000
        },
        "general_chat": {
            "model": "gpt-3.5-turbo",
            "temperature": 0.7,
            "max_tokens": 1000
        },
        "code_generation": {
            "model": "gpt-4-turbo",
            "temperature": 0.1,
            "max_tokens": 1500
        },
        "creative_writing": {
            "model": "gpt-4",
            "temperature": 0.8,
            "max_tokens": 2000
        }
    }
    
    config = model_config.get(task_type, model_config["general_chat"])
    
    # 预算限制检查
    if budget:
        cost_per_1k_tokens = get_model_cost(config["model"])
        estimated_cost = (config["max_tokens"] / 1000) * cost_per_1k_tokens
        
        if estimated_cost > budget:
            # 选择更经济的模型
            config["model"] = "gpt-3.5-turbo"
            config["temperature"] = config["temperature"]
            config["max_tokens"] = min(config["max_tokens"], 1000)
    
    return ChatOpenAI(**config)

def get_model_cost(model: str) -> float:
    """获取模型每1000个token的成本"""
    cost_map = {
        "gpt-4": 0.03,
        "gpt-4-turbo": 0.01,
        "gpt-3.5-turbo": 0.0015
    }
    return cost_map.get(model, 0.01)
​

⚙️ 高级配置

参数详解

from langchain_openai import ChatOpenAI

# 完整参数配置
advanced_llm = ChatOpenAI(
    # 模型选择
    model="gpt-4-1106-preview",
    
    # 创造性控制
    temperature=0.3,           # 0-2，越高越随机
    top_p=0.9,                # 核心采样
    frequency_penalty=0.1,    # 减少重复
    presence_penalty=0.2,     # 鼓励多样性
    
    # 输出控制
    max_tokens=2000,          # 最大输出长度
    timeout=60,              # 请求超时(秒)
    
    # 重试机制
    retry_attempts=3,        # 重试次数
    retry_delay=2,           # 重试延迟(秒)
    
    # 流式输出
    streaming=True,          # 启用流式输出
    verbose=True            # 详细日志
)
​

批量调用优化

import asyncio
from typing import List, Dict, Any

class OpenAIBatchProcessor:
    """OpenAI批量处理器"""
    
    def __init__(self, llm: ChatOpenAI, batch_size: int = 5, max_concurrent: int = 3):
        self.llm = llm
        self.batch_size = batch_size
        self.max_concurrent = max_concurrent
    
    async def async_batch_call(self, prompts: List[str]) -> List[str]:
        """异步批量调用"""
        semaphore = asyncio.Semaphore(self.max_concurrent)
        tasks = []
        
        for i in range(0, len(prompts), self.batch_size):
            batch = prompts[i:i + self.batch_size]
            task = self._process_batch_with_semaphore(batch, semaphore)
            tasks.append(task)
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # 展平结果
        flat_results = []
        for result in results:
            if isinstance(result, Exception):
                flat_results.append(f"处理错误: {str(result)}")
            else:
                flat_results.extend(result)
        
        return flat_results
    
    async def _process_batch_with_semaphore(self, batch: List[str], semaphore: asyncio.Semaphore):
        """使用信号量控制并发"""
        async with semaphore:
            return await self._process_batch(batch)
    
    async def _process_batch(self, batch: List[str]) -> List[str]:
        """处理单个批次"""
        try:
            # 创建批量请求
            responses = await asyncio.gather(*[
                self.llm.ainvoke(prompt) for prompt in batch
            ])
            
            return [response.content for response in responses]
            
        except Exception as e:
            return [f"批次错误: {str(e)}" for _ in batch]

# 使用示例
llm = ChatOpenAI()
processor = OpenAIBatchProcessor(llm)

# 同步接口
def batch_sync_call(prompts: List[str]) -> List[str]:
    """同步批量调用"""
    loop = asyncio.get_event_loop()
    return loop.run_until_complete(processor.async_batch_call(prompts))
​

2.1.3 Anthropic Claude集成

🔑 Claude API配置

环境变量设置

# Anthropic API密钥
export ANTHROPIC_API_KEY="sk-ant-api03-..."

# Claude模型配置
export ANTHROPIC_MODEL="claude-3-sonnet-20240229"
export ANTHROPIC_MAX_TOKENS=1000
​

Python配置

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import SystemMessage, HumanMessage

# 基础配置
claude = ChatAnthropic(
    model="claude-3-sonnet-20240229",
    temperature=0.5,
    max_tokens=1000
)

# 高级配置
claude_advanced = ChatAnthropic(
    model="claude-3-opus-20240229",
    temperature=0.3,
    max_tokens=2000,
    timeout=60,
    retry_attempts=3
)
​

🎯 Claude模型选择

Claude模型对比

动态模型选择

def select_claude_model(task_complexity: str, speed_requirement: str = "normal"):
    """根据任务复杂度选择Claude模型"""
    
    model_selection = {
        "simple": {
            "normal": "claude-3-haiku-20240307",
            "fast": "claude-3-haiku-20240307"
        },
        "moderate": {
            "normal": "claude-3-sonnet-20240229",
            "fast": "claude-3-sonnet-20240229"
        },
        "complex": {
            "normal": "claude-3-sonnet-20240229",
            "fast": "claude-3-haiku-20240307"  # 快速但降低质量
        }
    }
    
    return model_selection.get(task_complexity, {}).get(speed_requirement, "claude-3-sonnet-20240229")

# 使用示例
model = select_claude_model("complex", "normal")
claude = ChatAnthropic(model=model, temperature=0.3)
​

💬 对话历史管理

系统消息配置

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

class ClaudeConversationManager:
    """Claude对话管理器"""
    
    def __init__(self, system_prompt: str = ""):
        self.system_prompt = system_prompt
        self.conversation_history = []
    
    def add_message(self, role: str, content: str):
        """添加消息到对话历史"""
        if role == "system":
            message = SystemMessage(content=content)
        elif role == "human":
            message = HumanMessage(content=content)
        elif role == "ai":
            message = AIMessage(content=content)
        else:
            raise ValueError(f"未知角色: {role}")
        
        self.conversation_history.append(message)
    
    def get_context(self, max_messages: int = 10) -> List:
        """获取上下文消息"""
        # 获取最近的消息
        recent_messages = self.conversation_history[-max_messages:]
        
        # 添加系统消息
        if self.system_prompt:
            return [SystemMessage(content=self.system_prompt)] + recent_messages
        return recent_messages
    
    def get_response(self, user_input: str, claude: ChatAnthropic):
        """获取Claude响应"""
        # 添加用户消息
        self.add_message("human", user_input)
        
        # 获取上下文
        context = self.get_context()
        
        # 调用Claude
        response = claude.invoke(context)
        
        # 添加AI回复到历史
        self.add_message("ai", response.content)
        
        return response.content

# 使用示例
manager = ClaudeConversationManager(
    system_prompt="你是一个专业的数据科学家，擅长用通俗易懂的方式解释复杂概念。"
)

claude = ChatAnthropic(model="claude-3-sonnet-20240229")

# 对话
response = manager.get_response("什么是过拟合？", claude)
print(response)
​

多轮对话优化

def optimize_conversation_length(messages: List, max_context_length: int = 100000):
    """优化对话长度以避免上下文溢出"""
    
    total_length = sum(len(str(msg.content)) for msg in messages)
    
    if total_length > max_context_length:
        # 移除最旧的对话，保留最近的
        sorted_messages = sorted(
            enumerate(messages),
            key=lambda x: x[0],  # 按索引排序
            reverse=True  # 保留最新的
        )
        
        # 计算需要移除的消息数量
        removed_count = 0
        current_length = total_length
        
        for idx, message in sorted_messages:
            if current_length <= max_context_length:
                break
            current_length -= len(str(message.content))
            removed_count += 1
        
        # 返回优化后的消息列表
        return [msg for msg in messages if msg not in [m[1] for m in sorted_messages[:removed_count]]]
    
    return messages
​

2.1.4 Google Gemini集成

🔑 Gemini API配置

基础配置

from langchain_google import ChatGoogle

# Gemini Pro配置
gemini_pro = ChatGoogle(
    model="gemini-pro",
    temperature=0.7,
    max_output_tokens=2048,
    top_p=0.8,
    top_k=40
)

# Gemini Ultra配置
gemini_ultra = ChatGoogle(
    model="gemini-ultra",
    temperature=0.3,
    max_output_tokens=4096,
    top_p=0.8,
    top_k=32
)
​

环境变量

# Google API密钥
export GOOGLE_API_KEY="your-google-api-key"

# 模型配置
export GOOGLE_MODEL="gemini-pro"
export GOOGLE_TEMPERATURE=0.7
​

🎯 多模态支持

文本+图像处理

from langchain_google import ChatGoogle
from langchain_core.messages import HumanMessage
from langchain_core.documents import Document

class GeminiMultimodalProcessor:
    """Gemini多模态处理器"""
    
    def __init__(self, model: str = "gemini-pro"):
        self.model = ChatGoogle(
            model=model,
            temperature=0.3,
            max_output_tokens=2048
        )
    
    def process_text_and_image(self, text: str, image_data: bytes, image_type: str = "jpeg"):
        """处理文本和图像"""
        
        # 编码图像数据
        import base64
        encoded_image = base64.b64encode(image_data).decode()
        
        # 创建消息
        message = HumanMessage(
            content=[
                {
                    "type": "text",
                    "text": text
                },
                {
                    "type": "image_url",
                    "image_url": f"data:image/{image_type};base64,{encoded_image}"
                }
            ]
        )
        
        # 调用Gemini
        response = self.model.invoke([message])
        return response.content
    
    def analyze_image_with_text(self, image_path: str, analysis_prompt: str):
        """分析图像并配合文本"""
        
        # 读取图像
        with open(image_path, "rb") as image_file:
            image_data = image_file.read()
        
        return self.process_text_and_image(
            text=analysis_prompt,
            image_data=image_data
        )

# 使用示例
processor = GeminiMultimodalProcessor()

# 分析产品图片
analysis = processor.analyze_image_with_text(
    image_path="product.jpg",
    analysis_prompt="请分析这张产品图片，描述其主要特征和设计风格。"
)
print(analysis)
​

🌟 Gemini功能特性

代码生成和执行

class GeminiCodeAssistant:
    """Gemini代码助手"""
    
    def __init__(self):
        self.model = ChatGoogle(
            model="gemini-pro",
            temperature=0.2,
            max_output_tokens=2000
        )
    
    def generate_code(self, language: str, task: str, requirements: dict):
        """生成代码"""
        
        prompt = f"""
        请生成{language}代码来实现以下任务：
        
        任务：{task}
        
        要求：{requirements}
        
        请提供：
        1. 完整的代码实现
        2. 代码注释说明
        3. 使用示例
        """
        
        response = self.model.invoke(prompt)
        return response.content
    
    def review_code(self, code: str, language: str):
        """代码审查"""
        
        prompt = f"""
        请审查以下{language}代码：
        
        ```{language}
        {code}
        ```
        
        请提供：
        1. 代码质量评价
        2. 潜在问题识别
        3. 改进建议
        4. 优化方案
        """
        
        response = self.model.invoke(prompt)
        return response.content
    
    def debug_code(self, code: str, error_message: str, language: str):
        """代码调试"""
        
        prompt = f"""
        请帮助调试以下{language}代码：
        
        代码：
        ```{language}
        {code}
        ```
        
        错误信息：{error_message}
        
        请提供：
        1. 错误原因分析
        2. 修复建议
        3. 修正后的代码
        """
        
        response = self.model.invoke(prompt)
        return response.content

# 使用示例
code_assistant = GeminiCodeAssistant()

# 生成Python代码
python_code = code_assistant.generate_code(
    language="Python",
    task="数据分析和可视化",
    requirements={"pandas": True, "matplotlib": True, "统计功能": True}
)
print(python_code)
​

2.1.5 LLM性能优化

⚡ 缓存策略

LLM缓存设置

from langchain.cache import SQLiteCache
from langchain.globals import set_llm_cache
import functools
import sqlite3

def setup_llm_cache(cache_path: str = "./cache/llm_cache.db"):
    """设置LLM缓存"""
    # 确保缓存目录存在
    import os
    os.makedirs(os.path.dirname(cache_path), exist_ok=True)
    
    # 创建缓存数据库
    cache = SQLiteCache(database_path=cache_path)
    set_llm_cache(cache)
    
    return cache

# 设置缓存
cache = setup_llm_cache()

@functools.lru_cache(maxsize=128)
def cached_llm_call(prompt_hash: str, model: str, **kwargs):
    """带缓存的LLM调用"""
    # 这里可以添加实际的LLM调用逻辑
    pass
​

智能缓存策略

class SmartLLMCache:
    """智能LLM缓存"""
    
    def __init__(self, cache_path: str = "./cache/smart_cache.db"):
        self.cache_path = cache_path
        self.setup_cache()
    
    def setup_cache(self):
        """设置缓存"""
        cache = SQLiteCache(database_path=self.cache_path)
        set_llm_cache(cache)
    
    def get_cache_key(self, prompt: str, model: str, params: dict) -> str:
        """生成缓存键"""
        import hashlib
        
        # 创建包含所有参数的字符串
        param_str = f"{prompt}_{model}_{sorted(params.items())}"
        return hashlib.md5(param_str.encode()).hexdigest()
    
    def should_cache(self, prompt: str, model: str) -> bool:
        """判断是否应该缓存"""
        # 根据提示词长度和复杂度决定
        if len(prompt) < 50:
            return False
        
        # 排除包含敏感信息的问题
        sensitive_keywords = ["密码", "密钥", "API", "private"]
        return not any(keyword in prompt for keyword in sensitive_keywords)
    
    def smart_invoke(self, llm, prompt: str, **kwargs):
        """智能调用（带缓存判断）"""
        if not self.should_cache(prompt, kwargs.get("model", "gpt-4")):
            # 不缓存，直接调用
            return llm.invoke(prompt, **kwargs)
        
        # 检查缓存
        cache_key = self.get_cache_key(prompt, kwargs.get("model", "gpt-4"), kwargs)
        cached_result = self.get_cached_result(cache_key)
        
        if cached_result:
            return cached_result
        
        # 调用LLM并缓存结果
        result = llm.invoke(prompt, **kwargs)
        self.cache_result(cache_key, result)
        
        return result

# 使用示例
smart_cache = SmartLLMCache()
​

🔄 批量处理优化

高效批量处理

import asyncio
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Callable, Any

class BatchProcessor:
    """高级批量处理器"""
    
    def __init__(self, max_workers: int = 5, batch_size: int = 10):
        self.max_workers = max_workers
        self.batch_size = batch_size
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
    
    def process_batches(self, items: List[Any], process_func: Callable, **kwargs) -> List[Any]:
        """批量处理项目"""
        results = []
        
        # 分批处理
        for i in range(0, len(items), self.batch_size):
​