在LangChain中，语言模型是最核心的组件。LangChain支持三大类模型：

定义：接收字符串输入，返回字符串输出的模型。

特点：

• 传统的文本补全接口
• 通常通过调用 `predict()` 或 `generate()` 方法
• 适合简单的文本生成任务

代表模型：

• OpenAI: `text-davinci-003`, `gpt-3.5-turbo-instruct`
• 开源: `LLaMA`, `Mistral`, `Falcon`

使用场景：

• 文本补全
• 简单的文本转换
• 需要直接控制输入格式的场景

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)

定义：专为对话场景优化的模型，接收消息列表，返回消息。

特点：

• 基于消息角色的架构（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`

使用场景： * 聊天机器人 * 多轮对话系统 * 需要角色设定的场景

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)

定义：将文本转换为高维向量（嵌入）的模型。

特点： * 输出是数值向量，不是文本 * 捕捉语义信息 * 用于相似度计算和检索

代表模型： * OpenAI: `text-embedding-3-small`, `text-embedding-3-large`, `text-embedding-ada-002` * 开源: `sentence-transformers`系列

使用场景： * 语义搜索 * 文本聚类 * RAG系统中的文档检索

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)}")

`ChatOpenAI` 是与OpenAI聊天模型交互的主要接口。

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,                        # 停止序列
)

1. temperature（温度）

控制输出的随机性： * `0.0`: 最确定性，总是选择概率最高的token * `0.7`: 平衡值，适合大多数场景 * `1.0+`: 更具创造性，输出更多样

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()

2. max_tokens（最大token数）

限制模型输出的长度： * 一个token约等于4个英文字符或0.75个单词 * 中文通常1-2个字符为一个token * 设置为None时，模型自行决定长度

# 不同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)

3. frequency_penalty 和 presence_penalty

用于减少重复： * `frequency_penalty`: 基于token出现频率的惩罚 * `presence_penalty`: 只要出现过的token都会受到惩罚 * 范围都是 -2.0 到 2.0

# 减少重复内容的示例
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))

ChatOpenAI 提供多种调用方式：

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)

OpenAI提供了多个模型，如何选择？

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}")

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}")

Claude是Anthropic开发的AI助手，以安全性和有用性著称。

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)

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()

Claude支持一些特殊功能：

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)

LangChain可以通过HuggingFace集成各种开源模型。

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)

对于消费级硬件，llama.cpp是很好的选择。

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)

Ollama让在本地运行大模型变得非常简单。

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")

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)}")

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"
)

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)

# 文心一言
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"
)

当主要模型不可用时自动切换到备用模型：

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)

根据输入自动选择最合适的模型：

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]}...")

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())

避免重复调用API，节省成本和延迟：

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()

# 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)

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, "你好")

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)

- 三种模型类型

• 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: 减少重复

- 进阶技巧

• 模型降级策略
• 智能路由
• 批量请求
• 结果缓存

选择什么模型？
├── 需要最高质量？
│   ├── 是 → 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)
    └── 否 → 其他模型

- 实现一个模型对比工具，对同一个问题比较不同模型的回答 - 为你的应用设计一个智能模型路由系统 - 实现一个带成本预算限制的LLM调用器 - 测试本地模型的部署和调用