显示页面讨论过去修订反向链接回到顶部 本页面只读。您可以查看源文件,但不能更改它。如果您觉得这是系统错误,请联系管理员。 ====== 第二章:核心组件详解 - Models ====== ===== 2.1 模型类型概述 ===== 在LangChain中,语言模型是最核心的组件。LangChain支持三大类模型: ==== 2.1.1 LLMs(基础语言模型) ==== **定义**:接收字符串输入,返回字符串输出的模型。 **特点**: * 传统的文本补全接口 * 通常通过调用 `predict()` 或 `generate()` 方法 * 适合简单的文本生成任务 **代表模型**: * OpenAI: `text-davinci-003`, `gpt-3.5-turbo-instruct` * 开源: `LLaMA`, `Mistral`, `Falcon` **使用场景**: * 文本补全 * 简单的文本转换 * 需要直接控制输入格式的场景 <code python> from langchain.llms import OpenAI # 创建LLM实例 llm = OpenAI( model_name="gpt-3.5-turbo-instruct", temperature=0.7, max_tokens=256 ) # 生成文本 text = "人工智能正在改变" result = llm.predict(text) print(result) </code> ==== 2.1.2 Chat Models(对话模型) ==== **定义**:专为对话场景优化的模型,接收消息列表,返回消息。 **特点**: * 基于消息角色的架构(System、Human、AI) * 更好的对话上下文理解 * 现代主流模型的标准接口 **代表模型**: * OpenAI: `gpt-3.5-turbo`, `gpt-4`, `gpt-4-turbo` * Anthropic: `claude-3-opus`, `claude-3-sonnet`, `claude-3-haiku` * 开源: `LLaMA-2-Chat`, `Vicuna` **使用场景**: * 聊天机器人 * 多轮对话系统 * 需要角色设定的场景 <code python> from langchain.chat_models import ChatOpenAI from langchain.schema import HumanMessage, SystemMessage chat = ChatOpenAI(model="gpt-4") messages = [ SystemMessage(content="你是一个Python专家"), HumanMessage(content="解释什么是装饰器") ] response = chat.predict_messages(messages) print(response.content) </code> ==== 2.1.3 Embedding Models(嵌入模型) ==== **定义**:将文本转换为高维向量(嵌入)的模型。 **特点**: * 输出是数值向量,不是文本 * 捕捉语义信息 * 用于相似度计算和检索 **代表模型**: * OpenAI: `text-embedding-3-small`, `text-embedding-3-large`, `text-embedding-ada-002` * 开源: `sentence-transformers`系列 **使用场景**: * 语义搜索 * 文本聚类 * RAG系统中的文档检索 <code python> from langchain.embeddings import OpenAIEmbeddings embeddings = OpenAIEmbeddings(model="text-embedding-3-small") # 获取文本的向量表示 text = "这是一个测试句子" vector = embeddings.embed_query(text) print(f"向量维度: {len(vector)}") print(f"向量前5个值: {vector[:5]}") # 批量嵌入 texts = ["第一句", "第二句", "第三句"] vectors = embeddings.embed_documents(texts) print(f"批量嵌入数量: {len(vectors)}") </code> ==== 2.1.4 三种模型的对比 ==== | 特性 | LLMs | Chat Models | Embeddings | | 输入 | 字符串 | 消息列表 | 字符串 | | 输出 | 字符串 | 消息对象 | 向量 | | 主要用途 | 文本生成 | 对话 | 语义表示 | | 状态管理 | 无 | 无 | 无 | | 成本 | 按token计费 | 按token计费 | 通常较低 | ---- ===== 2.2 OpenAI 模型集成 ===== ==== 2.2.1 ChatOpenAI 详解 ==== `ChatOpenAI` 是与OpenAI聊天模型交互的主要接口。 === 基础配置参数 === <code python> from langchain_openai import ChatOpenAI llm = ChatOpenAI( # 模型选择 model="gpt-4", # 模型名称 # 生成参数 temperature=0.7, # 随机性 (0-2) max_tokens=None, # 最大输出token数 top_p=1.0, # 核采样概率 frequency_penalty=0.0, # 频率惩罚 (-2 to 2) presence_penalty=0.0, # 存在惩罚 (-2 to 2) # API配置 api_key="your-api-key", # API密钥 base_url=None, # 自定义API端点 timeout=None, # 请求超时时间 max_retries=2, # 最大重试次数 # 其他 streaming=False, # 是否流式输出 n=1, # 生成结果数量 stop=None, # 停止序列 ) </code> === 参数详解 === **1. temperature(温度)** 控制输出的随机性: * `0.0`: 最确定性,总是选择概率最高的token * `0.7`: 平衡值,适合大多数场景 * `1.0+`: 更具创造性,输出更多样 <code python> def demonstrate_temperature(): """演示temperature参数的影响""" prompt = "给一家咖啡店起一个有创意的名字" for temp in [0.0, 0.5, 1.0, 1.5]: llm = ChatOpenAI(temperature=temp) result = llm.predict(prompt) print(f"\nTemperature = {temp}:") print(result) demonstrate_temperature() </code> **2. max_tokens(最大token数)** 限制模型输出的长度: * 一个token约等于4个英文字符或0.75个单词 * 中文通常1-2个字符为一个token * 设置为None时,模型自行决定长度 <code python> # 不同max_tokens的效果 prompt = "简要介绍Python编程语言" for max_tok in [50, 100, 200]: llm = ChatOpenAI(max_tokens=max_tok) result = llm.predict(prompt) print(f"\nmax_tokens={max_tok}:") print(result[:100] + "..." if len(result) > 100 else result) </code> **3. frequency_penalty 和 presence_penalty** 用于减少重复: * `frequency_penalty`: 基于token出现频率的惩罚 * `presence_penalty`: 只要出现过的token都会受到惩罚 * 范围都是 -2.0 到 2.0 <code python> # 减少重复内容的示例 prompt = "请写一个关于'未来城市'的100字描述,不要重复用词" llm_low = ChatOpenAI(frequency_penalty=0.0) llm_high = ChatOpenAI(frequency_penalty=0.8) print("低惩罚:") print(llm_low.predict(prompt)) print("\n高惩罚:") print(llm_high.predict(prompt)) </code> ==== 2.2.2 调用方法详解 ==== ChatOpenAI 提供多种调用方式: <code python> from langchain_openai import ChatOpenAI from langchain.schema import HumanMessage, SystemMessage, AIMessage chat = ChatOpenAI() # 方法1: predict - 最简单的方式 result = chat.predict("你好") print(type(result)) # str # 方法2: predict_messages - 使用消息对象 messages = [ SystemMessage(content="你是一个助手"), HumanMessage(content="你好") ] result = chat.predict_messages(messages) print(type(result)) # AIMessage print(result.content) # 方法3: generate - 批量生成,获取更多元数据 from langchain.schema import Generation batch_messages = [ [HumanMessage(content="你好")], [HumanMessage(content="再见")] ] result = chat.generate(batch_messages) print(f"总token消耗: {result.llm_output['token_usage']}") print(f"第一个结果: {result.generations[0][0].text}") # 方法4: async/await - 异步调用 import asyncio async def async_chat(): result = await chat.apredict("异步测试") print(result) asyncio.run(async_chat()) # 方法5: streaming - 流式输出 chat_stream = ChatOpenAI(streaming=True) for chunk in chat_stream.stream("写一首短诗"): print(chunk.content, end="", flush=True) </code> ==== 2.2.3 模型选择指南 ==== OpenAI提供了多个模型,如何选择? | 模型 | 优点 | 缺点 | 适用场景 | | gpt-3.5-turbo | 快、便宜 | 能力有限 | 简单任务、原型开发 | | gpt-4 | 能力强 | 慢、贵 | 复杂推理、代码生成 | | gpt-4-turbo | 能力最强、支持长文本 | 最贵 | 高级应用、长文档处理 | | gpt-4o | 多模态、快 | 较新 | 需要图像理解的场景 | <code python> def select_model_for_task(task_type: str) -> str: """根据任务类型推荐模型""" models = { "简单问答": "gpt-3.5-turbo", "代码生成": "gpt-4", "复杂推理": "gpt-4-turbo", "创意写作": "gpt-4", "文本分类": "gpt-3.5-turbo", "多语言翻译": "gpt-4", "数学计算": "gpt-4-turbo", } return models.get(task_type, "gpt-3.5-turbo") # 使用示例 tasks = ["简单问答", "代码生成", "创意写作"] for task in tasks: model = select_model_for_task(task) print(f"{task} -> {model}") </code> ==== 2.2.4 OpenAI Embedding 模型 ==== <code python> from langchain_openai import OpenAIEmbeddings # 选择不同的嵌入模型 embeddings_small = OpenAIEmbeddings(model="text-embedding-3-small") # 1536维 embeddings_large = OpenAIEmbeddings(model="text-embedding-3-large") # 3072维 # 对比不同模型 test_text = "自然语言处理是人工智能的重要分支" vector_small = embeddings_small.embed_query(test_text) vector_large = embeddings_large.embed_query(test_text) print(f"text-embedding-3-small: {len(vector_small)} 维") print(f"text-embedding-3-large: {len(vector_large)} 维") # 计算两个句子的相似度 import numpy as np def cosine_similarity(a, b): return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)) text1 = "机器学习是AI的子领域" text2 = "深度学习是机器学习的一种方法" text3 = "今天天气很好" v1 = embeddings_small.embed_query(text1) v2 = embeddings_small.embed_query(text2) v3 = embeddings_small.embed_query(text3) print(f"\n'{text1}' 和 '{text2}' 的相似度: {cosine_similarity(v1, v2):.4f}") print(f"'{text1}' 和 '{text3}' 的相似度: {cosine_similarity(v1, v3):.4f}") </code> ---- ===== 2.3 Anthropic Claude 模型集成 ===== Claude是Anthropic开发的AI助手,以安全性和有用性著称。 ==== 2.3.1 ChatAnthropic 基础使用 ==== <code python> from langchain_anthropic import ChatAnthropic from langchain.schema import HumanMessage, SystemMessage # 创建Claude客户端 claude = ChatAnthropic( model="claude-3-sonnet-20240229", temperature=0.7, max_tokens=1024, anthropic_api_key="your-api-key" # 或使用环境变量 ANTHROPIC_API_KEY ) # 基本对话 messages = [ SystemMessage(content="你是一个专业的技术文档写手"), HumanMessage(content="写一段关于REST API的介绍") ] response = claude.predict_messages(messages) print(response.content) </code> ==== 2.3.2 Claude 模型对比 ==== | 模型 | 描述 | 最佳用途 | | claude-3-opus | 最强大 | 复杂推理、数学、编程 | | claude-3-sonnet | 平衡 | 大多数任务,性价比高 | | claude-3-haiku | 最快 | 简单任务、实时应用 | <code python> def compare_claude_models(): """对比不同Claude模型的表现""" models = [ "claude-3-haiku-20240307", "claude-3-sonnet-20240229", "claude-3-opus-20240229" ] prompt = "用一句话解释量子计算" for model in models: print(f"\n{'='*50}") print(f"模型: {model}") print('='*50) llm = ChatAnthropic(model=model) result = llm.predict(prompt) print(result) compare_claude_models() </code> ==== 2.3.3 Claude 的特殊功能 ==== Claude支持一些特殊功能: <code python> from langchain_anthropic import ChatAnthropic claude = ChatAnthropic(model="claude-3-sonnet-20240229") # 1. 长上下文窗口(200K tokens) long_text = "请总结以下长文档..." + "..." * 100000 response = claude.predict(long_text[:150000]) # Claude支持超长输入 # 2. 结构化提示 structured_prompt = """ Human: 请分析以下产品评价,输出JSON格式: 评价:"这个手机电池续航很好,但屏幕有点暗" 请按以下格式输出: { "sentiment": "positive/negative/mixed", "aspects": [ {"aspect": "...", "sentiment": "...", "comment": "..."} ] } Assistant:""" result = claude.predict(structured_prompt) print(result) </code> ---- ===== 2.4 本地模型集成 ===== ==== 2.4.1 使用 HuggingFace 模型 ==== LangChain可以通过HuggingFace集成各种开源模型。 <code python> from langchain_community.llms import HuggingFacePipeline from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline import torch # 方法1: 直接使用HuggingFace Pipeline def load_local_model(): model_id = "microsoft/DialoGPT-medium" # 可以替换为其他模型 tokenizer = AutoTokenizer.from_pretrained(model_id) model = AutoModelForCausalLM.from_pretrained(model_id) pipe = pipeline( "text-generation", model=model, tokenizer=tokenizer, max_length=100, temperature=0.7 ) llm = HuggingFacePipeline(pipeline=pipe) return llm # 使用本地模型 local_llm = load_local_model() result = local_llm.predict("你好") print(result) </code> ==== 2.4.2 使用 llama.cpp / llama-cpp-python ==== 对于消费级硬件,llama.cpp是很好的选择。 <code python> from langchain_community.llms import LlamaCpp # 加载GGUF格式的模型 llm = LlamaCpp( model_path="path/to/llama-2-7b-chat.Q4_K_M.gguf", n_ctx=2048, # 上下文窗口大小 n_gpu_layers=1, # 使用GPU的层数 temperature=0.7, max_tokens=512, verbose=True ) # 生成文本 result = llm.predict("解释什么是机器学习") print(result) </code> ==== 2.4.3 使用 Ollama ==== Ollama让在本地运行大模型变得非常简单。 <code python> from langchain_community.llms import Ollama # 连接到本地Ollama服务 ollama = Ollama( model="llama2", # 或 "mistral", "codellama" 等 base_url="http://localhost:11434" ) # 使用 result = ollama.predict("你好,请介绍一下自己") print(result) # 也可以这样创建 llm = Ollama(model="llama2:13b") </code> ==== 2.4.4 本地嵌入模型 ==== <code python> from langchain_community.embeddings import HuggingFaceEmbeddings # 使用开源嵌入模型 embeddings = HuggingFaceEmbeddings( model_name="sentence-transformers/all-MiniLM-L6-v2", model_kwargs={'device': 'cpu'}, # 或 'cuda' encode_kwargs={'normalize_embeddings': True} ) # 使用 text = "这是一个测试句子" vector = embeddings.embed_query(text) print(f"向量维度: {len(vector)}") </code> ---- ===== 2.5 其他模型提供商 ===== ==== 2.5.1 Azure OpenAI ==== <code python> from langchain_openai import AzureChatOpenAI, AzureOpenAIEmbeddings # Azure OpenAI配置 llm = AzureChatOpenAI( azure_endpoint="https://your-resource.openai.azure.com/", azure_deployment="your-deployment-name", openai_api_version="2024-02-01", openai_api_key="your-api-key" ) embeddings = AzureOpenAIEmbeddings( azure_endpoint="https://your-resource.openai.azure.com/", azure_deployment="your-embedding-deployment", openai_api_version="2024-02-01" ) </code> ==== 2.5.2 Google Vertex AI / Gemini ==== <code python> from langchain_google_vertexai import ChatVertexAI from langchain_google_genai import ChatGoogleGenerativeAI # Vertex AI gemini = ChatVertexAI( model_name="gemini-pro", project="your-project-id", location="us-central1" ) # 或使用Google Generative AI gemini = ChatGoogleGenerativeAI( model="gemini-pro", google_api_key="your-api-key" ) result = gemini.predict("你好") print(result) </code> ==== 2.5.3 国产大模型 ==== <code python> # 文心一言 from langchain_community.chat_models import QianfanChatEndpoint wenxin = QianfanChatEndpoint( qianfan_ak="your-access-key", qianfan_sk="your-secret-key" ) # 通义千问 from langchain_community.chat_models import Tongyi tongyi = Tongyi( dashscope_api_key="your-api-key" ) # 讯飞星火 from langchain_community.chat_models import SparkLLM spark = SparkLLM( spark_app_id="your-app-id", spark_api_key="your-api-key", spark_api_secret="your-secret" ) </code> ---- ===== 2.6 模型进阶技巧 ===== ==== 2.6.1 模型降级策略 ==== 当主要模型不可用时自动切换到备用模型: <code python> from langchain_openai import ChatOpenAI from langchain_anthropic import ChatAnthropic from langchain.schema import BaseMessage class FallbackLLM: """带降级策略的LLM包装器""" def __init__(self, primary, fallback): self.primary = primary self.fallback = fallback def predict(self, text: str) -> str: try: return self.primary.predict(text) except Exception as e: print(f"Primary model failed: {e}, falling back...") return self.fallback.predict(text) def predict_messages(self, messages: list[BaseMessage]) -> str: try: result = self.primary.predict_messages(messages) return result.content except Exception as e: print(f"Primary model failed: {e}, falling back...") result = self.fallback.predict_messages(messages) return result.content # 使用 primary = ChatOpenAI(model="gpt-4") fallback = ChatOpenAI(model="gpt-3.5-turbo") llm_with_fallback = FallbackLLM(primary, fallback) result = llm_with_fallback.predict("你好") print(result) </code> ==== 2.6.2 模型路由器 ==== 根据输入自动选择最合适的模型: <code python> from langchain_openai import ChatOpenAI class ModelRouter: """基于任务类型路由到不同模型""" def __init__(self): self.models = { "simple": ChatOpenAI(model="gpt-3.5-turbo"), "complex": ChatOpenAI(model="gpt-4"), "code": ChatOpenAI(model="gpt-4", temperature=0.2), } self.routing_prompt = """分析以下查询,分类为:simple(简单问答)、complex(复杂推理)、code(代码相关)。 只输出类别词。 查询: {query} 类别:""" self.classifier = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) def classify(self, query: str) -> str: result = self.classifier.predict(self.routing_prompt.format(query=query)) return result.strip().lower() def predict(self, query: str) -> str: category = self.classify(query) print(f"分类结果: {category}") model = self.models.get(category, self.models["simple"]) return model.predict(query) # 使用 router = ModelRouter() queries = [ "你好", # simple "解释量子纠缠的物理原理", # complex "写一个Python函数计算斐波那契数列" # code ] for q in queries: print(f"\n查询: {q}") result = router.predict(q) print(f"回复: {result[:100]}...") </code> ==== 2.6.3 批量请求优化 ==== <code python> from langchain_openai import ChatOpenAI from langchain.schema import HumanMessage import asyncio async def batch_predict(queries: list[str], batch_size: int = 5): """批量异步预测""" llm = ChatOpenAI() results = [] for i in range(0, len(queries), batch_size): batch = queries[i:i+batch_size] # 准备消息 messages_list = [[HumanMessage(content=q)] for q in batch] # 批量生成 batch_results = await llm.agenerate(messages_list) for gen in batch_results.generations: results.append(gen[0].text) return results # 使用示例 async def main(): queries = [f"问题{i}: 解释什么是Python" for i in range(10)] results = await batch_predict(queries) for q, r in zip(queries, results): print(f"{q} -> {r[:50]}...") asyncio.run(main()) </code> ==== 2.6.4 缓存机制 ==== 避免重复调用API,节省成本和延迟: <code python> from langchain.globals import set_llm_cache from langchain.cache import InMemoryCache, SQLiteCache from langchain_openai import ChatOpenAI # 内存缓存 set_llm_cache(InMemoryCache()) # 或持久化缓存 # set_llm_cache(SQLiteCache(database_path=".langchain.db")) llm = ChatOpenAI() # 第一次调用,会访问API result1 = llm.predict("你好") print("第一次调用完成") # 第二次调用同样的输入,直接从缓存返回 result2 = llm.predict("你好") print("第二次调用完成(使用缓存)") # 清除缓存 # from langchain.globals import get_llm_cache # get_llm_cache().clear() </code> ---- ===== 2.7 最佳实践 ===== ==== 2.7.1 API密钥管理 ==== <code python> # config.py - 集中管理配置 import os from dotenv import load_dotenv load_dotenv() class Config: OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY") # 模型默认配置 DEFAULT_MODEL = "gpt-3.5-turbo" DEFAULT_TEMPERATURE = 0.7 @classmethod def validate(cls): """验证必要配置""" if not cls.OPENAI_API_KEY: raise ValueError("OPENAI_API_KEY not set") # 使用 from config import Config from langchain_openai import ChatOpenAI Config.validate() llm = ChatOpenAI(api_key=Config.OPENAI_API_KEY) </code> ==== 2.7.2 错误处理 ==== <code python> from langchain_openai import ChatOpenAI from openai import RateLimitError, AuthenticationError import time def robust_predict(llm, prompt, max_retries=3): """健壮的预测函数,带重试机制""" for attempt in range(max_retries): try: return llm.predict(prompt) except RateLimitError: wait_time = 2 ** attempt # 指数退避 print(f"Rate limit hit, waiting {wait_time}s...") time.sleep(wait_time) except AuthenticationError as e: print(f"Authentication failed: {e}") raise except Exception as e: print(f"Attempt {attempt + 1} failed: {e}") if attempt == max_retries - 1: raise return None # 使用 llm = ChatOpenAI() result = robust_predict(llm, "你好") </code> ==== 2.7.3 成本监控 ==== <code python> from langchain.callbacks import get_openai_callback from langchain_openai import ChatOpenAI def track_cost(llm, prompts: list[str]): """追踪API调用成本""" with get_openai_callback() as cb: results = [] for prompt in prompts: result = llm.predict(prompt) results.append(result) print(f"总调用次数: {cb.successful_requests}") print(f"总token数: {cb.total_tokens}") print(f"提示token: {cb.prompt_tokens}") print(f"完成token: {cb.completion_tokens}") print(f"预估成本: ${cb.total_cost:.4f}") return results # 使用 llm = ChatOpenAI(model="gpt-4") prompts = ["问题1", "问题2", "问题3"] results = track_cost(llm, prompts) </code> ---- ===== 2.8 本章小结 ===== ==== 核心概念回顾 ==== - **三种模型类型** * LLMs: 文本补全模型 * Chat Models: 对话模型(最常用) * Embeddings: 文本向量模型 - **主要集成** * OpenAI: gpt-3.5-turbo, gpt-4系列 * Anthropic: Claude 3系列 * 本地: HuggingFace, Ollama, llama.cpp * 云服务: Azure, Vertex AI - **关键参数** * temperature: 控制创造性 * max_tokens: 控制输出长度 * frequency/presence_penalty: 减少重复 - **进阶技巧** * 模型降级策略 * 智能路由 * 批量请求 * 结果缓存 ==== 选择决策树 ==== <code> 选择什么模型? ├── 需要最高质量? │ ├── 是 → GPT-4 / Claude-3-Opus │ └── 否 → 继续 ├── 成本敏感? │ ├── 是 → GPT-3.5-turbo / Claude-3-Haiku │ └── 否 → 继续 ├── 数据隐私要求高? │ ├── 是 → 本地模型(Ollama/llama.cpp) │ └── 否 → GPT-4 / Claude-3-Sonnet └── 需要超长上下文? ├── 是 → Claude-3(200K) / GPT-4-turbo(128K) └── 否 → 其他模型 </code> ==== 作业 ==== - 实现一个模型对比工具,对同一个问题比较不同模型的回答 - 为你的应用设计一个智能模型路由系统 - 实现一个带成本预算限制的LLM调用器 - 测试本地模型的部署和调用 登录 Detach Close 该主题尚不存在 您访问的页面并不存在。如果允许,您可以使用创建该页面按钮来创建它。 langchain二次开发/核心组件详解_models.txt 最后更改: 2026/04/03 10:42由 张叶安 登录