Skip to main content
开始使用 RunPod LLM。

​
概述

本指南介绍了如何使用 LangChain RunPod LLM 类与托管在 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

# 确保环境变量已设置（或直接传递给 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
    model_kwargs={
        "max_new_tokens": 256,
        "temperature": 0.6,
        "top_k": 50,
        # 添加端点处理器支持的其他参数
    },
    # 可选：调整轮询
    # 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."

# 调用（同步）
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 模拟）
print("\n--- Sync Stream Response ---")
try:
    for chunk in llm.stream(prompt):
        print(chunk, end="", flush=True)
    print()  # 换行
except Exception as e:
    print(
        f"\nError streaming LLM: {e}. Ensure endpoint handler supports streaming output format."
    )

​
异步使用

# 异步调用
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}.")

# 异步流式传输
print("\n--- Async Stream Response ---")
try:
    async for chunk in llm.astream(prompt):
        print(chunk, end="", flush=True)
    print()  # 换行
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

# 假设 'llm' 变量已在 '实例化' 单元格中实例化
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}")

# 异步链
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" 键中返回生成的文本（例如，{"output": "Generated text..."} 或 {"output": {"text": "..."}}）。
  • 流式传输： 对于通过 /stream 端点的模拟流式传输，处理器必须在状态响应中用块字典列表填充 "stream" 键，如 [{"output": "token1"}, {"output": "token2"}]。

​
API 参考

有关 RunPod LLM 类、参数和方法的详细文档，请参阅源代码或生成的 API 参考（如果可用）。 源代码链接：https://github.com/runpod/langchain-runpod/blob/main/langchain_runpod/llms.py
在 GitHub 上编辑此页面 或提交问题。
连接这些文档到 Claude、VSCode 等，通过 MCP 获取实时答案。