语言模型 API

BaseLanguageModel 与 LLM 统一接口

概述

LangChain 的语言模型模块提供了与各种 LLM 提供商交互的统一接口。所有语言模型都继承自 BaseLanguageModel 基类,提供一致的 API。

graph TD
                    A[BaseLanguageModel] --> B[LLM]
                    A --> C[BaseChatModel]

                    B --> D[OpenAI]
                    B --> E[Anthropic]
                    B --> F[Hugging Face]

                    C --> G[ChatOpenAI]
                    C --> H[ChatAnthropic]
                    C --> I[ChatGoogleGenerativeAI]

                    style A fill:#e1f5fe
                    style C fill:#c8e6c9

核心类

BaseLanguageModel

所有语言模型的抽象基类,定义了统一接口规范。

class BaseLanguageModel(Serializable, Runnable[LanguageModelInput, LanguageModelOutput], Generic[LanguageModelInput, LanguageModelOutput]):
    """语言模型基类"""

    @property
    @abstractmethod
    def _llm_type(self) -> str:
        """返回模型类型标识"""

    @abstractmethod
    def _generate(
        self,
        prompts: List[str],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> LLMResult:
        """生成响应的核心方法"""

    def generate_prompt(
        self,
        prompts: List[PromptValue],
        stop: Optional[List[str]] = None,
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> LLMResult:
        """基于 PromptValue 生成响应"""

    async def agenerate_prompt(
        self,
        prompts: List[PromptValue],
        stop: Optional[List[str]] = None,
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> LLMResult:
        """异步生成响应"""

属性

属性 类型 说明
_llm_type str 模型类型标识(如 "openai", "anthropic")

相关链接

LLM

纯文本语言模型基类,处理字符串输入输出。

class LLM(BaseLanguageModel[str, str]):
    """纯文本语言模型"""

    def __call__(self, text: str) -> str:
        """调用模型的便捷方法"""

    def predict(self, text: str, **kwargs: Any) -> str:
        """预测文本(兼容方法)"""

    def predict_messages(
        self,
        messages: List[BaseMessage],
        **kwargs: Any
    ) -> BaseMessage:
        """预测消息响应"""

常用实现类

  • OpenAI - OpenAI GPT 模型
  • Anthropic - Anthropic Claude 模型
  • HuggingFacePipeline - Hugging Face 模型

核心方法

invoke

同步调用模型生成响应。

def invoke(
    self,
    input: LanguageModelInput,
    config: Optional[RunnableConfig] = None,
    stop: Optional[List[str]] = None,
    **kwargs: Any,
) -> LanguageModelOutput:
    """
    调用模型

    Args:
        input: 输入文本或消息列表
        config: 运行配置(包含 callbacks, metadata 等)
        stop: 停止词列表
        **kwargs: 额外参数

    Returns:
        模型输出(字符串或 AIMessage)
    """

参数说明

参数 类型 说明
input str | List[str] 输入提示文本
config RunnableConfig | None 运行配置,可包含 callbacks、tags、metadata
stop List[str] | None 停止生成词列表
**kwargs Any 模型特定参数(如 temperature, max_tokens)

返回值

类型 说明
str 生成的文本响应

stream

流式调用模型,逐 token 返回结果。

def stream(
    self,
    input: LanguageModelInput,
    config: Optional[RunnableConfig] = None,
    stop: Optional[List[str]] = None,
    **kwargs: Any,
) -> Iterator[LanguageModelOutput]:
    """
    流式调用模型

    Args:
        input: 输入文本
        config: 运行配置
        stop: 停止词
        **kwargs: 额外参数

    Yields:
        逐 token 返回的输出
    """

返回值

类型 说明
Iterator[str] 文本块的迭代器

batch

批量处理多个输入。

def batch(
    self,
    inputs: List[LanguageModelInput],
    config: Optional[Union[RunnableConfig, List[RunnableConfig]]] = None,
    stop: Optional[List[str]] = None,
    **kwargs: Any,
) -> List[LanguageModelOutput]:
    """
    批量调用模型

    Args:
        inputs: 输入列表
        config: 配置(单个或多个)
        stop: 停止词
        **kwargs: 额外参数

    Returns:
        输出列表
    """

ainvoke / astream / abatch

异步版本的调用方法。

async def ainvoke(
    self,
    input: LanguageModelInput,
    config: Optional[RunnableConfig] = None,
    stop: Optional[List[str]] = None,
    **kwargs: Any,
) -> LanguageModelOutput:
    """异步调用模型"""

async def astream(
    self,
    input: LanguageModelInput,
    config: Optional[RunnableConfig] = None,
    stop: Optional[List[str]] = None,
    **kwargs: Any,
) -> AsyncIterator[LanguageModelOutput]:
    """异步流式调用"""

async def abatch(
    self,
    inputs: List[LanguageModelInput],
    config: Optional[Union[RunnableConfig, List[RunnableConfig]]] = None,
    **kwargs: Any,
) -> List[LanguageModelOutput]:
    """异步批量调用"""

bind

绑定参数到模型,创建新的可运行对象。

def bind(
    self,
    **kwargs: Any,
) -> Runnable[LanguageModelInput, LanguageModelOutput]:
    """
    绑定参数

    Args:
        **kwargs: 要绑定的参数(如 temperature=0.7)

    Returns:
        绑定参数后的新 Runnable
    """

示例

python
from langchain_openai import OpenAI

llm = OpenAI(model="gpt-3.5-turbo-instruct")
cool_llm = llm.bind(temperature=0.3)
creative_llm = llm.bind(temperature=0.9)

with_types

配置输入输出类型。

def with_types(
    self,
    input_type: Optional[Type[LanguageModelInput]] = None,
    output_type: Optional[Type[LanguageModelOutput]] = None,
) -> Runnable[LanguageModelInput, LanguageModelOutput]:
    """
    配置类型

    Args:
        input_type: 输入类型
        output_type: 输出类型

    Returns:
        配置类型后的 Runnable
    """

get_num_tokens

计算文本的 token 数量。

def get_num_tokens(self, text: str) -> int:
    """
    计算 token 数量

    Args:
        text: 要计算的文本

    Returns:
        token 数量
    """

get_num_tokens_from_messages

计算消息列表的 token 数量。

def get_num_tokens_from_messages(
    self,
    messages: List[BaseMessage],
) -> int:
    """
    计算消息的 token 数量

    Args:
        messages: 消息列表

    Returns:
        token 总数
    """

使用示例

python
from langchain_openai import OpenAI

# 初始化模型
llm = OpenAI(
    model="gpt-3.5-turbo-instruct",
    temperature=0.7,
    max_tokens=1000,
    api_key="your-api-key"
)

# 基础调用
response = llm.invoke("讲一个关于 Python 的故事")
print(response)

# 流式调用
for chunk in llm.stream("用三句话介绍人工智能"):
    print(chunk, end="", flush=True)

# 批量调用
responses = llm.batch([
    "什么是 Python?",
    "什么是 JavaScript?",
    "什么是 Rust?"
])

# 绑定参数
cool_llm = llm.bind(temperature=0.1)
response = cool_llm.invoke("写一首关于编程的诗")

# 带配置的调用
response = llm.invoke(
    "写一首关于春天的诗",
    config={
        "tags": ["creative", "poetry"],
        "metadata": {"user_id": "user_123"}
    }
)

# 异步调用
import asyncio

async def async_example():
    response = await llm.ainvoke("你好")
    print(response)

    async for chunk in llm.astream("介绍 LangChain"):
        print(chunk, end="", flush=True)

asyncio.run(async_example())

相关 API