`langchain_community.agents.openai_assistant.base`.OpenAIAssistantV2Runnable¶

注意 (Note)

OpenAIAssistantV2Runnable 实现了标准的 Runnable 接口。 🏃

Runnable 接口 在 runnable 上还有其他可用方法，例如 with_types, with_retry, assign, bind, get_graph, 以及更多。

class langchain_community.agents.openai_assistant.base.OpenAIAssistantV2Runnable[源代码]¶

基类: OpenAIAssistantRunnable

Beta

此功能处于 Beta 阶段。它正在积极开发中，因此 API 可能会发生变化。

运行 OpenAI Assistant。

使用 OpenAI 工具的示例 (Example using OpenAI tools)

from langchain.agents.openai_assistant import OpenAIAssistantV2Runnable

interpreter_assistant = OpenAIAssistantV2Runnable.create_assistant(
    name="langchain assistant",
    instructions="You are a personal math tutor. Write and run code to answer math questions.",
    tools=[{"type": "code_interpreter"}],
    model="gpt-4-1106-preview"
)
output = interpreter_assistant.invoke({"content": "What's 10 - 4 raised to the 2.7"})

使用自定义工具和 AgentExecutor 的示例 (Example using custom tools and AgentExecutor)

from langchain.agents.openai_assistant import OpenAIAssistantV2Runnable
from langchain.agents import AgentExecutor
from langchain.tools import E2BDataAnalysisTool


tools = [E2BDataAnalysisTool(api_key="...")]
agent = OpenAIAssistantV2Runnable.create_assistant(
    name="langchain assistant e2b tool",
    instructions="You are a personal math tutor. Write and run code to answer math questions.",
    tools=tools,
    model="gpt-4-1106-preview",
    as_agent=True
)

agent_executor = AgentExecutor(agent=agent, tools=tools)
agent_executor.invoke({"content": "What's 10 - 4 raised to the 2.7"})

使用自定义工具和自定义执行的示例 (Example using custom tools and custom execution)

from langchain.agents.openai_assistant import OpenAIAssistantV2Runnable
from langchain.agents import AgentExecutor
from langchain_core.agents import AgentFinish
from langchain.tools import E2BDataAnalysisTool


tools = [E2BDataAnalysisTool(api_key="...")]
agent = OpenAIAssistantV2Runnable.create_assistant(
    name="langchain assistant e2b tool",
    instructions="You are a personal math tutor. Write and run code to answer math questions.",
    tools=tools,
    model="gpt-4-1106-preview",
    as_agent=True
)

def execute_agent(agent, tools, input):
    tool_map = {tool.name: tool for tool in tools}
    response = agent.invoke(input)
    while not isinstance(response, AgentFinish):
        tool_outputs = []
        for action in response:
            tool_output = tool_map[action.tool].invoke(action.tool_input)
            tool_outputs.append({"output": tool_output, "tool_call_id": action.tool_call_id})
        response = agent.invoke(
            {
                "tool_outputs": tool_outputs,
                "run_id": action.run_id,
                "thread_id": action.thread_id
            }
        )

    return response

response = execute_agent(agent, tools, {"content": "What's 10 - 4 raised to the 2.7"})
next_response = execute_agent(agent, tools, {"content": "now add 17.241", "thread_id": response.thread_id})

param as_agent: bool = False¶: 用作 LangChain 代理，与 AgentExecutor 兼容。(Use as a LangChain agent, compatible with the AgentExecutor.)

param assistant_id: str [必需]¶: OpenAI assistant id。

param async_client: Any = None¶: OpenAI 或 AzureOpenAI 异步客户端。(OpenAI or AzureOpenAI async client.)

param check_every_ms: float = 1000.0¶: 检查运行进度的频率，单位为毫秒。(Frequency with which to check run progress in ms.)

param client: Any [可选]¶: OpenAI 或 AzureOpenAI 客户端。(OpenAI or AzureOpenAI client.)

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

默认实现使用 asyncio.gather 并行运行 ainvoke。(Default implementation runs ainvoke in parallel using asyncio.gather.)

batch 的默认实现非常适用于 IO 绑定 runnable。(The default implementation of batch works well for IO bound runnables.)

如果子类可以更有效地进行批处理，则应覆盖此方法；例如，如果底层 Runnable 使用支持批处理模式的 API。(Subclasses should override this method if they can batch more efficiently; e.g., if the underlying Runnable uses an API which supports a batch mode.)

参数 (Parameters)

inputs (List[Input]) – Runnable 的输入列表。(A list of inputs to the Runnable.)
config (Optional[Union[RunnableConfig, List[RunnableConfig]]]) – 调用 Runnable 时要使用的配置。(A config to use when invoking the Runnable.) 该配置支持标准键，如 ‘tags’、‘metadata’ 用于跟踪目的，‘max_concurrency’ 用于控制并行执行的工作量以及其他键。(The config supports standard keys like ‘tags’, ‘metadata’ for tracing purposes, ‘max_concurrency’ for controlling how much work to do in parallel, and other keys.) 有关更多详细信息，请参阅 RunnableConfig。(Please refer to the RunnableConfig for more details.) 默认为 None。(Defaults to None.)
return_exceptions (bool) – 是否返回异常而不是引发异常。(Whether to return exceptions instead of raising them.) 默认为 False。(Defaults to False.)
kwargs (Optional[Any]) – 传递给 Runnable 的其他关键字参数。(Additional keyword arguments to pass to the Runnable.)

返回 (Returns)

来自 Runnable 的输出列表。(A list of outputs from the Runnable.)

返回类型 (Return type)

List[Output]

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

并行运行列表中输入的 ainvoke，并在完成后生成结果。(Run ainvoke in parallel on a list of inputs, yielding results as they complete.)

参数 (Parameters)

inputs (Sequence[Input]) – Runnable 的输入列表。(A list of inputs to the Runnable.)
config (Optional[Union[RunnableConfig, Sequence[RunnableConfig]]]) – 调用 Runnable 时要使用的配置。(A config to use when invoking the Runnable.) 该配置支持标准键，如 ‘tags’、‘metadata’ 用于跟踪目的，‘max_concurrency’ 用于控制并行执行的工作量以及其他键。(The config supports standard keys like ‘tags’, ‘metadata’ for tracing purposes, ‘max_concurrency’ for controlling how much work to do in parallel, and other keys.) 有关更多详细信息，请参阅 RunnableConfig。(Please refer to the RunnableConfig for more details.) 默认为 None。(Defaults to None.) 默认为 None。(Defaults to None.)
return_exceptions (bool) – 是否返回异常而不是引发异常。(Whether to return exceptions instead of raising them.) 默认为 False。(Defaults to False.)
kwargs (Optional[Any]) – 传递给 Runnable 的其他关键字参数。(Additional keyword arguments to pass to the Runnable.)

生成 (Yields)

输入索引和来自 Runnable 的输出的元组。(A tuple of the index of the input and the output from the Runnable.)

返回类型 (Return type)

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

async classmethod acreate_assistant(name: str, instructions: str, tools: Sequence[Union[BaseTool, dict]], model: str, *, async_client: Optional[Union[openai.AsyncOpenAI, openai.AsyncAzureOpenAI]] = None, tool_resources: Optional[Union[AssistantToolResources, dict, NotGiven]] = None, **kwargs: Any) → OpenAIAssistantRunnable[源代码]¶

创建 AsyncOpenAI Assistant 并实例化 Runnable。(Create an AsyncOpenAI Assistant and instantiate the Runnable.)

参数 (Parameters)

name (str) – Assistant 名称。(Assistant name.)
instructions (str) – Assistant 指令。(Assistant instructions.)
tools (Sequence[Union[BaseTool, dict]]) – Assistant 工具。(Assistant tools.) 可以以 OpenAI 格式或 BaseTools 形式传入。(Can be passed in OpenAI format or as BaseTools.)
tool_resources (Optional[Union[AssistantToolResources, dict, NotGiven]]) – Assistant 工具资源。(Assistant tool resources.) 可以以 OpenAI 格式传入 (Can be passed in OpenAI format)
model (str) – 要使用的 Assistant 模型。(Assistant model to use.)
async_client (Optional[Union[openai.AsyncOpenAI, openai.AsyncAzureOpenAI]]) – AsyncOpenAI 客户端。(AsyncOpenAI client.)
specified. (Will create default async_client if not) – (如果未指定，将创建默认 async_client。)
kwargs (Any) –

返回 (Returns)

AsyncOpenAIAssistantRunnable 配置为使用创建的 assistant 运行。(AsyncOpenAIAssistantRunnable configured to run using the created assistant.)

返回类型 (Return type)

OpenAIAssistantRunnable

async ainvoke(input: dict, config: Optional[RunnableConfig] = None, **kwargs: Any) → OutputType[源代码]¶

异步调用 assistant。(Async invoke assistant.)

参数 (Parameters)

input (dict) –
Runnable 输入字典，可以包含：content：启动新运行时用户消息。(Runnable input dict that can have: content: User message when starting a new run.) thread_id：要使用的现有线程。(thread_id: Existing thread to use.) run_id：要使用的现有运行。(run_id: Existing run to use.) 仅在为初始调用后所需的操作提供工具输出时才应提供。(Should only be supplied when providing the tool output for a required action after an initial invocation.)

file_ids：（已弃用）包含在新运行中的文件 id。(file_ids: (deprecated) File ids to include in new run.) 使用 ‘attachments’ 代替。(Use ‘attachments’ instead)

’attachments’ 代替 (use ‘attachments’ instead)
attachments：包含在新运行中的 Assistant 文件。(attachments: Assistant files to include in new run.) (v2 API)。message_metadata：与新消息关联的元数据。(message_metadata: Metadata to associate with new message.) thread_metadata：与新线程关联的元数据。(thread_metadata: Metadata to associate with new thread.) 仅在新线程被创建时相关。(Only relevant when new thread being created.)

当新线程被创建时相关。(when new thread being created.)

instructions：其他运行说明。(instructions: Additional run instructions.) model：覆盖此运行的 Assistant 模型。(model: Override Assistant model for this run.) tools：覆盖此运行的 Assistant 工具。(tools: Override Assistant tools for this run.) tool_resources：覆盖此运行的 Assistant 工具资源 (v2 API)。(tool_resources: Override Assistant tool resources for this run (v2 API).) run_metadata：与新运行关联的元数据。(run_metadata: Metadata to associate with new run.)

run_metadata：与新运行关联的元数据。(run_metadata: Metadata to associate with new run.)
config (Optional[RunnableConfig]) – Runnable 配置 (Runnable config)
kwargs (Any) –

返回 (Returns)

如果 self.as_agent，将返回 (If self.as_agent, will return): Union[List[OpenAIAssistantAction], OpenAIAssistantFinish]。否则，将返回 OpenAI 类型 Union[List[ThreadMessage], List[RequiredActionFunctionToolCall]]。(Otherwise, will return OpenAI types Union[List[ThreadMessage], List[RequiredActionFunctionToolCall]].)

返回类型 (Return type)

OutputType

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

Beta

此 API 处于 Beta 阶段，将来可能会发生变化。(This API is in beta and may change in the future.)

从 Runnable 创建 BaseTool。(Create a BaseTool from a Runnable.)

as_tool 将从 Runnable 实例化具有名称、描述和 args_schema 的 BaseTool。( as_tool will instantiate a BaseTool with a name, description, and args_schema from a Runnable.) 在可能的情况下，模式从 runnable.get_input_schema 推断。(Where possible, schemas are inferred from runnable.get_input_schema.) 或者（例如，如果 Runnable 接受字典作为输入，并且未键入特定的字典键），可以使用 args_schema 直接指定模式。(Alternatively (e.g., if the Runnable takes a dict as input and the specific dict keys are not typed), the schema can be specified directly with args_schema.) 您也可以传递 arg_types 以仅指定必需的参数及其类型。(You can also pass arg_types to just specify the required arguments and their types.)

参数 (Parameters)

args_schema (Optional[Type[BaseModel]]) – 工具的模式。(The schema for the tool.) 默认为 None。(Defaults to None.)
name (Optional[str]) – 工具的名称。(The name of the tool.) 默认为 None。(Defaults to None.)
description (Optional[str]) – 工具的描述。(The description of the tool.) 默认为 None。(Defaults to None.)
arg_types (Optional[Dict[str, Type]]) – 参数名称到类型的字典。(A dictionary of argument names to types.) 默认为 None。(Defaults to None.)

返回 (Returns)

BaseTool 实例。(A BaseTool instance.)

返回类型 (Return type)

BaseTool

类型化字典输入 (Typed dict input)

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 指定模式 ( dict input, specifying schema via 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 指定模式 ( dict input, specifying schema via 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]})

字符串输入 (String input)

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 版本中的新增功能。(New in version 0.2.14.)

async astream(input: Input, config: Optional[RunnableConfig] = None, **kwargs: Optional[Any]) → AsyncIterator[Output]¶

astream 的默认实现，它调用 ainvoke。(Default implementation of astream, which calls ainvoke.) 如果子类支持流式输出，则应覆盖此方法。(Subclasses should override this method if they support streaming output.)

参数 (Parameters)

input (Input) – Runnable 的输入。(The input to the Runnable.)
config (Optional[RunnableConfig]) – 用于 Runnable 的配置。(The config to use for the Runnable.) 默认为 None。(Defaults to None.)
kwargs (Optional[Any]) – 传递给 Runnable 的其他关键字参数。(Additional keyword arguments to pass to the Runnable.)

生成 (Yields)

Runnable 的输出。(The output of the Runnable.)

返回类型 (Return type)

AsyncIterator[Output]

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]]¶

Beta

此 API 处于 Beta 阶段，将来可能会发生变化。(This API is in beta and may change in the future.)

生成事件流。

用于创建一个迭代器，遍历 StreamEvents，这些 StreamEvents 提供关于 Runnable 进度的实时信息，包括来自中间结果的 StreamEvents。

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

event: str - 事件名称的格式为：
格式：on_[runnable_type]_(start|stream|end)。
name: str - 生成事件的 Runnable 的名称。
run_id: str - 随机生成的 ID，与给定 Runnable 执行相关联，该 Runnable 发出了事件。作为父 Runnable 执行一部分而被调用的子 Runnable 会被分配其自己唯一的 ID。
Runnable 发出事件的给定执行的关联随机生成 ID。作为父 Runnable 执行的一部分而调用的子 Runnable 将被分配其唯一的 ID。
parent_ids: List[str] - 生成事件的父 runnables 的 ID 列表。
根 Runnable 将有一个空列表。父 ID 的顺序是从根到直接父级。仅适用于 API 的 v2 版本。API 的 v1 版本将返回一个空列表。
tags: Optional[List[str]] - 生成事件的 Runnable 的标签。
事件的标签。
metadata: Optional[Dict[str, Any]] - 生成事件的 Runnable 的元数据。
生成事件的元数据。
data: Dict[str, Any]

下面是一个表格，说明了各种链可能发出的一些事件。为了简洁起见，元数据字段已从表格中省略。链定义已包含在表格之后。

注意此参考表适用于 V2 版本的模式。

event	name	chunk	input	output
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!, goodbye world!”
on_chain_end	format_docs		[Document(…)]	“hello world!, goodbye world!”
on_tool_start	some_tool		{“x”: 1, “y”: “2”}
on_tool_end	some_tool			{“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 版本中显示！

自定义事件具有以下格式

属性	类型	描述
name	str	用户为事件定义的名称。
data	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)

some_tool:

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

prompt:

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)

参数 (Parameters)

input (Any) – Runnable 的输入。
config (Optional[RunnableConfig]) – 用于 Runnable 的配置。
version (Literal['v1', 'v2']) – 要使用的模式版本，可以是 v2 或 v1。用户应使用 v2。v1 用于向后兼容，将在 0.4.0 中弃用。在 API 稳定之前，不会分配默认值。自定义事件将仅在 v2 中显示。
include_names (Optional[Sequence[str]]) – 仅包括来自具有匹配名称的 runnables 的事件。
include_types (Optional[Sequence[str]]) – 仅包括来自具有匹配类型的 runnables 的事件。
include_tags (Optional[Sequence[str]]) – 仅包括来自具有匹配标签的 runnables 的事件。
exclude_names (Optional[Sequence[str]]) – 排除来自具有匹配名称的 runnables 的事件。
exclude_types (Optional[Sequence[str]]) – 排除来自具有匹配类型的 runnables 的事件。
exclude_tags (Optional[Sequence[str]]) – 排除来自具有匹配标签的 runnables 的事件。
kwargs (Any) – 传递给 Runnable 的其他关键字参数。这些将传递给 astream_log，因为此 astream_events 的实现构建于 astream_log 之上。

生成 (Yields)

StreamEvents 的异步流。

Raises

NotImplementedError – 如果版本不是 v1 或 v2。

返回类型 (Return type)

AsyncIterator[Union[StandardStreamEvent, CustomStreamEvent]]

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

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

batch 的默认实现非常适用于 IO 绑定 runnable。(The default implementation of batch works well for IO bound runnables.)

如果子类可以更有效地进行批处理，则应覆盖此方法；例如，如果底层 Runnable 使用支持批处理模式的 API。(Subclasses should override this method if they can batch more efficiently; e.g., if the underlying Runnable uses an API which supports a batch mode.)

参数 (Parameters)

inputs (List[Input]) –
config (Optional[Union[RunnableConfig, List[RunnableConfig]]]) –
return_exceptions (bool) –
kwargs (Optional[Any]) –

返回类型 (Return type)

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]]]¶

并行运行列表中输入的 invoke，并在完成时产生结果。

参数 (Parameters)

inputs (Sequence[Input]) –
config (Optional[Union[RunnableConfig, Sequence[RunnableConfig]]]) –
return_exceptions (bool) –
kwargs (Optional[Any]) –

返回类型 (Return type)

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

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

配置可在运行时设置的 Runnables 的备选项。

参数 (Parameters)

which (ConfigurableField) – 将用于选择备选项的 ConfigurableField 实例。
default_key (str) – 如果未选择备选项，则使用的默认键。默认为“default”。
prefix_keys (bool) – 是否用 ConfigurableField id 作为键的前缀。默认为 False。
**kwargs (Union[Runnable[Input, Output], Callable[[], Runnable[Input, Output]]]) – 键到 Runnable 实例或返回 Runnable 实例的可调用对象的字典。

返回 (Returns)

配置了备选项的新 Runnable。

返回类型 (Return type)

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 字段。

参数 (Parameters): **kwargs (Union[ConfigurableField, ConfigurableFieldSingleOption, ConfigurableFieldMultiOption]) – 要配置的 ConfigurableField 实例的字典。
返回 (Returns): 配置了字段的新 Runnable。
返回类型 (Return type): 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
)

classmethod create_assistant(name: str, instructions: str, tools: Sequence[Union[BaseTool, dict]], model: str, *, client: Optional[Union[openai.OpenAI, openai.AzureOpenAI]] = None, tool_resources: Optional[Union[AssistantToolResources, dict, NotGiven]] = None, **kwargs: Any) → OpenAIAssistantRunnable[source]¶

创建一个 OpenAI Assistant 并实例化 Runnable。

参数 (Parameters)

name (str) – Assistant 名称。(Assistant name.)
instructions (str) – Assistant 指令。(Assistant instructions.)
tools (Sequence[Union[BaseTool, dict]]) – Assistant 工具。(Assistant tools.) 可以以 OpenAI 格式或 BaseTools 形式传入。(Can be passed in OpenAI format or as BaseTools.)
tool_resources (Optional[Union[AssistantToolResources, dict, NotGiven]]) – Assistant 工具资源。(Assistant tool resources.) 可以以 OpenAI 格式传入 (Can be passed in OpenAI format)
model (str) – 要使用的 Assistant 模型。(Assistant model to use.)
client (Optional[Union[openai.OpenAI, openai.AzureOpenAI]]) – OpenAI 或 AzureOpenAI 客户端。如果未指定，将创建默认的 OpenAI 客户端 (Assistant v2)。
kwargs (Any) –

返回 (Returns)

配置为使用创建的助手运行的 OpenAIAssistantRunnable。

返回类型 (Return type)

OpenAIAssistantRunnable

invoke(input: dict, config: Optional[RunnableConfig] = None, **kwargs: Any) → OutputType[source]¶

调用助手。

参数 (Parameters)

input (dict) –
Runnable 输入字典，可以包含：content：启动新运行时用户消息。(Runnable input dict that can have: content: User message when starting a new run.) thread_id：要使用的现有线程。(thread_id: Existing thread to use.) run_id：要使用的现有运行。(run_id: Existing run to use.) 仅在为初始调用后所需的操作提供工具输出时才应提供。(Should only be supplied when providing the tool output for a required action after an initial invocation.)

file_ids：（已弃用）包含在新运行中的文件 id。(file_ids: (deprecated) File ids to include in new run.) 使用 ‘attachments’ 代替。(Use ‘attachments’ instead)

’attachments’ 代替 (use ‘attachments’ instead)
attachments：包含在新运行中的 Assistant 文件。(attachments: Assistant files to include in new run.) (v2 API)。message_metadata：与新消息关联的元数据。(message_metadata: Metadata to associate with new message.) thread_metadata：与新线程关联的元数据。(thread_metadata: Metadata to associate with new thread.) 仅在新线程被创建时相关。(Only relevant when new thread being created.)

当新线程被创建时相关。(when new thread being created.)

instructions：其他运行说明。(instructions: Additional run instructions.) model：覆盖此运行的 Assistant 模型。(model: Override Assistant model for this run.) tools：覆盖此运行的 Assistant 工具。(tools: Override Assistant tools for this run.) tool_resources：覆盖此运行的 Assistant 工具资源 (v2 API)。(tool_resources: Override Assistant tool resources for this run (v2 API).) run_metadata：与新运行关联的元数据。(run_metadata: Metadata to associate with new run.)

run_metadata：与新运行关联的元数据。(run_metadata: Metadata to associate with new run.)
config (Optional[RunnableConfig]) – Runnable 配置 (Runnable config)
kwargs (Any) –

返回 (Returns)

如果 self.as_agent，将返回 (If self.as_agent, will return): Union[List[OpenAIAssistantAction], OpenAIAssistantFinish]。否则，将返回 OpenAI 类型 Union[List[ThreadMessage], List[RequiredActionFunctionToolCall]]。(Otherwise, will return OpenAI types Union[List[ThreadMessage], List[RequiredActionFunctionToolCall]].)

Raises

BaseException – 如果在调用期间发生错误。

返回类型 (Return type)

OutputType

stream(input: Input, config: Optional[RunnableConfig] = None, **kwargs: Optional[Any]) → Iterator[Output]¶

流的默认实现，它调用 invoke。如果子类支持流式输出，则应重写此方法。

参数 (Parameters)

input (Input) – Runnable 的输入。(The input to the Runnable.)
config (Optional[RunnableConfig]) – 用于 Runnable 的配置。(The config to use for the Runnable.) 默认为 None。(Defaults to None.)
kwargs (Optional[Any]) – 传递给 Runnable 的其他关键字参数。(Additional keyword arguments to pass to the Runnable.)

生成 (Yields)

Runnable 的输出。(The output of the Runnable.)

返回类型 (Return type)

Iterator[Output]

to_json() → Union[SerializedConstructor, SerializedNotImplemented]¶

将 Runnable 序列化为 JSON。

返回 (Returns): Runnable 的 JSON 序列化表示形式。
返回类型 (Return type): Union[SerializedConstructor, SerializedNotImplemented]

langchain_community.agents.openai_assistant.base.OpenAIAssistantV2Runnable¶

`langchain_community.agents.openai_assistant.base`.OpenAIAssistantV2Runnable¶