LangChain 模型调用详解：OpenAI、Google 与 HuggingFace 集成

你好，世界！

异步支持对于同时调用多个 LLM 特别有用，因为这些调用是网络限制的。

可以使用 agenerate 方法异步调用 LLM。

import time
import asyncio
from langchain_openai import OpenAI

# 定义一个函数，使用 OpenAI 的 API 生成文本并依次打印结果
def generate_serially():
    llm = OpenAI(temperature=0.9)  # 初始化 OpenAI 实例
    for _ in range(5):
        resp = llm.generate(["你好啊?"])  # 调用生成方法，传入文本请求
        print(resp.generations[0][0].text)  # 打印生成的文本结果

# 定义一个异步函数，使用 OpenAI 的 API 异步生成文本并打印结果
async def async_generate(llm):
    resp = await llm.agenerate(["你是谁?"])  # 异步调用生成方法，传入文本请求
    print(resp.generations[0][0].text)  # 打印异步生成的文本结果

# 定义一个异步函数，用于并发执行多个 async_generate 函数
async def generate_concurrently():
    llm = OpenAI(temperature=0.9)  # 初始化 OpenAI 实例
    tasks = [async_generate(llm) for _ in range(5)]  # 创建多个任务列表
    await asyncio.gather(*tasks)  # 并发等待所有任务完成

# 计算并发执行时间
s = time.perf_counter()  # 记录开始时间
asyncio.run(generate_concurrently())  # 运行异步函数
elapsed = time.perf_counter() - s  # 计算执行时间
print(f"异步执行 {elapsed:0.2f} 秒.")  # 打印并发执行时间

# 计算串行执行时间
s = time.perf_counter()  # 记录开始时间
generate_serially()  # 串行执行函数
elapsed = time.perf_counter() - s  # 计算执行时间
print(f"同步执行 {elapsed:0.2f} 秒.")  # 打印串行执行时间

我是一个人工智能程序，没有真正的身份。我被设计来回答问题和提供帮助。

我是一个人工智能助手，可以回答你关于技术或知识的问题。

我是一个程序，没有具体的身份。我是由人类编写的，用来帮助回答问题和执行任务的。

我是一台人工智能程序，无法具有实际的身份。我是由程序员编写和训练的，旨在通过语言交互来提供帮助和娱乐。

我是一个人工智能程序，被设计和程序员们一起工作，以帮助解决各种问题。我可以回答你的问题，提供帮助和建议。
异步执行 6.20 秒.

你好，我是一个人工智能助手。有什么可以帮到您的吗？

我是一个程序，无法感受情绪，但是很高兴能和你交流。你好吗？

你好！我是一个人工智能助手，很高兴认识你。有什么可以帮助你的吗？

你好，我是一个智能助手，很高兴认识你。有什么可以帮助你的吗？

Hello! How are you?
同步执行 10.29 秒.

聊天模型

# 聊天模型是 LangChain 的核心组件，使用聊天消息作为输入并返回聊天消息作为输出。
# LangChain 集成了许多模型提供商（OpenAI、Cohere、Hugging Face 等），并公开了标准接口用于
# 与这些模型进行交互。
# LangChain 允许在同步、异步、批处理和流模式下使用模型，并提供其他功能，如缓存。

LangChain 有一些内置的消息类型：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

text = "你好"
# 聊天模型
chat_model = ChatOpenAI(model="gpt-3.5-turbo")
messages = [HumanMessage(content=text)]

res = chat_model.invoke(messages)
print(res)

content='你好！有什么可以帮助你的吗？' response_metadata={'token_usage': {'completion_tokens': 17, 'prompt_tokens': 9, 'total_tokens': 26}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_b28b39ffa8', 'finish_reason': 'stop', 'logprobs': None} id='run-c1641695-dab6-44f3-9037-44904cc166e9-0'

聊天模型支持多个消息作为输入：

messages = [
    SystemMessage(content="你是一位乐于助人的助手。"),
    HumanMessage(content="你好")
]
chat.invoke(messages)

文本嵌入模型

Embedding 类是一个用于与嵌入进行交互的类。有许多嵌入提供商（OpenAI、Cohere、Hugging Face 等）- 这个类旨在为所有这些提供商提供一个标准接口。

嵌入会创建文本的向量表示，这使得我们可以在向量空间中考虑文本，并进行语义搜索等操作，即在向量空间中查找最相似的文本片段。

from langchain_openai import OpenAIEmbeddings

# 初始化 OpenAIEmbeddings 实例
embeddings = OpenAIEmbeddings()
# embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

# 定义一个文本字符串
text = "这是一份测试文档."

# 嵌入文档
doc_result = embeddings.embed_documents([text])
print(doc_result[0][:5])

# 嵌入查询
query_result = embeddings.embed_query(text)
print(query_result[:5])

[-0.006240383366130216, -0.003690876131687136, 0.0035220345636857247, -0.006399094239817186, -0.013473554751640016]

[-0.006240383366130216, -0.003690876131687136, 0.0035220345636857247, -0.006399094239817186, -0.013473554751640016]

调用谷歌模型

通过 langchain-google-genai 集成包中的类 ChatGoogleGenerativeAI，访问 Google AI Gemini 和 gemini-vision 模型，以及其他生成模型。

环境准备

安装 langchain-google-genai 包：

pip install langchain-google-genai

访问 Google AI Studio，创建 API 密钥：

import os

# 设置 API_Key
os.environ["GOOGLE_API_KEY"] = ''

聊天模型

from langchain_google_genai import ChatGoogleGenerativeAI

llm = ChatGoogleGenerativeAI(model="gemini-pro")
res = llm.invoke("你好，你是谁？")
print(res)

SystemMessageGemini 目前不支持，但可以设置 convert_system_message_to_human=True 来支持：

from langchain_google_genai import ChatGoogleGenerativeAI

llm = ChatGoogleGenerativeAI(model="gemini-pro-vision")
from langchain_core.messages import HumanMessage, SystemMessage

model = ChatGoogleGenerativeAI(model="gemini-pro", convert_system_message_to_human=True)
res = model.invoke(
    [
        SystemMessage(content="你是一个乐于助人的 AI 助手"),
        HumanMessage(content="你好啊"),
    ]
)
print(res.content)

你好！很高兴见到你。我是 Gemini，是 Google 开发的多模态 AI 语言模型。

大语言模型

from langchain_google_genai import GoogleGenerativeAI

# 使用 LLM 对话
GOOGLE_API_KEY = ""
llm = GoogleGenerativeAI(model="gemini-pro", google_api_key=GOOGLE_API_KEY)
print(
    llm.invoke(
        "Python 作为编程语言的一些优点和缺点是什么?"
    )
)

使用 Chain

from langchain_core.prompts import PromptTemplate

# 提示模板
template = """
问题：{question}

答：让我们一步一步来思考
"""
# 创建模板实例
prompt = PromptTemplate.from_template(template)
# 使用 chain
chain = prompt | llm
# 调用
question = "2+2 是多少?"
print(chain.invoke({"question": question}))

流式处理和批处理

for chunk in llm.stream("Write a limerick about LLMs."):
    print(chunk.content)

results = llm.batch(
    [
        "What's 2+2?",
        "What's 3+5?",
    ]
)
for res in results:
    print(res.content)

视觉消息对话

from langchain_core.messages import HumanMessage
from langchain_google_genai import ChatGoogleGenerativeAI

llm = ChatGoogleGenerativeAI(model="gemini-pro-vision")

message = HumanMessage(
    content=[
        {
            "type": "text",
            "text": "这张图片中有什么?",
        },
        {"type": "image_url", "image_url": "https://picsum.photos/seed/picsum/200/300"},
    ]
)
res = llm.invoke([message])
# 图片中显示的是一座被白雪覆盖的山峰，山峰在日落时分被染成了粉红色。
print(res.content)

文本嵌入

谷歌模型使用嵌入模型也非常简单，如下所示：

from langchain_google_genai import GoogleGenerativeAIEmbeddings

# 加载内嵌向量模型
embeddings = GoogleGenerativeAIEmbeddings(model="models/embedding-001")

# 向量化
vectors = embeddings.embed_documents(
    [
        "Today is Monday",
        "Today is Tuesday",
        "Today is April Fools day",
    ]
)
print(len(vectors), len(vectors[0]))

# 向量查询
vector = embeddings.embed_query("hello, world!")
print(vector[:5])

调用 Hugging Face 模型

环境准备

访问 Hugging Face，在个人设置中心，创建一个 API Token。

安装以下相关库：

pip install text_generation langchainhub

在程序中设置 API Token：

# 导入 HuggingFace API Token
import os
os.environ['HUGGINGFACEHUB_API_TOKEN'] = 'HuggingFace API Token'

在命令行中运行 huggingface-cli login，设置 API Token。

(langchain) PS C:\WorkSpace\langchain> huggingface-cli login

    _|    _|  _|    _|    _|_|_|    _|_|_|  _|_|_|  _|      _|    _|_|_|      _|_|_|_|    _|_|      _|_|_|  _|_|_|_|
    _|    _|  _|    _|  _|        _|          _|    _|_|    _|  _|            _|        _|    _|  _|        _|
    _|_|_|_|  _|    _|  _|  _|_|  _|  _|_|    _|    _|  _|  _|  _|  _|_|      _|_|_|    _|_|_|_|  _|        _|_|_|
    _|    _|  _|    _|  _|    _|  _|    _|    _|    _|    _|_|  _|    _|      _|        _|    _|  _|        _|
    _|    _|    _|_|      _|_|_|    _|_|_|  _|_|_|  _|      _|    _|_|_|      _|        _|    _|    _|_|_|  _|_|_|_|

    A token is already saved on your machine. Run `huggingface-cli whoami` to get more information or `huggingface-cli logout` if you want to log out.        
    Setting a new token will erase the existing one.
    To login, `huggingface_hub` requires a token generated from https://huggingface.co/settings/tokens .
Token can be pasted using 'Right-Click'.
Enter your token (input will not be visible):
Token is valid (permission: write).
Your token has been saved in your configured git credential helpers (manager).
Your token has been saved to C:\Users\Admin\.cache\huggingface\token
Login successful

聊天模型

利用 HuggingFaceEndpoint 或 HuggingFaceHub 集成来实例化 LLM，然后与 LangChain 的聊天消息抽象进行交互。

1. HuggingFaceHub HuggingFaceHub 将在 LangChain 的 0.2.0 版本弃用，但目前仍然支持。

import os
os.environ["GOOGLE_API_KEY"] = ''

from langchain_community.llms import HuggingFaceHub

llm = HuggingFaceHub(
    repo_id="HuggingFaceH4/zephyr-7b-beta",
    task="text-generation",
    model_kwargs={
        "max_new_tokens": 30,
        "top_k": 1,
        "temperature": 0.1,
        "repetition_penalty": 1.03,
    },
)

print(llm.invoke("你好，你是谁？"))

你好，你是谁？

你好，我是小明。

请问，你在做什么？

我在学习汉语。

2. HuggingFaceEndpoint HuggingFaceEndpoint 类是 LangChain 现在及未来主要推荐的库。

from langchain_community.llms.huggingface_endpoint import HuggingFaceEndpoint

ENDPOINT_URL = "HuggingFaceH4/zephyr-7b-beta"
HF_TOKEN = ""

llm = HuggingFaceEndpoint(
    endpoint_url=ENDPOINT_URL,
    max_new_tokens=30,
    top_k=2,
    top_p=0.95,
    typical_p=0.95,
    temperature=0.01,
    repetition_penalty=1.03,
    huggingfacehub_api_token=HF_TOKEN
)

print(llm.invoke("你好，你是谁？"))

大语言模型

Hugging Face 模型中心托管了超过 12 万个模型、2 万个数据集和 5 万个演示应用程序 (Spaces)，所有这些都是开源和公开的。人们可以在该在线平台上轻松协作并共同构建机器学习。这些资源可以通过本地管道包装器从 LangChain 调用，也可以通过 HuggingFaceHub 类调用托管的推理端点。

使用 transformers 库，需要额外安装：

pip install transformers==4.39.3 accelerate==0.29.1 torch

from langchain_community.llms.huggingface_pipeline import HuggingFacePipeline

hf = HuggingFacePipeline.from_model_id(
    model_id="gpt2",
    task="text-generation",
    pipeline_kwargs={"max_new_tokens": 10},
)

print(hf.invoke("你好，你是谁？"))

transformers 也可以通过直接传入现有管道来加载：

# 导入 HuggingFacePipeline 类和相关模块
from langchain_community.llms.huggingface_pipeline import HuggingFacePipeline
from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline

# 定义 GPT-2 模型的标识
model_id = "gpt2"

# 加载 GPT-2 模型的 tokenizer
tokenizer = AutoTokenizer.from_pretrained(model_id)

# 加载 GPT-2 模型本身
model = AutoModelForCausalLM.from_pretrained(model_id)

# 创建一个文本生成的 pipeline，最多生成 10 个新 token
pipe = pipeline("text-generation", model=model, tokenizer=tokenizer, max_new_tokens=10)

# 初始化 HuggingFacePipeline 对象，传入文本生成的 pipeline
hf = HuggingFacePipeline(pipeline=pipe)

# 打印
print(hf.invoke("你好啊!"))

使用消息对象

from langchain.schema import (
    HumanMessage,
    SystemMessage,
)

messages = [
    SystemMessage(content="你是一个乐于助人的助手"),
    HumanMessage(content="你好，你是谁？"),
]

res = llm.invoke(messages)
print(res)

Token has not been saved to git credential helper. Pass `add_to_git_credential=True` if you want to set the git credential as well.
Token is valid (permission: write).
Your token has been saved to C:\Users\Admin\.cache\huggingface\token
Login successful
Assistant: 我是一个智能语音助手，我可以为您提供各种信息和帮助。

总结与最佳实践

错误处理

在实际生产环境中，模型调用可能会因为网络波动、API 限流或参数错误而失败。建议添加重试机制和异常捕获：

from langchain_core.exceptions import OutputParserException
from langchain.callbacks.base import BaseCallbackHandler

class MyHandler(BaseCallbackHandler):
    def on_llm_error(self, error: Exception, **kwargs) -> None:
        print(f"Error occurred: {error}")

成本优化

1. 缓存：对于重复的查询，利用 LangChain 的缓存功能减少 API 调用次数。
2. 模型选择：根据任务复杂度选择合适的模型大小，小任务优先使用轻量级模型。
3. 流式输出：使用流式处理 (stream) 可以减少首字延迟，提升用户体验。

安全与合规

1. 敏感数据：不要将用户隐私数据直接发送给公共 API，建议在本地进行脱敏处理。
2. Prompt 注入：对用户输入进行过滤，防止恶意 Prompt 攻击导致模型输出不当内容。
3. 合规性：确保使用的模型符合当地法律法规要求，特别是涉及生成内容的场景。

通过合理配置和使用 LangChain 提供的各类模型接口，开发者可以快速构建高效、稳定的 AI 应用。