Skip to main content
Open In Colab在 GitHub 上打开

RunPod LLM

开始使用 RunPod LLM。

概述

本指南介绍如何使用 LangChainRunPodLLM 类与 RunPod Serverless 上托管的文本生成模型进行交互。

设置

  1. 安装软件包： 
    pip install -qU langchain-runpod
  2. 部署 LLM 终端节点：按照 RunPod 提供商指南中的设置步骤在 RunPod Serverless 上部署兼容的文本生成终端节点并获取其终端节点 ID。
  3. 设置环境变量：确保RUNPOD_API_KEY和RUNPOD_ENDPOINT_ID已设置。
import getpass
import os

# Make sure environment variables are set (or pass them directly to RunPod)
if "RUNPOD_API_KEY" not in os.environ:
    os.environ["RUNPOD_API_KEY"] = getpass.getpass("Enter your RunPod API Key: ")
if "RUNPOD_ENDPOINT_ID" not in os.environ:
    os.environ["RUNPOD_ENDPOINT_ID"] = input("Enter your RunPod Endpoint ID: ")

实例

初始化RunPod类。您可以通过以下方式传递特定于模型的参数model_kwargs并配置轮询行为。

from langchain_runpod import RunPod

llm = RunPod(
    # runpod_endpoint_id can be passed here if not set in env
    model_kwargs={
        "max_new_tokens": 256,
        "temperature": 0.6,
        "top_k": 50,
        # Add other parameters supported by your endpoint handler
    },
    # Optional: Adjust polling
    # poll_interval=0.3,
    # max_polling_attempts=100
)

调用

使用标准的 LangChain.invoke()和.ainvoke()方法来调用模型。还支持通过以下方式进行流式处理.stream()和.astream()（通过轮询 RunPod 进行模拟/stream端点）。

prompt = "Write a tagline for an ice cream shop on the moon."

# Invoke (Sync)
try:
    response = llm.invoke(prompt)
    print("--- Sync Invoke Response ---")
    print(response)
except Exception as e:
    print(
        f"Error invoking LLM: {e}. Ensure endpoint ID/API key are correct and endpoint is active/compatible."
    )
# Stream (Sync, simulated via polling /stream)
print("\n--- Sync Stream Response ---")
try:
    for chunk in llm.stream(prompt):
        print(chunk, end="", flush=True)
    print()  # Newline
except Exception as e:
    print(
        f"\nError streaming LLM: {e}. Ensure endpoint handler supports streaming output format."
    )

异步使用

# AInvoke (Async)
try:
    async_response = await llm.ainvoke(prompt)
    print("--- Async Invoke Response ---")
    print(async_response)
except Exception as e:
    print(f"Error invoking LLM asynchronously: {e}.")
# AStream (Async)
print("\n--- Async Stream Response ---")
try:
    async for chunk in llm.astream(prompt):
        print(chunk, end="", flush=True)
    print()  # Newline
except Exception as e:
    print(
        f"\nError streaming LLM asynchronously: {e}. Ensure endpoint handler supports streaming output format."
    )

链接

LLM 与 LangChain 表达式语言 （LCEL） 链无缝集成。

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate

# Assumes 'llm' variable is instantiated from the 'Instantiation' cell
prompt_template = PromptTemplate.from_template("Tell me a joke about {topic}")
parser = StrOutputParser()

chain = prompt_template | llm | parser

try:
    chain_response = chain.invoke({"topic": "bears"})
    print("--- Chain Response ---")
    print(chain_response)
except Exception as e:
    print(f"Error running chain: {e}")

# Async chain
try:
    async_chain_response = await chain.ainvoke({"topic": "robots"})
    print("--- Async Chain Response ---")
    print(async_chain_response)
except Exception as e:
    print(f"Error running async chain: {e}")

终端节点注意事项

  • 输入：端点处理程序应期望{"input": {"prompt": "...", ...}}.
  • 输出：处理程序应返回"output"key 的 key（例如{"output": "Generated text..."}或{"output": {"text": "..."}}).
  • 流：对于通过/stream端点，则处理程序必须填充"stream"键中，其中包含 chunk 字典列表，例如[{"output": "token1"}, {"output": "token2"}].

API 参考

有关RunPodLLM 类、参数和方法，请参阅源代码或生成的 API 参考（如果可用）。

源代码链接： https://github.com/runpod/langchain-runpod/blob/main/langchain_runpod/llms.py