langchain_community.chat_models.gigachat.GigaChat

备注

GigaChat 实现了标准 Runnable 接口。🏃

Runnable 接口 在可运行对象上提供了额外的函数,例如 with_typeswith_retryassignbind 和更多。

class langchain_community.chat_models.gigachat.GigaChat[source]

继承自: _BaseGigaChat, BaseChatModel

GigaChat 大语言模型API。

使用时,您应传入用于访问GigaChat API的登录名和密码或使用令牌。

示例

from langchain_community.chat_models import GigaChat
giga = GigaChat(credentials=..., scope=..., verify_ssl_certs=False)
param access_token: Optional[str] = None

GigaChat的访问令牌

param auth_url: Optional[str] = None

认证URL

参数 base_url: 可选[str] = None

基本API URL

参数 ca_bundle_file: 可选[str] = None
参数 cache: 联合[BaseCache, bool, None] = None

是否缓存响应。

  • 如果是真,将使用全局缓存。

  • 如果是假,将不使用缓存。

  • 如果是None,如果已设置全局缓存,则使用全局缓存,否则不使用缓存。

  • 如果是BaseCache的实例,将使用提供的缓存。

当前不支持模型流的缓存。

参数 callback_manager: 可选[BaseCallbackManager] = None

[已弃用] 要添加到运行跟踪中的回调管理器。

参数 callbacks: Callbacks = None

要添加到运行跟踪中的回调。

参数 cert_file: 可选[str] = None
参数 credentials: Optional[str] = None

认证令牌

参数 custom_get_token_ids: Optional[Callable[[str], List[int]]] = None

可选的编码器,用于计算令牌。

参数 key_file: Optional[str] = None
参数 key_file_password: Optional[str] = None
参数 max_tokens: Optional[int] = None

生成令牌的最大数量

参数 metadata: Optional[Dict[str, Any]] = None

添加到运行痕迹的元数据

参数 model: Optional[str] = None

要使用的模型名称

参数 password: 可选[str] = None

认证密码

参数 profanity: bool = True

已弃用:检查粗鲁语言

参数 profanity_check: 可选[bool] = None

检查粗鲁语言

参数 rate_limiter: 可选[BaseRateLimiter] = None

可选的速率限制器,用于限制请求数量。

参数 repetition_penalty: 可选[float] = None

对重复标记施加的惩罚

参数 scope: 可选[str] = None

访问令牌的权限作用域

参数 streaming: bool = False

是否流式传输结果。

参数 tags: 可选[List[str]] = None

要添加到运行跟踪的标签。

参数 temperature: Optional[float] = None

使用的采样温度。

参数 timeout: Optional[float] = None

请求的超时时间。

参数 top_p: Optional[float] = None

用于核采样(nucleus sampling)的top_p值。值必须在0.0到1.0之间。

参数 update_interval: Optional[float] = None

在发送token之间的最小时间间隔(以秒为单位)。

参数 use_api_for_tokens: bool = False

使用GigaChat API来计算token数量。

参数 user: Optional[str] = None

用于认证的用户名。

参数 verbose: bool = False

详细日志。

参数 verify_ssl_certs: Optional[bool] = None

对所有请求进行证书检查。

__call__(messages: List[BaseMessage], stop: Optional[List[str]] = None, callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None, **kwargs: Any) BaseMessage

自版本 langchain-core==0.1.7 起已弃用: 请使用 invoke 代替。

参数
返回类型

BaseMessage

async abatch(inputs: List[Input], config: Optional[Union[RunnableConfig, List[RunnableConfig]]]] = None, *, return_exceptions: bool = False, **kwargs: Optional[Any]) List[Output]

默认实现使用asyncio.gather并行运行run_invoكe。

批处理的默认实现对I/O密集型可运行项效果良好。

子类应该覆盖此方法以提高批处理效率;例如,如果底层的可运行项使用支持批处理模式的API。

参数
  • inputs (List[Input]) – 可运行项的输入列表。

  • config (Optional[Union[RunnableConfig, List[RunnableConfig]]]) – 在调用可运行项时使用的配置。配置支持用于跟踪的标准键,如'tags'、'metadata','max_concurrency'用于控制并行执行的工作量,以及其他键。请参阅RunnableConfig以获取更多详情。默认为None。

  • return_exceptions (bool) – 是否返回异常而不是引发异常。默认为False。

  • kwargs (Optional[Any]) – 传递给可运行项的额外关键字参数。

返回

可运行项的输出列表。

返回类型

List[Output]

async abatch_as_completed(inputs: Sequence[Input], config: Optional[Union[RunnableConfig, Sequence[RunnableConfig]]] = None, **kwargs: Optional[Any]) AsyncIterator[Tuple[int, Union[Output, Exception]]]

在输入列表上并行运行ainvoke,并在完成时产生结果。

参数
  • inputs (Sequence[Input]) – Runnable的输入列表。

  • config (Optional[Union[RunnableConfig, Sequence[RunnableConfig]]]) – 调用Runnable时使用的配置。配置支持标准键,如‘tags’、’metadata’用于跟踪目的,‘max_concurrency’用于控制并行执行的工作量,以及其他键。请参阅RunnableConfig以获取更多详细信息。默认为None。

  • return_exceptions (bool) – 是否返回异常而不是引发异常。默认为False。

  • kwargs (Optional[Any]) – 传递给可运行项的额外关键字参数。

产出

输入索引和Runnable输出组成的元组。

返回类型

A tuple of the index of the input and the output from the Runnable.

async agenerate(messages: List[List[BaseMessage]], stop: Optional[List[str]] = None, callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None, *, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, run_name: Optional[str] = None, run_id: Optional[UUID] = None, **kwargs: Any) LLMResult

异步将一系列提示传递给模型并返回生成内容。

该方法应利用批处理调用对提供批处理API的模型进行操作。

当您想
  1. 使用批处理调用优势时,

  2. 需要模型的输出不仅仅是最顶部的生成的值时,

  3. 正在构建对底层语言模型类型(例如,纯文本完成模型与聊天模型)无感知的链时。

    类型。

参数
  • messages列表[ 列表[

  • stop可选[ 列表[

  • callbacks可选[ 联合[

  • **kwargs任何)-任意附加关键字参数。这些通常被传递到模型提供商API调用。

  • tags可选[ 列表[

  • metadata可选[ 字典[

  • run_name可选[ str]

  • run_id可选[ UUID]

  • **kwargs

返回

一个LLMResult,其中包含每个输入提示的候选生成列和一个额外的模型提供商特定的输出。

prompt和额外的模型提供商特定输出。

返回类型

LLMResult

async agenerate_prompt(prompts: List[PromptValue], stop:

异步地传递一系列提示并返回模型生成。

该方法应利用批处理调用对提供批处理API的模型进行操作。

当您想
  1. 使用批处理调用优势时,

  2. 需要模型的输出不仅仅是最顶部的生成的值时,

  3. 正在构建对底层语言模型类型(例如,纯文本完成模型与聊天模型)无感知的链时。

    类型。

参数
  • 提示 (列表[PromptValue]) – PromptValue 的列表。PromptValue 是一个对象,可以转换为与任何语言模型匹配的格式(纯文本生成模型的字符串和聊天模型的 BaseMessages)。

  • stop可选[ 列表[

  • callbacks可选[ 联合[

  • **kwargs任何)-任意附加关键字参数。这些通常被传递到模型提供商API调用。

返回

一个LLMResult,其中包含每个输入提示的候选生成列和一个额外的模型提供商特定的输出。

prompt和额外的模型提供商特定输出。

返回类型

LLMResult

async aget_model(model: str) gm.Model

获取关于模型的信息(异步)

参数

model (str) –

返回类型

gm.Model

async aget_models() gm.Models

获取 Gigachat 可用的模型(异步)

返回类型

gm.Models

async ainvoke(input: LanguageModelInput, config: Optional[RunnableConfig] = None, *, stop: Optional[List[str]] = None, **kwargs: Any) BaseMessage

ainvoke 的默认实现,从线程中调用 invoke。

默认实现允许在使用没有实现内部异步 invoke 版本的 Runnable 的情况下使用异步代码。

如果子类可以异步运行,则应重写此方法。

参数
  • input (LanguageModelInput) –

  • config (可选的RunnableConfig) –

  • stop (可选列表[str]]) –

  • kwargs (任意类型) –

返回类型

BaseMessage

async apredict(text: str, *, stop: Optional[Sequence[str]] = None, **kwargs: Any) str

自 langchain-core==0.1.7 版本弃用: 请使用 ainvoke 代替。

参数
  • text (str) –

  • stop (可选[字符串序列]) –

  • kwargs (任意类型) –

返回类型

str

async apredict_messages(messages: List[BaseMessage], *, stop: Optional[Sequence[str]] = None, **kwargs: Any) BaseMessage

自 langchain-core==0.1.7 版本弃用: 请使用 ainvoke 代替。

参数
  • messages (列表[BaseMessage]) –

  • stop (可选[字符串序列]) –

  • kwargs (任意类型) –

返回类型

BaseMessage

as_tool(args_schema: Optional[Type[BaseModel]] = None, *, name: Optional[str] = None, description: Optional[str] = None, arg_types: Optional[Dict[str, Type]] = None) BaseTool

测试版

此API为测试版,未来可能发生变化。

从可运行对象创建BaseTool。

as_tool将从可运行对象实例化一个具有名称、描述和args_schema的BaseTool。尽可能情况下,将根据runnable.get_input_schema推断模式。另外(例如,如果可运行对象接受字典作为输入,而且具体的字典键没有类型),可以使用args_schema直接指定模式。还可以通过传递arg_types来只指定所需的参数及其类型。

参数
  • args_schema (可选[Type[BaseModel]]) – 工具的模式。默认为None。

  • name (可选[str]) – 工具的名称。默认为None。

  • description (可选[str]) – 工具的描述。默认为None。

  • arg_types (可选[Dict[str, Type]]) – 参数名称到类型的字典。默认为None。

返回

BaseTool实例。

返回类型

BaseTool

类型化字典输入

from typing import List
from typing_extensions import TypedDict
from langchain_core.runnables import RunnableLambda

class Args(TypedDict):
    a: int
    b: List[int]

def f(x: Args) -> str:
    return str(x["a"] * max(x["b"]))

runnable = RunnableLambda(f)
as_tool = runnable.as_tool()
as_tool.invoke({"a": 3, "b": [1, 2]})

dict输入,通过args_schema指定模式

from typing import Any, Dict, List
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.runnables import RunnableLambda

def f(x: Dict[str, Any]) -> str:
    return str(x["a"] * max(x["b"]))

class FSchema(BaseModel):
    """Apply a function to an integer and list of integers."""

    a: int = Field(..., description="Integer")
    b: List[int] = Field(..., description="List of ints")

runnable = RunnableLambda(f)
as_tool = runnable.as_tool(FSchema)
as_tool.invoke({"a": 3, "b": [1, 2]})

dict输入,通过arg_types指定模式

from typing import Any, Dict, List
from langchain_core.runnables import RunnableLambda

def f(x: Dict[str, Any]) -> str:
    return str(x["a"] * max(x["b"]))

runnable = RunnableLambda(f)
as_tool = runnable.as_tool(arg_types={"a": int, "b": List[int]})
as_tool.invoke({"a": 3, "b": [1, 2]})

字符串输入

from langchain_core.runnables import RunnableLambda

def f(x: str) -> str:
    return x + "a"

def g(x: str) -> str:
    return x + "z"

runnable = RunnableLambda(f) | g
as_tool = runnable.as_tool()
as_tool.invoke("b")

新在版本0.2.14。

async astream(input: LanguageModelInput, config: Optional[RunnableConfig] = None, **kwargs: Anyscroll: Optional[List[str]] = None) AsyncIterator[BaseMessageChunk]

astream 的默认实现,它调用了ainvoke。如果子类支持流式输出,则应该重写此方法。

参数
  • input (LanguageModelInput) – Runnable 的输入。

  • config (Optional[RunnableConfig]) – 用于 Runnable 的配置。默认为 None。

  • kwargs (Any) – 将传递给 Runnable 的额外关键字参数。

  • stop (可选列表[str]]) –

产出

Runnable 的输出。

返回类型

AsyncIterator[BaseMessageChunk]

astream_events(input: Any, config: Optional[RunnableConfig] = None, *, version: Literal['v1', 'v2'], include_names: Optional[Sequence[str]] = None, include_types: Optional[Sequence[str]] = None, include_tags: Optional[Sequence[str]] = None, exclude_names: Optional[Sequence[str]] = None, exclude_types: Optional[Sequence[str]] = None, exclude_tags: Optional[Sequence[str]] = None, **kwargs: Any) AsyncIterator[Union[StandardStreamEvent, CustomStreamEvent]]

测试版

此API为测试版,未来可能发生变化。

生成事件流。

用于对提供实时信息的 StreamEvents 进行迭代,包括中间结果的 StreamEvents。

StreamEvent 是一个具有以下模式的字典

  • event: str - 事件名称格式为

    :on_[runnable_type]_(start|stream|end)。

  • name: str - 生成事件的 Runnable 的名称。

  • run_id: str - 与生成的 Runnable 的给定执行相关的随机生成的 ID。发出事件的子 Runnable 作为父 Runnable 执行的一部分被分配其自己的唯一 ID。

    parent_ids: List[str] - 生成事件的父 Runnable 的 ID。根 Runnable 将有一个空列表。父 ID 的顺序是从根到直接父。仅适用于 API 的 v2 版本。API 的 v1 版本将返回一个空列表。

  • tags: Optional[List[str]] - 生成事件的 Runnable 的标签。

    connection: Optional[List[str]] - Fluent's connectivity settings for the given service.

  • memory: Optional[Dict[str, Any]] - Memory used by the Runnable.

    output: Optional[Union[str, List[str]]] - Output of the Runnable.

  • metadata可选[Dict[str, Any]] - 由事件生成器的可运行物体的元数据。

  • dataDict[str, Any]

以下是一个表格,展示了各个链可能产生的某些事件。为简洁起见,省略了元数据字段。表后包含链定义。

注意 此参考表适用于架构的 V2 版本。

事件

名称

分块

输入

输出

on_chat_model_start

【模型名称】

{“messages”:[[SystemMessage,HumanMessage]]}

on_chat_model_stream

【模型名称】

AIMessageChunk(content="hello")

on_chat_model_end

【模型名称】

{“messages”:[[SystemMessage,HumanMessage]]}

AIMessageChunk(content="hello world")

on_llm_start

【模型名称】

{‘input’:‘hello’}

on_llm_stream

【模型名称】

“Hello”

on_llm_end

【模型名称】

“Hello human!”

on_chain_start

format_docs

on_chain_stream

format_docs

“hello world!,再见世界!”

on_chain_end

format_docs

[Document(…)]

“hello world!,再见世界!”

on_tool_start

某工具

{“x”:1,“y”:2”}

on_tool_end

某工具

{“x”:1,“y”:2”}

on_retriever_start

[检索器名称]

{“query”:“hello”}

on_retriever_end

[检索器名称]

{“query”:“hello”}

[Document(…), ..]

on_prompt_start

[模板名称]

{“question”:“hello”}

on_prompt_end

[模板名称]

{“question”:“hello”}

ChatPromptValue(messages: [SystemMessage, …])

除了标准事件之外,用户还可以发送自定义事件(见以下示例)。

自定义事件仅在 API 的 v2 版本中呈现!

自定义事件的格式如下

属性

类型

描述

名称

str

事件的自定义名称。

数据

Any

与事件关联的数据。这可以是一切,我们建议将其制作成可序列化为 JSON 的形式。

以下是与上述标准事件相关的声明

format_docs:

def format_docs(docs: List[Document]) -> str:
    '''Format the docs.'''
    return ", ".join([doc.page_content for doc in docs])

format_docs = RunnableLambda(format_docs)

某工具:

@tool
def some_tool(x: int, y: str) -> dict:
    '''Some_tool.'''
    return {"x": x, "y": y}

提示:

template = ChatPromptTemplate.from_messages(
    [("system", "You are Cat Agent 007"), ("human", "{question}")]
).with_config({"run_name": "my_template", "tags": ["my_template"]})

示例

from langchain_core.runnables import RunnableLambda

async def reverse(s: str) -> str:
    return s[::-1]

chain = RunnableLambda(func=reverse)

events = [
    event async for event in chain.astream_events("hello", version="v2")
]

# will produce the following events (run_id, and parent_ids
# has been omitted for brevity):
[
    {
        "data": {"input": "hello"},
        "event": "on_chain_start",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
    {
        "data": {"chunk": "olleh"},
        "event": "on_chain_stream",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
    {
        "data": {"output": "olleh"},
        "event": "on_chain_end",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
]

示例:发送自定义事件

from langchain_core.callbacks.manager import (
    adispatch_custom_event,
)
from langchain_core.runnables import RunnableLambda, RunnableConfig
import asyncio


async def slow_thing(some_input: str, config: RunnableConfig) -> str:
    """Do something that takes a long time."""
    await asyncio.sleep(1) # Placeholder for some slow operation
    await adispatch_custom_event(
        "progress_event",
        {"message": "Finished step 1 of 3"},
        config=config # Must be included for python < 3.10
    )
    await asyncio.sleep(1) # Placeholder for some slow operation
    await adispatch_custom_event(
        "progress_event",
        {"message": "Finished step 2 of 3"},
        config=config # Must be included for python < 3.10
    )
    await asyncio.sleep(1) # Placeholder for some slow operation
    return "Done"

slow_thing = RunnableLambda(slow_thing)

async for event in slow_thing.astream_events("some_input", version="v2"):
    print(event)
参数
  • inputAny)– 可运行的输入。

  • config可选[langchain_core.runnables.config.RunnableConfig])– 为可运行使用的配置。

  • version字面量[‘v1’,‘v2’])– 要使用的架构版本,可使用 v2v1。用户应使用 v2v1 用于反向兼容性,将在 0.4.0 中弃用。默认值将在 API 稳定之前不赋值。自定义事件仅在 v2 中呈现。

  • include_names可选[str])– 仅包含具有匹配名称的可运行事件。

  • include_types可选[str])– 仅包含具有匹配类型的可运行事件。

  • include_tags可选[str])– 仅包含具有匹配标签的可运行事件。

  • exclude_names可选[str])– 排除具有匹配名称的可运行事件。

  • exclude_types可选[str])– 排除具有匹配类型的可运行事件。

  • exclude_tags可选[str])– 排除具有匹配标签的可运行事件。

  • kwargsAny)– 传递给可运行的附加关键字参数。这些将被传递给 astream_log,因为 astream_events 的此实现是建立在 astream_log 之上的。

产出

StreamEvents 的异步流。

异常

NotImplementedError – 如果版本不是 v1v2.

返回类型

AsyncIterator[Union[StandardStreamEvent, CustomStreamEvent]]

async atokens_count(input_: List[str], model: Optional[str] = None) List[gm.TokensCount]

获取字符串列表的令牌(异步)

参数
  • input_ (List[str]) –

  • model (Optional[str]) –

返回类型

List[gm.TokensCount]

batch(inputs: List[Input], config: Optional[Union[RunnableConfig, List[RunnableConfig]]], *, return_exceptions: bool = False, **kwargs: Optional[Any]) List[Output]

默认实现使用线程池执行器并行运行invoke。

批处理的默认实现对I/O密集型可运行项效果良好。

子类应该覆盖此方法以提高批处理效率;例如,如果底层的可运行项使用支持批处理模式的API。

参数
  • inputs (List[Input]) –

  • config (Optional[Union[RunnableConfig, List[RunnableConfig]]]) –

  • return_exceptions (bool) –

  • kwargs (Optional[Any]) –

返回类型

List[Output]

batch_as_completed(inputs: Sequence[Input], config: Optional[Union[RunnableConfig, Sequence[RunnableConfig]]]] = None, *, return_exceptions: bool = False, **kwargs: Optional[Any]) Iterator[Tuple[int, Union[Output, Exception]]]

并行运行调用列表,并根据完成顺序输出结果。

参数
  • inputs (Sequence[Input]) –

  • config (Optional[Union[RunnableConfig, Sequence[RunnableConfig]]]) –

  • return_exceptions (bool) –

  • kwargs (Optional[Any]) –

返回类型

Iterator[Tuple[int, Union[Output, Exception]]]

bind_tools(tools: Sequence[Union[Dict[str, Any], Type, Callable, BaseTool]], **kwargs: Any) Runnable[LanguageModelInput, BaseMessage]
参数
  • tools (Sequence[Union[Dict[str, Any], Type, Callable, BaseTool]]) –

  • kwargs (任意类型) –

返回类型

Runnable[LanguageModelInput, BaseMessage]

call_as_llm(message: str, stop: Optional[List[str]] = None, **kwargs: Any) str

自版本 langchain-core==0.1.7 起已弃用: 请使用 invoke 代替。

参数
  • message (str) –

  • stop (可选列表[str]]) –

  • kwargs (任意类型) –

返回类型

str

configurable_alternatives(which: ConfigurableField, *, default_key: str = 'default', prefix_keys: bool = False, **kwargs: Union[Runnable[Input, Output], Callable[[], Runnable[Input, Output]]]) RunnableSerializable[Input, Output]

配置运行动作的运行时可设置的备选方案。

参数
  • which (ConfigurableField) – 用于选择备选方案的ConfigurableField实例。

  • default_key (str) – 如果未选择备选方案时使用默认键。默认为“default”。

  • prefix_keys (bool) – 是否在键前缀加上ConfigurableField ID。默认为False。

  • **kwargs (Union[Runnable[Input, Output], Callable[[], Runnable[Input, Output]]]) – 键到运行动作实例或返回运行动作实例的可调用对象字典。

返回

配置备选方案的新运行动作。

返回类型

RunnableSerializable[Input, Output]

from langchain_anthropic import ChatAnthropic
from langchain_core.runnables.utils import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatAnthropic(
    model_name="claude-3-sonnet-20240229"
).configurable_alternatives(
    ConfigurableField(id="llm"),
    default_key="anthropic",
    openai=ChatOpenAI()
)

# uses the default model ChatAnthropic
print(model.invoke("which organization created you?").content)

# uses ChatOpenAI
print(
    model.with_config(
        configurable={"llm": "openai"}
    ).invoke("which organization created you?").content
)
configurable_fields(**kwargs: Union[ConfigurableField, ConfigurableFieldSingleOption, ConfigurableFieldMultiOption]) RunnableSerializable[Input, Output]

在运行时配置特定的Runnable字段。

参数

**kwargs (Union[ConfigurableField, ConfigurableFieldSingleOption, ConfigurableFieldMultiOption]) – 要配置的ConfigurableField实例的字典。

返回

带有配置字段的新的Runnable。

返回类型

RunnableSerializable[Input, Output]

from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatOpenAI(max_tokens=20).configurable_fields(
    max_tokens=ConfigurableField(
        id="output_token_number",
        name="Max tokens in the output",
        description="The maximum number of tokens in the output",
    )
)

# max_tokens = 20
print(
    "max_tokens_20: ",
    model.invoke("tell me something about chess").content
)

# max_tokens = 200
print("max_tokens_200: ", model.with_config(
    configurable={"output_token_number": 200}
    ).invoke("tell me something about chess").content
)
generate(messages: List[List[BaseMessage]], stop: Optional[List[str]] = None, callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None, *, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, run_name: Optional[str] = None, run_id: Optional[UUID] = None, **kwargs: Any) LLMResult

将一系列提示传递给模型并返回模型生成的结果。

该方法应利用批处理调用对提供批处理API的模型进行操作。

当您想
  1. 使用批处理调用优势时,

  2. 需要模型的输出不仅仅是最顶部的生成的值时,

  3. 正在构建对底层语言模型类型(例如,纯文本完成模型与聊天模型)无感知的链时。

    类型。

参数
  • messages列表[ 列表[

  • stop可选[ 列表[

  • callbacks可选[ 联合[

  • **kwargs任何)-任意附加关键字参数。这些通常被传递到模型提供商API调用。

  • tags可选[ 列表[

  • metadata可选[ 字典[

  • run_name可选[ str]

  • run_id可选[ UUID]

  • **kwargs

返回

一个LLMResult,其中包含每个输入提示的候选生成列和一个额外的模型提供商特定的输出。

prompt和额外的模型提供商特定输出。

返回类型

LLMResult

generate_prompt(prompts: List[PromptValue], stop: Optional[List[str]] = None, callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None, **kwargs: Any) LLMResult

将一系列提示传递给模型并返回模型生成的结果。

该方法应利用批处理调用对提供批处理API的模型进行操作。

当您想
  1. 使用批处理调用优势时,

  2. 需要模型的输出不仅仅是最顶部的生成的值时,

  3. 正在构建对底层语言模型类型(例如,纯文本完成模型与聊天模型)无感知的链时。

    类型。

参数
  • 提示 (列表[PromptValue]) – PromptValue 的列表。PromptValue 是一个对象,可以转换为与任何语言模型匹配的格式(纯文本生成模型的字符串和聊天模型的 BaseMessages)。

  • stop可选[ 列表[

  • callbacks可选[ 联合[

  • **kwargs任何)-任意附加关键字参数。这些通常被传递到模型提供商API调用。

返回

一个LLMResult,其中包含每个输入提示的候选生成列和一个额外的模型提供商特定的输出。

prompt和额外的模型提供商特定输出。

返回类型

LLMResult

get_model(model: str) gm.Model

获取模型信息

参数

model (str) –

返回类型

gm.Model

get_models() gm.Models

获取Gigachat可用的模型

返回类型

gm.Models

get_num_tokens(text: str) int

计算大约的词元数

参数

text (str) –

返回类型

整数

get_num_tokens_from_messages(messages: List[BaseMessage]) int

获取消息中的词元数。

用于检查输入是否适合模型的上下文窗口。

参数

messages (列表[BaseMessage]) – 要分词的消息输入。

返回

消息中标记数总和。

返回类型

整数

get_token_ids(text: str) List[int]

返回文本中标记的有序ID。

参数

text (str) – 要分词的字符串输入。

返回

与文本中的标记对应的ID列表,按照出现的顺序排列。

在文本中出现的顺序。

返回类型

列表[int]

invoke(input: LanguageModelInput, config: Optional[RunnableConfig] = None, **kwargs: Any) BaseMessage

将单个输入转换为输出。重写以实现。

参数
  • input (LanguageModelInput) – Runnable 的输入。

  • config (可选[RunnableConfig]) – 在调用Runnable时使用的配置。配置支持标准的键,如’tags’,‘metadata’用于跟踪目的,‘max_concurrency’用于控制并行执行的工作量,以及其他键。请参阅RunnableConfig以获取更多详细信息。

  • stop (可选列表[str]]) –

  • kwargs (任意类型) –

返回

Runnable 的输出。

返回类型

BaseMessage

predict(text: str, *, stop: Optional[Sequence[str]] = None, **kwargs: Any) str

自版本 langchain-core==0.1.7 起已弃用: 请使用 invoke 代替。

参数
  • text (str) –

  • stop (可选[字符串序列]) –

  • kwargs (任意类型) –

返回类型

str

predict_messages(messages: List[BaseMessage], *, stop: Optional[Sequence[str]] = None, **kwargs: Any) BaseMessage

自版本 langchain-core==0.1.7 起已弃用: 请使用 invoke 代替。

参数
  • messages (列表[BaseMessage]) –

  • stop (可选[字符串序列]) –

  • kwargs (任意类型) –

返回类型

BaseMessage

stream(input: LanguageModelInput, config: Optional[RunnableConfig] = None, *, stop: Optional[List[str]] = None, **kwargs: Any) Iterator[BaseMessageChunk]

stream的默认实现,调用invoke。子类应该覆盖此方法以支持流式输出。

参数
  • input (LanguageModelInput) – Runnable 的输入。

  • config (Optional[RunnableConfig]) – 用于 Runnable 的配置。默认为 None。

  • kwargs (Any) – 将传递给 Runnable 的额外关键字参数。

  • stop (可选列表[str]]) –

产出

Runnable 的输出。

返回类型

迭代器[BaseMessageChunk]

to_json() 联合[SerializedConstructor, SerializedNotImplemented]

将可运行对象序列化成JSON。

返回

可序列化的可运行对象表示。

返回类型

联合[SerializedConstructor, SerializedNotImplemented]

tokens_count(input_: List[str], model: 可选[str] = None) List[gm.TokensCount]

获取字符串列表的令牌。

参数
  • input_ (List[str]) –

  • model (Optional[str]) –

返回类型

List[gm.TokensCount]

with_structured_output(schema: 联合[Dict, Type], *, include_raw: bool = False, **kwargs: Any) Runnable[LanguageModelInput, 联合[Dict, BaseModel]]

返回符合给定模式输出的模型包装器。

参数
  • schema (联合[Dict, Type]) –

    输出模式。可以传入:
    • 一个OpenAI函数/工具模式,

    • 一个JSON模式,

    • 一个TypedDict类(支持添加在0.2.26),

    • 或一个Pydantic类。

    如果 schema 是一个 Pydantic 类,则模型输出将是该类的一个 Pydantic 实例,并且模型生成的字段将由 Pydantic 类进行验证。否则,模型输出将是字典,不会进行验证。有关如何指定 Pydantic 或 TypedDict 类的类型和描述的更多信息,请参阅 langchain_core.utils.function_calling.convert_to_openai_tool()

    在版本 0.2.26 中已更改:添加了对 TypedDict 类的支持。

  • include_raw (布尔值) – 如果为 False,则只返回解析后的结构化输出。如果在模型输出解析过程中发生错误,则将引发错误。如果为 True,则将返回原始模型响应(一个 BaseMessage)和解析后的模型响应。如果在输出解析过程中发生错误,它将被捕获并返回。最终输出始终是一个包含“raw”、“parsed”和“parsing_error”键的字典。

  • kwargs (任意类型) –

返回

一个 Runnable,它接受与 langchain_core.language_models.chat.BaseChatModel 相同的输入。

如果 include_raw 为 False 且 schema 是 Pydantic 类,Runnable 输出 schema 的一个实例(即 Pydantic 对象)。

否则,如果 include_raw 为 False,Runnable 则输出一个字典。

如果 include_raw 为 True,Runnable 将输出具有以下键的字典
  • "raw": BaseMessage

  • "parsed": 如果发生解析错误,则 None,否则类型取决于 schema,如上所述。

  • "parsing_error": Optional[BaseException]

返回类型

Runnable[LanguageModelInput, Union[Dict, BaseModel]]

示例:Pydantic 模式(include_raw=False)
from langchain_core.pydantic_v1 import BaseModel

class AnswerWithJustification(BaseModel):
    '''An answer to the user question along with justification for the answer.'''
    answer: str
    justification: str

llm = ChatModel(model="model-name", temperature=0)
structured_llm = llm.with_structured_output(AnswerWithJustification)

structured_llm.invoke("What weighs more a pound of bricks or a pound of feathers")

# -> AnswerWithJustification(
#     answer='They weigh the same',
#     justification='Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ.'
# )
示例:Pydantic 模式(include_raw=True)
from langchain_core.pydantic_v1 import BaseModel

class AnswerWithJustification(BaseModel):
    '''An answer to the user question along with justification for the answer.'''
    answer: str
    justification: str

llm = ChatModel(model="model-name", temperature=0)
structured_llm = llm.with_structured_output(AnswerWithJustification, include_raw=True)

structured_llm.invoke("What weighs more a pound of bricks or a pound of feathers")
# -> {
#     'raw': AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_Ao02pnFYXD6GN1yzc0uXPsvF', 'function': {'arguments': '{"answer":"They weigh the same.","justification":"Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ."}', 'name': 'AnswerWithJustification'}, 'type': 'function'}]}),
#     'parsed': AnswerWithJustification(answer='They weigh the same.', justification='Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ.'),
#     'parsing_error': None
# }
示例:字典模式(include_raw=False)
from langchain_core.pydantic_v1 import BaseModel
from langchain_core.utils.function_calling import convert_to_openai_tool

class AnswerWithJustification(BaseModel):
    '''An answer to the user question along with justification for the answer.'''
    answer: str
    justification: str

dict_schema = convert_to_openai_tool(AnswerWithJustification)
llm = ChatModel(model="model-name", temperature=0)
structured_llm = llm.with_structured_output(dict_schema)

structured_llm.invoke("What weighs more a pound of bricks or a pound of feathers")
# -> {
#     'answer': 'They weigh the same',
#     'justification': 'Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume and density of the two substances differ.'
# }
property lc_serializable: bool

使用 GigaChat 的示例