`ChatAnthropic`¶

参考文档

此页面包含 ChatAnthropic 的参考文档。有关使用 ChatAnthropic 的概念指南、教程和示例，请参阅文档。

langchain_anthropic.chat_models.ChatAnthropic ¶

基类：BaseChatModel

Anthropic 聊天模型。

有关最新模型的列表，请参阅 Anthropic 的文档。

设置

安装 langchain-anthropic 并设置环境变量 ANTHROPIC_API_KEY。

pip install -U langchain-anthropic
export ANTHROPIC_API_KEY="your-api-key"

关键初始化参数 — 完成参数：model: 要使用的 Anthropic 模型名称。例如 'claude-sonnet-4-5-20250929'。temperature: 采样温度。范围从 0.0 到 1.0。max_tokens: 要生成的最大令牌数。

关键初始化参数 — 客户端参数：timeout: 请求超时时间。anthropic_proxy: 用于 Anthropic 客户端的代理，将用于每次 API 调用。如果未传入，将从环境变量 ANTHROPIC_PROXY 中读取。max_retries: 请求失败时的最大重试次数。api_key: Anthropic API 密钥。如果未传入，将从环境变量 ANTHROPIC_API_KEY 中读取。base_url: API 请求的基础 URL。仅在使用代理或服务模拟器时指定。

有关支持的初始化参数及其描述的完整列表，请参见参数部分。

实例化

from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(
    model="claude-sonnet-4-5-20250929",
    temperature=0,
    max_tokens=1024,
    timeout=None,
    max_retries=2,
    # api_key="...",
    # base_url="...",
    # other params...
)

注意

任何未明确支持的参数将在每次调用模型时直接传递给 anthropic.Anthropic.messages.create(...) API。例如

from langchain_anthropic import ChatAnthropic
import anthropic

ChatAnthropic(..., extra_headers={}).invoke(...)

# results in underlying API call of:

anthropic.Anthropic(..).messages.create(..., extra_headers={})

# which is also equivalent to:

ChatAnthropic(...).invoke(..., extra_headers={})

调用

messages = [
    (
        "system",
        "You are a helpful translator. Translate the user sentence to French.",
    ),
    ("human", "I love programming."),
]
model.invoke(messages)

AIMessage(
    content="J'aime la programmation.",
    response_metadata={
        "id": "msg_01Trik66aiQ9Z1higrD5XFx3",
        "model": "claude-sonnet-4-5-20250929",
        "stop_reason": "end_turn",
        "stop_sequence": None,
        "usage": {"input_tokens": 25, "output_tokens": 11},
    },
    id="run-5886ac5f-3c2e-49f5-8a44-b1e92808c929-0",
    usage_metadata={
        "input_tokens": 25,
        "output_tokens": 11,
        "total_tokens": 36,
    },
)

流

for chunk in model.stream(messages):
    print(chunk.text, end="")

AIMessageChunk(content="J", id="run-272ff5f9-8485-402c-b90d-eac8babc5b25")
AIMessageChunk(content="'", id="run-272ff5f9-8485-402c-b90d-eac8babc5b25")
AIMessageChunk(content="a", id="run-272ff5f9-8485-402c-b90d-eac8babc5b25")
AIMessageChunk(content="ime", id="run-272ff5f9-8485-402c-b90d-eac8babc5b25")
AIMessageChunk(content=" la", id="run-272ff5f9-8485-402c-b90d-eac8babc5b25")
AIMessageChunk(content=" programm", id="run-272ff5f9-8485-402c-b90d-eac8babc5b25")
AIMessageChunk(content="ation", id="run-272ff5f9-8485-402c-b90d-eac8babc5b25")
AIMessageChunk(content=".", id="run-272ff5f9-8485-402c-b90d-eac8babc5b25")

stream = model.stream(messages)
full = next(stream)
for chunk in stream:
    full += chunk
full

AIMessageChunk(content="J'aime la programmation.", id="run-b34faef0-882f-4869-a19c-ed2b856e6361")

异步

await model.ainvoke(messages)

# stream:
# async for chunk in (await model.astream(messages))

# batch:
# await model.abatch([messages])

AIMessage(
    content="J'aime la programmation.",
    response_metadata={
        "id": "msg_01Trik66aiQ9Z1higrD5XFx3",
        "model": "claude-sonnet-4-5-20250929",
        "stop_reason": "end_turn",
        "stop_sequence": None,
        "usage": {"input_tokens": 25, "output_tokens": 11},
    },
    id="run-5886ac5f-3c2e-49f5-8a44-b1e92808c929-0",
    usage_metadata={
        "input_tokens": 25,
        "output_tokens": 11,
        "total_tokens": 36,
    },
)

工具调用

from pydantic import BaseModel, Field


class GetWeather(BaseModel):
    '''Get the current weather in a given location'''

    location: str = Field(..., description="The city and state, e.g. San Francisco, CA")


class GetPopulation(BaseModel):
    '''Get the current population in a given location'''

    location: str = Field(..., description="The city and state, e.g. San Francisco, CA")


model_with_tools = model.bind_tools([GetWeather, GetPopulation])
ai_msg = model_with_tools.invoke("Which city is hotter today and which is bigger: LA or NY?")
ai_msg.tool_calls

[
    {
        "name": "GetWeather",
        "args": {"location": "Los Angeles, CA"},
        "id": "toolu_01KzpPEAgzura7hpBqwHbWdo",
    },
    {
        "name": "GetWeather",
        "args": {"location": "New York, NY"},
        "id": "toolu_01JtgbVGVJbiSwtZk3Uycezx",
    },
    {
        "name": "GetPopulation",
        "args": {"location": "Los Angeles, CA"},
        "id": "toolu_01429aygngesudV9nTbCKGuw",
    },
    {
        "name": "GetPopulation",
        "args": {"location": "New York, NY"},
        "id": "toolu_01JPktyd44tVMeBcPPnFSEJG",
    },
]

更多信息请参阅 ChatAnthropic.bind_tools() 方法。

结构化输出

from typing import Optional

from pydantic import BaseModel, Field


class Joke(BaseModel):
    '''Joke to tell user.'''

    setup: str = Field(description="The setup of the joke")
    punchline: str = Field(description="The punchline to the joke")
    rating: int | None = Field(description="How funny the joke is, from 1 to 10")


structured_model = model.with_structured_output(Joke)
structured_model.invoke("Tell me a joke about cats")

Joke(
    setup="Why was the cat sitting on the computer?",
    punchline="To keep an eye on the mouse!",
    rating=None,
)

更多信息请参阅 ChatAnthropic.with_structured_output()。

图像输入

更多详情请参阅多模态指南。

import base64

import httpx
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage

image_url = "https://upload.wikimedia.org/wikipedia/commons/thumb/d/dd/Gfp-wisconsin-madison-the-nature-boardwalk.jpg/2560px-Gfp-wisconsin-madison-the-nature-boardwalk.jpg"
image_data = base64.b64encode(httpx.get(image_url).content).decode("utf-8")

model = ChatAnthropic(model="claude-sonnet-4-5-20250929")
message = HumanMessage(
    content=[
        {
            "type": "text",
            "text": "Can you highlight the differences between these two images?",
        },
        {
            "type": "image",
            "base64": image_data,
            "mime_type": "image/jpeg",
        },
        {
            "type": "image",
            "url": image_url,
        },
    ],
)
ai_msg = model.invoke([message])
ai_msg.content

"After examining both images carefully, I can see that they are actually identical."

文件 API

您还可以传入通过 Anthropic 的文件 API管理的文件。

from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(
    model="claude-sonnet-4-5-20250929",
    betas=["files-api-2025-04-14"],
)
input_message = {
    "role": "user",
    "content": [
        {
            "type": "text",
            "text": "Describe this document.",
        },
        {
            "type": "image",
            "id": "file_abc123...",
        },
    ],
}
model.invoke([input_message])

PDF 输入

更多详情请参阅多模态指南。

from base64 import b64encode
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage
import requests

url = "https://www.w3.org/WAI/ER/tests/xhtml/testfiles/resources/pdf/dummy.pdf"
data = b64encode(requests.get(url).content).decode()

model = ChatAnthropic(model="claude-sonnet-4-5-20250929")
ai_msg = model.invoke(
    [
        HumanMessage(
            [
                "Summarize this document.",
                {
                    "type": "file",
                    "mime_type": "application/pdf",
                    "base64": data,
                },
            ]
        )
    ]
)
ai_msg.content

"This appears to be a simple document..."

文件 API

您还可以传入通过 Anthropic 的文件 API管理的文件。

from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(
    model="claude-sonnet-4-5-20250929",
    betas=["files-api-2025-04-14"],
)
input_message = {
    "role": "user",
    "content": [
        {
            "type": "text",
            "text": "Describe this document.",
        },
        {
            "type": "file",
            "id": "file_abc123...",
        },
    ],
}
model.invoke([input_message])

扩展思考

某些 Claude 模型支持扩展思考功能，它将输出导致最终答案的逐步推理过程。

要使用它，请在初始化 ChatAnthropic 时指定 thinking 参数。

它也可以在调用时作为关键字参数传入。

您需要指定一个令牌预算才能使用此功能。请参阅使用示例

from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(
    model="claude-sonnet-4-5-20250929",
    max_tokens=5000,
    thinking={"type": "enabled", "budget_tokens": 2000},
)

response = model.invoke("What is the cube root of 50.653?")
response.content

[
    {
        "signature": "...",
        "thinking": "To find the cube root of 50.653...",
        "type": "thinking",
    },
    {"text": "The cube root of 50.653 is ...", "type": "text"},
]

不同模型版本中思考的差异

Claude Messages API 在 Claude Sonnet 3.7 和 Claude 4 模型中处理思考的方式不同。更多信息请参考他们的文档。

引用

Anthropic 支持引用功能，该功能允许 Claude 根据用户提供的源文档为其答案附加上下文。当查询中包含带有 "citations": {"enabled": True} 的文档内容块时，Claude 可能会在其响应中生成引用。

from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(model="claude-3-5-haiku-20241022")

messages = [
    {
        "role": "user",
        "content": [
            {
                "type": "document",
                "source": {
                    "type": "text",
                    "media_type": "text/plain",
                    "data": "The grass is green. The sky is blue.",
                },
                "title": "My Document",
                "context": "This is a trustworthy document.",
                "citations": {"enabled": True},
            },
            {"type": "text", "text": "What color is the grass and sky?"},
        ],
    }
]
response = model.invoke(messages)
response.content

[
    {"text": "Based on the document, ", "type": "text"},
    {
        "text": "the grass is green",
        "type": "text",
        "citations": [
            {
                "type": "char_location",
                "cited_text": "The grass is green. ",
                "document_index": 0,
                "document_title": "My Document",
                "start_char_index": 0,
                "end_char_index": 20,
            }
        ],
    },
    {"text": ", and ", "type": "text"},
    {
        "text": "the sky is blue",
        "type": "text",
        "citations": [
            {
                "type": "char_location",
                "cited_text": "The sky is blue.",
                "document_index": 0,
                "document_title": "My Document",
                "start_char_index": 20,
                "end_char_index": 36,
            }
        ],
    },
    {"text": ".", "type": "text"},
]

令牌使用情况

ai_msg = model.invoke(messages)
ai_msg.usage_metadata

{"input_tokens": 25, "output_tokens": 11, "total_tokens": 36}

默认情况下，包含令牌使用情况的消息块将在流式传输期间包含在内。

stream = model.stream(messages)
full = next(stream)
for chunk in stream:
    full += chunk
full.usage_metadata

{"input_tokens": 25, "output_tokens": 11, "total_tokens": 36}

可以通过在流方法中设置 stream_usage=False，或在初始化 ChatAnthropic 时设置 stream_usage=False 来禁用这些消息块。

提示缓存

提示缓存减少了重复性任务或具有一致元素的提示的处理时间和成本。

注意

只有某些模型支持提示缓存。有关完整列表，请参阅 Claude 文档。

from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(model="claude-sonnet-4-5-20250929")

messages = [
    {
        "role": "system",
        "content": [
            {
                "type": "text",
                "text": "Below is some long context:",
            },
            {
                "type": "text",
                "text": f"{long_text}",
                "cache_control": {"type": "ephemeral"},
            },
        ],
    },
    {
        "role": "user",
        "content": "What's that about?",
    },
]

response = model.invoke(messages)
response.usage_metadata["input_token_details"]

{"cache_read": 0, "cache_creation": 1458}

或者，您可以在调用时启用提示缓存。您可能希望根据运行时条件（例如上下文的长度）有条件地进行缓存。此外，这对于应用层级决定缓存什么内容很有用。

response = model.invoke(
    messages,
    cache_control={"type": "ephemeral"},
)

扩展缓存

缓存生命周期默认为 5 分钟。如果这太短，您可以通过将 ttl 设置为 '1h' 来应用一小时缓存。

model = ChatAnthropic(
    model="claude-sonnet-4-5-20250929",
)

messages = [
    {
        "role": "user",
        "content": [
            {
                "type": "text",
                "text": f"{long_text}",
                "cache_control": {"type": "ephemeral", "ttl": "1h"},
            },
        ],
    }
]

response = model.invoke(messages)

缓存的令牌计数的详细信息将包含在响应的 usage_metadata 的 InputTokenDetails 中。

response = model.invoke(messages)
response.usage_metadata

{
    "input_tokens": 1500,
    "output_tokens": 200,
    "total_tokens": 1700,
    "input_token_details": {
        "cache_read": 0,
        "cache_creation": 1000,
        "ephemeral_1h_input_tokens": 750,
        "ephemeral_5m_input_tokens": 250,
    },
}

详情请参阅 Claude 文档。

!!! note title="扩展上下文窗口 (beta)"

Claude Sonnet 4 supports a 1-million token context window, available in beta for
organizations in usage tier 4 and organizations with custom rate limits.

```python
from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(
    model="claude-sonnet-4-5-20250929",
    betas=["context-1m-2025-08-07"],  # Enable 1M context beta
)

long_document = """
This is a very long document that would benefit from the extended 1M
context window...
[imagine this continues for hundreds of thousands of tokens]
"""

messages = [
    HumanMessage(f"""
Please analyze this document and provide a summary:

{long_document}

What are the key themes and main conclusions?
""")
]

response = model.invoke(messages)
```

See [Claude documentation](https://docs.claude.com/en/docs/build-with-claude/context-windows#1m-token-context-window)
for detail.

!!! note title="高令牌效率的工具使用 (beta)"

See LangChain [docs](https://docs.langchain.org.cn/oss/python/integrations/chat/anthropic)
for more detail.

```python
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool

model = ChatAnthropic(
    model="claude-sonnet-4-5-20250929",
    temperature=0,
    model_kwargs={
        "extra_headers": {
            "anthropic-beta": "token-efficient-tools-2025-02-19"
        }
    }
)

@tool
def get_weather(location: str) -> str:
    """Get the weather at a location."""
    return "It's sunny."

model_with_tools = model.bind_tools([get_weather])
response = model_with_tools.invoke(
    "What's the weather in San Francisco?"
)
print(response.tool_calls)
print(f'Total tokens: {response.usage_metadata["total_tokens"]}')
```

```txt
[{'name': 'get_weather', 'args': {'location': 'San Francisco'}, 'id': 'toolu_01HLjQMSb1nWmgevQUtEyz17', 'type': 'tool_call'}]
Total tokens: 408
```

!!! note title="上下文管理"

Anthropic supports a context editing feature that will automatically manage the
model's context window (e.g., by clearing tool results).

See [Anthropic documentation](https://docs.claude.com/en/docs/build-with-claude/context-editing)
for details and configuration options.

```python
from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(
    model="claude-sonnet-4-5-20250929",
    betas=["context-management-2025-06-27"],
    context_management={"edits": [{"type": "clear_tool_uses_20250919"}]},
)
model_with_tools = model.bind_tools([{"type": "web_search_20250305", "name": "web_search"}])
response = model_with_tools.invoke("Search for recent developments in AI")
```

!!! note title="内置工具"

See LangChain [docs](https://docs.langchain.org.cn/oss/python/integrations/chat/anthropic#built-in-tools)
for more detail.

??? note "Web search"

    ```python
    from langchain_anthropic import ChatAnthropic

    model = ChatAnthropic(model="claude-3-5-haiku-20241022")

    tool = {
        "type": "web_search_20250305",
        "name": "web_search",
        "max_uses": 3,
    }
    model_with_tools = model.bind_tools([tool])

    response = model_with_tools.invoke("How do I update a web app to TypeScript 5.5?")
    ```

??? note "Web fetch (beta)"

    ```python
    from langchain_anthropic import ChatAnthropic

    model = ChatAnthropic(
        model="claude-3-5-haiku-20241022",
        betas=["web-fetch-2025-09-10"],  # Enable web fetch beta
    )

    tool = {
        "type": "web_fetch_20250910",
        "name": "web_fetch",
        "max_uses": 3,
    }
    model_with_tools = model.bind_tools([tool])

    response = model_with_tools.invoke("Please analyze the content at https://example.com/article")
    ```

??? note "Code execution"

    ```python
    model = ChatAnthropic(
        model="claude-sonnet-4-5-20250929",
        betas=["code-execution-2025-05-22"],
    )

    tool = {"type": "code_execution_20250522", "name": "code_execution"}
    model_with_tools = model.bind_tools([tool])

    response = model_with_tools.invoke(
        "Calculate the mean and standard deviation of [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]"
    )
    ```

??? note "Remote MCP"

    ```python
    from langchain_anthropic import ChatAnthropic

    mcp_servers = [
        {
            "type": "url",
            "url": "https://mcp.deepwiki.com/mcp",
            "name": "deepwiki",
            "tool_configuration": {  # optional configuration
                "enabled": True,
                "allowed_tools": ["ask_question"],
            },
            "authorization_token": "PLACEHOLDER",  # optional authorization
        }
    ]

    model = ChatAnthropic(
        model="claude-sonnet-4-5-20250929",
        betas=["mcp-client-2025-04-04"],
        mcp_servers=mcp_servers,
    )

    response = model.invoke(
        "What transport protocols does the 2025-03-26 version of the MCP "
        "spec (modelcontextprotocol/modelcontextprotocol) support?"
    )
    ```

??? note "Text editor"

    ```python
    from langchain_anthropic import ChatAnthropic

    model = ChatAnthropic(model="claude-sonnet-4-5-20250929")

    tool = {"type": "text_editor_20250124", "name": "str_replace_editor"}
    model_with_tools = model.bind_tools([tool])

    response = model_with_tools.invoke(
        "There's a syntax error in my primes.py file. Can you help me fix it?"
    )
    print(response.text)
    response.tool_calls
    ```

    ```txt
    I'd be happy to help you fix the syntax error in your primes.py file. First, let's look at the current content of the file to identify the error.

    [{'name': 'str_replace_editor',
    'args': {'command': 'view', 'path': '/repo/primes.py'},
    'id': 'toolu_01VdNgt1YV7kGfj9LFLm6HyQ',
    'type': 'tool_call'}]
    ```

??? note "Memory tool"

    ```python
    from langchain_anthropic import ChatAnthropic

    model = ChatAnthropic(
        model="claude-sonnet-4-5-20250929",
        betas=["context-management-2025-06-27"],
    )
    model_with_tools = model.bind_tools([{"type": "memory_20250818", "name": "memory"}])
    response = model_with_tools.invoke("What are my interests?")
    ```

!!! note title="响应元数据"

```python
ai_msg = model.invoke(messages)
ai_msg.response_metadata
```

```python
{
    "id": "msg_013xU6FHEGEq76aP4RgFerVT",
    "model": "claude-sonnet-4-5-20250929",
    "stop_reason": "end_turn",
    "stop_sequence": None,
    "usage": {"input_tokens": 25, "output_tokens": 11},
}
```

方法	描述
`is_lc_serializable`	该类是否可在 langchain 中序列化。
`get_lc_namespace`	获取 LangChain 对象的命名空间。
`set_default_max_tokens`	设置默认的 max_tokens。
`build_extra`	构建模型关键字参数。
`bind_tools`	将类工具（tool-like）对象绑定到此聊天模型。
`with_structured_output`	返回与给定模式匹配的格式化输出的模型包装器。
`get_num_tokens_from_messages`	计算输入消息序列中的令牌数。
`get_name`	获取 `Runnable` 的名称。
`get_input_schema`	获取可用于验证 `Runnable` 输入的 Pydantic 模型。
`get_input_jsonschema`	获取表示 `Runnable` 输入的 JSON 模式。
`get_output_schema`	获取可用于验证 `Runnable` 输出的 Pydantic 模型。
`get_output_jsonschema`	获取表示 `Runnable` 输出的 JSON 模式。
`config_schema`	此 `Runnable` 接受的配置类型，指定为 Pydantic 模型。
`get_config_jsonschema`	获取表示 `Runnable` 配置的 JSON 模式。
`get_graph`	返回此 `Runnable` 的图形表示。
`get_prompts`	返回此 `Runnable` 使用的提示列表。
`__or__`	Runnable "or" 运算符。
`__ror__`	Runnable "reverse-or" 运算符。
`pipe`	管道连接 `Runnable` 对象。
`pick`	从此 `Runnable` 的输出 `dict` 中选择键。
`assign`	向此 `Runnable` 的 `dict` 输出分配新字段。
`invoke`	将单个输入转换为输出。
`ainvoke`	将单个输入转换为输出。
`batch`	默认实现使用线程池执行器并行运行 invoke。
`batch_as_completed`	在输入列表上并行运行 `invoke`。
`abatch`	默认实现使用 `asyncio.gather` 并行运行 `ainvoke`。
`abatch_as_completed`	在输入列表上并行运行 `ainvoke`。
`stream`	`stream` 的默认实现，它调用 `invoke`。
`astream`	`astream` 的默认实现，它调用 `ainvoke`。
`astream_log`	流式传输 `Runnable` 的所有输出，如回调系统所报告。
`astream_events`	生成事件流。
`transform`	将输入转换为输出。
`atransform`	将输入转换为输出。
`bind`	将参数绑定到 `Runnable`，返回一个新的 `Runnable`。
`with_config`	将配置绑定到 `Runnable`，返回一个新的 `Runnable`。
`with_listeners`	将生命周期侦听器绑定到 `Runnable`，返回一个新的 `Runnable`。
`with_alisteners`	将异步生命周期侦听器绑定到 `Runnable`。
`with_types`	将输入和输出类型绑定到 `Runnable`，返回一个新的 `Runnable`。
`with_retry`	创建一个新的 `Runnable`，它在发生异常时重试原始的 `Runnable`。
`map`	返回一个新的 `Runnable`，它将输入列表映射到输出列表。
`with_fallbacks`	向 `Runnable` 添加回退机制，返回一个新的 `Runnable`。
`as_tool`	从 `Runnable` 创建一个 `BaseTool`。
`__init__`
`lc_id`	为此类返回一个用于序列化目的的唯一标识符。
`to_json`	将 `Runnable` 序列化为 JSON。
`to_json_not_implemented`	序列化一个“未实现”的对象。
`configurable_fields`	在运行时配置特定的 `Runnable` 字段。
`configurable_alternatives`	为可在运行时设置的 `Runnable` 对象配置备选项。
`set_verbose`	如果 verbose 是 `None`，则设置它。
`generate_prompt`	将一系列提示传递给模型并返回模型生成的内容。
`agenerate_prompt`	异步地将一系列提示传递并返回模型生成的内容。
`get_token_ids`	返回文本中 token 的有序 ID。
`get_num_tokens`	获取文本中存在的 token 数量。
`generate`	将一系列提示传递给模型并返回模型生成的内容。
`agenerate`	异步地将一系列提示传递给模型并返回生成的内容。
`dict`	返回 LLM 的字典。

model `类属性` `实例属性` ¶

model: str = Field(alias='model_name')

要使用的模型名称。

max_tokens `类属性` `实例属性` ¶

max_tokens: int | None = Field(default=None, alias='max_tokens_to_sample')

表示每次生成要预测的令牌数。

temperature `类属性` `实例属性` ¶

temperature: float | None = None

一个非负浮点数，用于调整生成中的随机性程度。

top_k `类属性` `实例属性` ¶

top_k: int | None = None

每一步要考虑的最可能令牌的数量。

top_p `类属性` `实例属性` ¶

top_p: float | None = None

每一步要考虑的令牌的总概率质量。

default_request_timeout `类属性` `实例属性` ¶

default_request_timeout: float | None = Field(None, alias='timeout')

向 Anthropic Completion API 发送请求的超时时间。

max_retries `类属性` `实例属性` ¶

max_retries: int = 2

发送到 Anthropic Completion API 的请求允许的重试次数。

stop_sequences `类属性` `实例属性` ¶

stop_sequences: list[str] | None = Field(None, alias='stop')

默认的停止序列。

anthropic_api_url `类属性` `实例属性` ¶

anthropic_api_url: str | None = Field(
    alias="base_url",
    default_factory=from_env(
        ["ANTHROPIC_API_URL", "ANTHROPIC_BASE_URL"], default="https://api.anthropic.com"
    ),
)

API 请求的基础 URL。仅在使用代理或服务模拟器时指定。

如果未传入值，将首先尝试从 ANTHROPIC_API_URL 读取值，如果未设置，则从 ANTHROPIC_BASE_URL 读取。如果两者均未设置，将使用默认值 https://api.anthropic.com。

anthropic_api_key `类属性` `实例属性` ¶

anthropic_api_key: SecretStr = Field(
    alias="api_key", default_factory=secret_from_env("ANTHROPIC_API_KEY", default="")
)

如果未提供，将自动从环境变量 ANTHROPIC_API_KEY 中读取。

anthropic_proxy `类属性` `实例属性` ¶

anthropic_proxy: str | None = Field(
    default_factory=from_env("ANTHROPIC_PROXY", default=None)
)

用于 Anthropic 客户端的代理，将用于每次 API 调用。

如果未提供，将尝试从 ANTHROPIC_PROXY 环境变量中读取。

default_headers `类属性` `实例属性` ¶

default_headers: Mapping[str, str] | None = None

传递给 Anthropic 客户端的标头，将用于每次 API 调用。

betas `类属性` `实例属性` ¶

betas: list[str] | None = None

要启用的 beta 功能列表。如果指定，调用将被路由到 client.beta.messages.create。

示例：betas=["mcp-client-2025-04-04"]

streaming `类属性` `实例属性` ¶

streaming: bool = False

是否使用流式传输。

stream_usage `类属性` `实例属性` ¶

stream_usage: bool = True

是否在流式输出中包含使用元数据。如果为 True，则在流期间将生成包含使用元数据的额外消息块。

thinking `类属性` `实例属性` ¶

thinking: dict[str, Any] | None = Field(default=None)

用于 Claude 推理的参数，例如 {"type": "enabled", "budget_tokens": 10_000}

mcp_servers `类属性` `实例属性` ¶

mcp_servers: list[dict[str, Any]] | None = None

用于请求的 MCP 服务器列表。

示例：mcp_servers=[{"type": "url", "url": "https://mcp.example.com/mcp", "name": "example-mcp"}]

context_management `类属性` `实例属性` ¶

context_management: dict[str, Any] | None = None

用于上下文管理的配置。

lc_secrets `属性` ¶

lc_secrets: dict[str, str]

返回一个秘密密钥到环境变量的映射。

name `类属性` `实例属性` ¶

name: str | None = None

Runnable 的名称。用于调试和追踪。

InputType `属性` ¶

InputType: TypeAlias

获取此 Runnable 的输入类型。

OutputType `属性` ¶

OutputType: Any

获取此 Runnable 的输出类型。

input_schema `属性` ¶

input_schema: type[BaseModel]

此 Runnable 接受的输入类型，指定为 Pydantic 模型。

output_schema `属性` ¶

output_schema: type[BaseModel]

输出模式。

此 Runnable 产生的输出类型，指定为 Pydantic 模型。

config_specs `属性` ¶

config_specs: list[ConfigurableFieldSpec]

列出此 Runnable 的可配置字段。

lc_attributes `属性` ¶

lc_attributes: dict

应包含在序列化 kwargs 中的属性名称列表。

这些属性必须被构造函数接受。

默认为空字典。

cache `类属性` `实例属性` ¶

cache: BaseCache | bool | None = Field(default=None, exclude=True)

是否缓存响应。

如果为 True，将使用全局缓存。
如果为 False，将不使用缓存
如果为 None，如果设置了全局缓存，则使用全局缓存，否则不使用缓存。
如果是 BaseCache 的实例，将使用提供的缓存。

目前不支持模型的流式方法的缓存。

verbose `类属性` `实例属性` ¶

verbose: bool = Field(default_factory=_get_verbosity, exclude=True, repr=False)

是否打印响应文本。

callbacks `类属性` `实例属性` ¶

callbacks: Callbacks = Field(default=None, exclude=True)

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

tags `类属性` `实例属性` ¶

tags: list[str] | None = Field(default=None, exclude=True)

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

metadata `类属性` `实例属性` ¶

metadata: dict[str, Any] | None = Field(default=None, exclude=True)

添加到运行跟踪中的元数据。

custom_get_token_ids `类属性` `实例属性` ¶

custom_get_token_ids: Callable[[str], list[int]] | None = Field(
    default=None, exclude=True
)

用于计算 token 的可选编码器。

rate_limiter `类属性` `实例属性` ¶

rate_limiter: BaseRateLimiter | None = Field(default=None, exclude=True)

一个可选的速率限制器，用于限制请求数量。

disable_streaming `类属性` `实例属性` ¶

disable_streaming: bool | Literal['tool_calling'] = False

是否为此模型禁用流式传输。

如果绕过流式传输，则 stream/astream/astream_events 将转而调用 invoke/ainvoke。

如果为 True，将始终绕过流式传输情况。
如果为 'tool_calling'，仅当模型被调用时带有 tools 关键字参数时，才会绕过流式传输情况。换句话说，仅当提供 tools 参数时，LangChain 才会自动切换到非流式行为 (invoke)。这提供了两全其美的方案。
如果为 False（默认值），则在可用时始终使用流式传输情况。

此标志的主要原因是，代码可能是使用 stream 编写的，而用户可能希望将给定模型换成另一个实现不完全支持流式传输的模型。

output_version `类属性` `实例属性` ¶

output_version: str | None = Field(
    default_factory=from_env("LC_OUTPUT_VERSION", default=None)
)

要存储在消息内容中的 AIMessage 输出格式的版本。

AIMessage.content_blocks 将懒解析 content 的内容为标准格式。此标志可用于额外将标准格式存储在消息内容中，例如，用于序列化目的。

支持的值

'v0'：内容中特定于提供商的格式（可以使用 content_blocks 进行懒解析）
'v1'：内容中的标准化格式（与 content_blocks 一致）

合作伙伴包（例如 langchain-openai）也可以使用此字段以向后兼容的方式推出新的内容格式。

在版本 1.0 中添加

profile `属性` ¶

profile: ModelProfile

返回模型的性能分析信息。

此属性将依赖于 langchain-model-profiles 包来检索聊天模型的能力，例如上下文窗口大小和支持的功能。

引发	描述
`ImportError`	如果未安装 `langchain-model-profiles`。

返回	描述
`ModelProfile`	一个 `ModelProfile` 对象，包含模型的性能分析信息。

is_lc_serializable `类方法` ¶

is_lc_serializable() -> bool

该类是否可在 langchain 中序列化。

get_lc_namespace `类方法` ¶

get_lc_namespace() -> list[str]

获取 LangChain 对象的命名空间。

返回	描述
`list[str]`	`["langchain", "chat_models", "anthropic"]`

set_default_max_tokens `类方法` ¶

set_default_max_tokens(values: dict[str, Any]) -> Any

设置默认的 max_tokens。

build_extra `类方法` ¶

build_extra(values: dict) -> Any

构建模型关键字参数。

bind_tools ¶

bind_tools(
    tools: Sequence[dict[str, Any] | type | Callable | BaseTool],
    *,
    tool_choice: dict[str, str] | str | None = None,
    parallel_tool_calls: bool | None = None,
    **kwargs: Any,
) -> Runnable[LanguageModelInput, AIMessage]

将类工具（tool-like）对象绑定到此聊天模型。

参数	描述
`工具`	要绑定到此聊天模型的一系列工具定义。支持 Anthropic 格式的工具模式和任何由 `langchain_core.utils.function_calling.convert_to_openai_tool` 处理的工具定义。类型： `Sequence[dict[str, Any] \| type \| Callable \| BaseTool]`
`工具选择`	要求模型调用的工具。选项包括工具名称，以字符串形式或以字典形式 `{"type": "tool", "name": "<<tool_name>>"}`：调用相应的工具； `'auto'`、`{"type: "auto"}` 或 `None`：自动选择一个工具（包括不选择工具）； `'any'` 或 `{"type: "any"}`：强制至少调用一个工具；类型： `dict[str, str] \| str \| None` 默认值： `None`
`parallel_tool_calls`	设置为 `False` 以禁用并行工具使用。默认为 `None`（无指定，允许并行工具使用）。在版本 0.3.2 中添加。类型： `bool \| None` 默认值： `None`
`kwargs`	任何其他参数都直接传递给 `bind`。类型： `Any` 默认值： `{}`

示例

from langchain_anthropic import ChatAnthropic
from pydantic import BaseModel, Field


class GetWeather(BaseModel):
    '''Get the current weather in a given location'''

    location: str = Field(..., description="The city and state, e.g. San Francisco, CA")


class GetPrice(BaseModel):
    '''Get the price of a specific product.'''

    product: str = Field(..., description="The product to look up.")


model = ChatAnthropic(model="claude-sonnet-4-5-20250929", temperature=0)
model_with_tools = model.bind_tools([GetWeather, GetPrice])
model_with_tools.invoke(
    "What is the weather like in San Francisco",
)
# -> AIMessage(
#     content=[
#         {'text': '<thinking>\nBased on the user\'s question, the relevant function to call is GetWeather, which requires the "location" parameter.\n\nThe user has directly specified the location as "San Francisco". Since San Francisco is a well known city, I can reasonably infer they mean San Francisco, CA without needing the state specified.\n\nAll the required parameters are provided, so I can proceed with the API call.\n</thinking>', 'type': 'text'},
#         {'text': None, 'type': 'tool_use', 'id': 'toolu_01SCgExKzQ7eqSkMHfygvYuu', 'name': 'GetWeather', 'input': {'location': 'San Francisco, CA'}}
#     ],
#     response_metadata={'id': 'msg_01GM3zQtoFv8jGQMW7abLnhi', 'model': 'claude-sonnet-4-5-20250929', 'stop_reason': 'tool_use', 'stop_sequence': None, 'usage': {'input_tokens': 487, 'output_tokens': 145}},
#     id='run-87b1331e-9251-4a68-acef-f0a018b639cc-0'
# )

示例 — 使用 tool_choice 'any' 强制调用工具

```python
from langchain_anthropic import ChatAnthropic
from pydantic import BaseModel, Field


class GetWeather(BaseModel):
    '''Get the current weather in a given location'''

    location: str = Field(..., description="The city and state, e.g. San Francisco, CA")


class GetPrice(BaseModel):
    '''Get the price of a specific product.'''

    product: str = Field(..., description="The product to look up.")


model = ChatAnthropic(model="claude-sonnet-4-5-20250929", temperature=0)
model_with_tools = model.bind_tools([GetWeather, GetPrice], tool_choice="any")
model_with_tools.invoke(
    "what is the weather like in San Francisco",
)
```

示例 — 使用 tool_choice '<name_of_tool>' 强制调用特定工具

from langchain_anthropic import ChatAnthropic
from pydantic import BaseModel, Field


class GetWeather(BaseModel):
    '''Get the current weather in a given location'''

    location: str = Field(..., description="The city and state, e.g. San Francisco, CA")


class GetPrice(BaseModel):
    '''Get the price of a specific product.'''

    product: str = Field(..., description="The product to look up.")


model = ChatAnthropic(model="claude-sonnet-4-5-20250929", temperature=0)
model_with_tools = model.bind_tools([GetWeather, GetPrice], tool_choice="GetWeather")
model_with_tools.invoke("What is the weather like in San Francisco")

示例 — 缓存特定工具

from langchain_anthropic import ChatAnthropic, convert_to_anthropic_tool
from pydantic import BaseModel, Field


class GetWeather(BaseModel):
    '''Get the current weather in a given location'''

    location: str = Field(..., description="The city and state, e.g. San Francisco, CA")


class GetPrice(BaseModel):
    '''Get the price of a specific product.'''

    product: str = Field(..., description="The product to look up.")


# We'll convert our pydantic class to the anthropic tool format
# before passing to bind_tools so that we can set the 'cache_control'
# field on our tool.
cached_price_tool = convert_to_anthropic_tool(GetPrice)
# Currently the only supported "cache_control" value is
# {"type": "ephemeral"}.
cached_price_tool["cache_control"] = {"type": "ephemeral"}

# We need to pass in extra headers to enable use of the beta cache
# control API.
model = ChatAnthropic(
    model="claude-sonnet-4-5-20250929",
    temperature=0,
)
model_with_tools = model.bind_tools([GetWeather, cached_price_tool])
model_with_tools.invoke("What is the weather like in San Francisco")

这将输出

AIMessage(
    content=[
        {
            "text": "Certainly! I can help you find out the current weather in San Francisco. To get this information, I'll use the GetWeather function. Let me fetch that data for you right away.",
            "type": "text",
        },
        {
            "id": "toolu_01TS5h8LNo7p5imcG7yRiaUM",
            "input": {"location": "San Francisco, CA"},
            "name": "GetWeather",
            "type": "tool_use",
        },
    ],
    response_metadata={
        "id": "msg_01Xg7Wr5inFWgBxE5jH9rpRo",
        "model": "claude-sonnet-4-5-20250929",
        "stop_reason": "tool_use",
        "stop_sequence": None,
        "usage": {
            "input_tokens": 171,
            "output_tokens": 96,
            "cache_creation_input_tokens": 1470,
            "cache_read_input_tokens": 0,
        },
    },
    id="run-b36a5b54-5d69-470e-a1b0-b932d00b089e-0",
    tool_calls=[
        {
            "name": "GetWeather",
            "args": {"location": "San Francisco, CA"},
            "id": "toolu_01TS5h8LNo7p5imcG7yRiaUM",
            "type": "tool_call",
        }
    ],
    usage_metadata={
        "input_tokens": 171,
        "output_tokens": 96,
        "total_tokens": 267,
    },
)

如果我们再次调用该工具，我们可以看到 AIMessage.response_metadata 中的“usage”信息显示我们有一次缓存命中

AIMessage(
    content=[
        {
            "text": "To get the current weather in San Francisco, I can use the GetWeather function. Let me check that for you.",
            "type": "text",
        },
        {
            "id": "toolu_01HtVtY1qhMFdPprx42qU2eA",
            "input": {"location": "San Francisco, CA"},
            "name": "GetWeather",
            "type": "tool_use",
        },
    ],
    response_metadata={
        "id": "msg_016RfWHrRvW6DAGCdwB6Ac64",
        "model": "claude-sonnet-4-5-20250929",
        "stop_reason": "tool_use",
        "stop_sequence": None,
        "usage": {
            "input_tokens": 171,
            "output_tokens": 82,
            "cache_creation_input_tokens": 0,
            "cache_read_input_tokens": 1470,
        },
    },
    id="run-88b1f825-dcb7-4277-ac27-53df55d22001-0",
    tool_calls=[
        {
            "name": "GetWeather",
            "args": {"location": "San Francisco, CA"},
            "id": "toolu_01HtVtY1qhMFdPprx42qU2eA",
            "type": "tool_call",
        }
    ],
    usage_metadata={
        "input_tokens": 171,
        "output_tokens": 82,
        "total_tokens": 253,
    },
)

with_structured_output ¶

with_structured_output(
    schema: dict | type, *, include_raw: bool = False, **kwargs: Any
) -> Runnable[LanguageModelInput, dict | BaseModel]

返回与给定模式匹配的格式化输出的模型包装器。

参数	描述
`模式`	输出模式。可以作为以下形式传入：一个 Anthropic 工具模式，一个 OpenAI 函数/工具模式，一个 JSON Schema，一个 `TypedDict` 类，或一个 Pydantic 类。如果 `schema` 是一个 Pydantic 类，那么模型输出将是该类的 Pydantic 实例，并且模型生成的字段将由 Pydantic 类进行验证。否则，模型输出将是一个字典，并且不会被验证。有关在指定 Pydantic 或 `TypedDict` 类时如何正确指定模式字段的类型和描述的更多信息，请参阅 `langchain_core.utils.function_calling.convert_to_openai_tool`。类型： `dict \| type`
`包含原始数据`	如果为 `False`，则仅返回解析后的结构化输出。如果在模型输出解析期间发生错误，它将被引发。如果为 `True`，则将返回原始模型响应（一个 `BaseMessage`）和解析后的模型响应。如果在输出解析期间发生错误，它将被捕获并一并返回。最终输出总是一个带有键 `'raw'`、`'parsed'` 和 `'parsing_error'` 的 `dict`。类型： `bool` 默认值： `False`
`kwargs`	额外的关键字参数将被忽略。类型： `Any` 默认值： `{}`

返回描述

Runnable[LanguageModelInput, dict | BaseModel]

一个接受与 langchain_core.language_models.chat.BaseChatModel 相同输入的 Runnable。如果 include_raw 为 False 且 schema 是一个 Pydantic 类，则 Runnable 输出一个 schema 的实例（即一个 Pydantic 对象）。否则，如果 include_raw 为 False，则 Runnable 输出一个 dict。

如果 include_raw 为 True，则 Runnable 输出一个带有以下键的 dict

'raw': BaseMessage
'parsed': 如果出现解析错误则为 None，否则类型取决于上面描述的 schema。
'parsing_error': BaseException | None

示例：Pydantic 模式 (include_raw=False)

from langchain_anthropic import ChatAnthropic
from pydantic import BaseModel


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

    answer: str
    justification: str


model = ChatAnthropic(model="claude-sonnet-4-5-20250929", temperature=0)
structured_model = model.with_structured_output(AnswerWithJustification)

structured_model.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_anthropic import ChatAnthropic
from pydantic import BaseModel


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

    answer: str
    justification: str


model = ChatAnthropic(model="claude-sonnet-4-5-20250929", temperature=0)
structured_model = model.with_structured_output(AnswerWithJustification, include_raw=True)

structured_model.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
# }

示例：dict 模式 (include_raw=False)

from langchain_anthropic import ChatAnthropic

schema = {
    "name": "AnswerWithJustification",
    "description": "An answer to the user question along with justification for the answer.",
    "input_schema": {
        "type": "object",
        "properties": {
            "answer": {"type": "string"},
            "justification": {"type": "string"},
        },
        "required": ["answer", "justification"],
    },
}
model = ChatAnthropic(model="claude-sonnet-4-5-20250929", temperature=0)
structured_model = model.with_structured_output(schema)

structured_model.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.'
# }

get_num_tokens_from_messages ¶

get_num_tokens_from_messages(
    messages: list[BaseMessage],
    tools: Sequence[dict[str, Any] | type | Callable | BaseTool] | None = None,
    **kwargs: Any,
) -> int

计算输入消息序列中的令牌数。

参数	描述
`messages`	要进行分词的消息输入。类型： `list[BaseMessage]`
`工具`	如果提供，则为要转换为工具模式的 `dict`、`BaseModel`、函数或 `BaseTool` 对象的序列。类型： `Sequence[dict[str, Any] \| type \| Callable \| BaseTool] \| None` 默认值： `None`
`kwargs`	额外的关键字参数将传递给 Anthropic 的 `messages.count_tokens` 方法。类型： `Any` 默认值： `{}`

基本用法

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage

model = ChatAnthropic(model="claude-sonnet-4-5-20250929")

messages = [
    SystemMessage(content="You are a scientist"),
    HumanMessage(content="Hello, Claude"),
]
model.get_num_tokens_from_messages(messages)

传递工具模式

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool

model = ChatAnthropic(model="claude-sonnet-4-5-20250929")

@tool(parse_docstring=True)
def get_weather(location: str) -> str:
    """Get the current weather in a given location

    Args:
        location: The city and state, e.g. San Francisco, CA
    """
    return "Sunny"

messages = [
    HumanMessage(content="What's the weather like in San Francisco?"),
]
model.get_num_tokens_from_messages(messages, tools=[get_weather])

行为在 0.3.0 版本中发生变化

使用 Anthropic 的令牌计数 API来计算消息中的令牌数。

get_name ¶

get_name(suffix: str | None = None, *, name: str | None = None) -> str

获取 Runnable 的名称。

参数	描述
`后缀`	一个可选的后缀，附加到名称上。类型: `str \| None` 默认值: `None`
`name`	一个可选的名称，用于代替 `Runnable` 的名称。类型: `str \| None` 默认值: `None`

返回	描述
`str`	`Runnable` 的名称。

get_input_schema ¶

get_input_schema(config: RunnableConfig | None = None) -> type[BaseModel]

获取可用于验证 Runnable 输入的 Pydantic 模型。

利用 configurable_fields 和 configurable_alternatives 方法的 Runnable 对象将具有一个动态输入模式，该模式取决于调用 Runnable 时使用的配置。

此方法允许获取特定配置的输入模式。

参数	描述
`配置`	生成模式时使用的配置。类型： `RunnableConfig \| None` 默认值： `None`

返回	描述
`type[BaseModel]`	一个可用于验证输入的 Pydantic 模型。

get_input_jsonschema ¶

get_input_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]

获取表示 Runnable 输入的 JSON 模式。

参数	描述
`配置`	生成模式时使用的配置。类型： `RunnableConfig \| None` 默认值： `None`

返回	描述
`dict[str, Any]`	表示 `Runnable` 输入的 JSON 模式。

示例

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


runnable = RunnableLambda(add_one)

print(runnable.get_input_jsonschema())

在 0.3.0 版本中新增。

get_output_schema ¶

get_output_schema(config: RunnableConfig | None = None) -> type[BaseModel]

获取可用于验证 Runnable 输出的 Pydantic 模型。

利用 configurable_fields 和 configurable_alternatives 方法的 Runnable 对象将具有一个动态输出模式，该模式取决于调用 Runnable 时使用的配置。

此方法允许获取特定配置的输出模式。

参数	描述
`配置`	生成模式时使用的配置。类型： `RunnableConfig \| None` 默认值： `None`

返回	描述
`type[BaseModel]`	一个可用于验证输出的 Pydantic 模型。

get_output_jsonschema ¶

get_output_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]

获取表示 Runnable 输出的 JSON 模式。

参数	描述
`配置`	生成模式时使用的配置。类型： `RunnableConfig \| None` 默认值： `None`

返回	描述
`dict[str, Any]`	表示 `Runnable` 输出的 JSON 模式。

示例

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


runnable = RunnableLambda(add_one)

print(runnable.get_output_jsonschema())

在 0.3.0 版本中新增。

config_schema ¶

config_schema(*, include: Sequence[str] | None = None) -> type[BaseModel]

此 Runnable 接受的配置类型，指定为 Pydantic 模型。

要将字段标记为可配置，请参阅 configurable_fields 和 configurable_alternatives 方法。

参数	描述
`包含`	要包含在配置模式中的字段列表。类型: `Sequence[str] \| None` 默认值: `None`

返回	描述
`type[BaseModel]`	一个可用于验证配置的 Pydantic 模型。

get_config_jsonschema ¶

get_config_jsonschema(*, include: Sequence[str] | None = None) -> dict[str, Any]

获取表示 Runnable 配置的 JSON 模式。

参数	描述
`包含`	要包含在配置模式中的字段列表。类型: `Sequence[str] \| None` 默认值: `None`

返回	描述
`dict[str, Any]`	表示 `Runnable` 配置的 JSON 模式。

在 0.3.0 版本中新增。

get_graph ¶

get_graph(config: RunnableConfig | None = None) -> Graph

返回此 Runnable 的图形表示。

get_prompts ¶

get_prompts(config: RunnableConfig | None = None) -> list[BasePromptTemplate]

返回此 Runnable 使用的提示列表。

or ¶

__or__(
    other: Runnable[Any, Other]
    | Callable[[Iterator[Any]], Iterator[Other]]
    | Callable[[AsyncIterator[Any]], AsyncIterator[Other]]
    | Callable[[Any], Other]
    | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any],
) -> RunnableSerializable[Input, Other]

Runnable "or" 运算符。

将此 Runnable 与另一个对象组合以创建 RunnableSequence。

参数	描述
`other`	另一个 `Runnable` 或类 `Runnable` 对象。类型： `Runnable[Any, Other] \| Callable[[Iterator[Any]], Iterator[Other]] \| Callable[[AsyncIterator[Any]], AsyncIterator[Other]] \| Callable[[Any], Other] \| Mapping[str, Runnable[Any, Other] \| Callable[[Any], Other] \| Any]`

返回	描述
`RunnableSerializable[Input, Other]`	一个新的 `Runnable`。

ror ¶

__ror__(
    other: Runnable[Other, Any]
    | Callable[[Iterator[Other]], Iterator[Any]]
    | Callable[[AsyncIterator[Other]], AsyncIterator[Any]]
    | Callable[[Other], Any]
    | Mapping[str, Runnable[Other, Any] | Callable[[Other], Any] | Any],
) -> RunnableSerializable[Other, Output]

Runnable "reverse-or" 运算符。

将此 Runnable 与另一个对象组合以创建 RunnableSequence。

参数	描述
`other`	另一个 `Runnable` 或类 `Runnable` 对象。类型： `Runnable[Other, Any] \| Callable[[Iterator[Other]], Iterator[Any]] \| Callable[[AsyncIterator[Other]], AsyncIterator[Any]] \| Callable[[Other], Any] \| Mapping[str, Runnable[Other, Any] \| Callable[[Other], Any] \| Any]`

返回	描述
`RunnableSerializable[Other, Output]`	一个新的 `Runnable`。

pipe ¶

pipe(
    *others: Runnable[Any, Other] | Callable[[Any], Other], name: str | None = None
) -> RunnableSerializable[Input, Other]

管道连接 Runnable 对象。

将此 Runnable 与类 Runnable 对象组合以构成一个 RunnableSequence。

等同于 RunnableSequence(self, *others) 或 self | others[0] | ...

示例

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


def mul_two(x: int) -> int:
    return x * 2


runnable_1 = RunnableLambda(add_one)
runnable_2 = RunnableLambda(mul_two)
sequence = runnable_1.pipe(runnable_2)
# Or equivalently:
# sequence = runnable_1 | runnable_2
# sequence = RunnableSequence(first=runnable_1, last=runnable_2)
sequence.invoke(1)
await sequence.ainvoke(1)
# -> 4

sequence.batch([1, 2, 3])
await sequence.abatch([1, 2, 3])
# -> [4, 6, 8]

参数	描述
`*其他`	其他要组合的 `Runnable` 或类 `Runnable` 对象类型： `Runnable[Any, Other] \| Callable[[Any], Other]` 默认值： `()`
`name`	生成的 `RunnableSequence` 的一个可选名称。类型: `str \| None` 默认值: `None`

返回	描述
`RunnableSerializable[Input, Other]`	一个新的 `Runnable`。

pick ¶

pick(keys: str | list[str]) -> RunnableSerializable[Any, Any]

从此 Runnable 的输出 dict 中选择键。

选择单个键

import json

from langchain_core.runnables import RunnableLambda, RunnableMap

as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)
chain = RunnableMap(str=as_str, json=as_json)

chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3]}

json_only_chain = chain.pick("json")
json_only_chain.invoke("[1, 2, 3]")
# -> [1, 2, 3]

选择键列表

from typing import Any

import json

from langchain_core.runnables import RunnableLambda, RunnableMap

as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)


def as_bytes(x: Any) -> bytes:
    return bytes(x, "utf-8")


chain = RunnableMap(str=as_str, json=as_json, bytes=RunnableLambda(as_bytes))

chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3], "bytes": b"[1, 2, 3]"}

json_and_bytes_chain = chain.pick(["json", "bytes"])
json_and_bytes_chain.invoke("[1, 2, 3]")
# -> {"json": [1, 2, 3], "bytes": b"[1, 2, 3]"}

参数	描述
`keys`	从输出字典中选择的一个键或键列表。类型: `str \| list[str]`

返回	描述
`RunnableSerializable[Any, Any]`	一个新的 `Runnable`。

assign ¶

assign(
    **kwargs: Runnable[dict[str, Any], Any]
    | Callable[[dict[str, Any]], Any]
    | Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]],
) -> RunnableSerializable[Any, Any]

向此 Runnable 的 dict 输出分配新字段。

from langchain_community.llms.fake import FakeStreamingListLLM
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import SystemMessagePromptTemplate
from langchain_core.runnables import Runnable
from operator import itemgetter

prompt = (
    SystemMessagePromptTemplate.from_template("You are a nice assistant.")
    + "{question}"
)
model = FakeStreamingListLLM(responses=["foo-lish"])

chain: Runnable = prompt | model | {"str": StrOutputParser()}

chain_with_assign = chain.assign(hello=itemgetter("str") | model)

print(chain_with_assign.input_schema.model_json_schema())
# {'title': 'PromptInput', 'type': 'object', 'properties':
{'question': {'title': 'Question', 'type': 'string'}}}
print(chain_with_assign.output_schema.model_json_schema())
# {'title': 'RunnableSequenceOutput', 'type': 'object', 'properties':
{'str': {'title': 'Str',
'type': 'string'}, 'hello': {'title': 'Hello', 'type': 'string'}}}

参数	描述
`**kwargs`	一个键到 `Runnable` 或类 `Runnable` 对象的映射，这些对象将使用此 `Runnable` 的整个输出字典来调用。类型： `Runnable[dict[str, Any], Any] \| Callable[[dict[str, Any]], Any] \| Mapping[str, Runnable[dict[str, Any], Any] \| Callable[[dict[str, Any]], Any]]` 默认值： `{}`

返回	描述
`RunnableSerializable[Any, Any]`	一个新的 `Runnable`。

invoke ¶

invoke(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> AIMessage

将单个输入转换为输出。

参数	描述
`输入`	`Runnable` 的输入。类型： `Input`
`配置`	调用 `Runnable` 时使用的配置。该配置支持标准键，如用于追踪目的的 `'tags'`、`'metadata'`，用于控制并行工作量的 `'max_concurrency'`，以及其他键。请参阅 `RunnableConfig` 以获取更多详细信息。类型： `RunnableConfig \| None` 默认值： `None`

返回	描述
`输出`	`Runnable` 的输出。

ainvoke `异步` ¶

ainvoke(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> AIMessage

将单个输入转换为输出。

参数	描述
`输入`	`Runnable` 的输入。类型： `Input`
`配置`	调用 `Runnable` 时使用的配置。该配置支持标准键，如用于追踪目的的 `'tags'`、`'metadata'`，用于控制并行工作量的 `'max_concurrency'`，以及其他键。请参阅 `RunnableConfig` 以获取更多详细信息。类型： `RunnableConfig \| None` 默认值： `None`

返回	描述
`输出`	`Runnable` 的输出。

batch ¶

batch(
    inputs: list[Input],
    config: RunnableConfig | list[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> list[Output]

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

批处理的默认实现对于 IO 密集型的 runnable 效果很好。

如果子类能够更有效地进行批处理，则必须重写此方法；例如，如果底层的 Runnable 使用支持批处理模式的 API。

参数	描述
`inputs`	`Runnable` 的输入列表。类型: `list[Input]`
`配置`	调用 `Runnable` 时使用的配置。该配置支持标准键，如用于追踪目的的 `'tags'`、`'metadata'`，用于控制并行工作量的 `'max_concurrency'`，以及其他键。请参阅 `RunnableConfig` 以获取更多详细信息。类型： `RunnableConfig \| list[RunnableConfig] \| None` 默认值： `None`
`返回异常`	是否返回异常而不是引发它们。类型： `bool` 默认值： `False`
`**kwargs`	要传递给 `Runnable` 的其他关键字参数。类型: `Any \| None` 默认值: `{}`

返回	描述
`list[Output]`	来自 `Runnable` 的输出列表。

batch_as_completed ¶

batch_as_completed(
    inputs: Sequence[Input],
    config: RunnableConfig | Sequence[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> Iterator[tuple[int, Output | Exception]]

在输入列表上并行运行 invoke。

在结果完成时生成它们。

参数	描述
`inputs`	`Runnable` 的输入列表。类型: `Sequence[Input]`
`配置`	调用 `Runnable` 时使用的配置。该配置支持标准键，如用于追踪目的的 `'tags'`、`'metadata'`，用于控制并行工作量的 `'max_concurrency'`，以及其他键。请参阅 `RunnableConfig` 以获取更多详细信息。类型： `RunnableConfig \| Sequence[RunnableConfig] \| None` 默认值： `None`
`返回异常`	是否返回异常而不是引发它们。类型： `bool` 默认值： `False`
`**kwargs`	要传递给 `Runnable` 的其他关键字参数。类型: `Any \| None` 默认值: `{}`

YIELDS	描述
`tuple[int, Output \| Exception]`	由输入索引和 `Runnable` 输出组成的元组。

abatch `异步` ¶

abatch(
    inputs: list[Input],
    config: RunnableConfig | list[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> list[Output]

默认实现使用 asyncio.gather 并行运行 ainvoke。

batch 的默认实现对于 IO 密集型的 runnable 效果很好。

如果子类能够更有效地进行批处理，则必须重写此方法；例如，如果底层的 Runnable 使用支持批处理模式的 API。

参数	描述
`inputs`	`Runnable` 的输入列表。类型: `list[Input]`
`配置`	调用 `Runnable` 时使用的配置。该配置支持标准键，如用于追踪目的的 `'tags'`、`'metadata'`，用于控制并行工作量的 `'max_concurrency'`，以及其他键。请参阅 `RunnableConfig` 以获取更多详细信息。类型： `RunnableConfig \| list[RunnableConfig] \| None` 默认值： `None`
`返回异常`	是否返回异常而不是引发它们。类型： `bool` 默认值： `False`
`**kwargs`	要传递给 `Runnable` 的其他关键字参数。类型: `Any \| None` 默认值: `{}`

返回	描述
`list[Output]`	来自 `Runnable` 的输出列表。

abatch_as_completed `异步` ¶

abatch_as_completed(
    inputs: Sequence[Input],
    config: RunnableConfig | Sequence[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> AsyncIterator[tuple[int, Output | Exception]]

在输入列表上并行运行 ainvoke。

在结果完成时生成它们。

参数	描述
`inputs`	`Runnable` 的输入列表。类型: `Sequence[Input]`
`配置`	调用 `Runnable` 时使用的配置。该配置支持标准键，如用于追踪目的的 `'tags'`、`'metadata'`，用于控制并行工作量的 `'max_concurrency'`，以及其他键。请参阅 `RunnableConfig` 以获取更多详细信息。类型： `RunnableConfig \| Sequence[RunnableConfig] \| None` 默认值： `None`
`返回异常`	是否返回异常而不是引发它们。类型： `bool` 默认值： `False`
`**kwargs`	要传递给 `Runnable` 的其他关键字参数。类型: `Any \| None` 默认值: `{}`

YIELDS	描述
`AsyncIterator[tuple[int, Output \| Exception]]`	一个由输入索引和 `Runnable` 输出组成的元组。

stream ¶

stream(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> Iterator[AIMessageChunk]

stream 的默认实现，它调用 invoke。

如果子类支持流式输出，则必须重写此方法。

参数	描述
`输入`	`Runnable` 的输入。类型： `Input`
`配置`	用于 `Runnable` 的配置。类型： `RunnableConfig \| None` 默认值： `None`
`**kwargs`	要传递给 `Runnable` 的其他关键字参数。类型: `Any \| None` 默认值: `{}`

YIELDS	描述
`输出`	`Runnable` 的输出。

astream `异步` ¶

astream(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> AsyncIterator[AIMessageChunk]

astream 的默认实现，它调用 ainvoke。

如果子类支持流式输出，则必须重写此方法。

参数	描述
`输入`	`Runnable` 的输入。类型： `Input`
`配置`	用于 `Runnable` 的配置。类型： `RunnableConfig \| None` 默认值： `None`
`**kwargs`	要传递给 `Runnable` 的其他关键字参数。类型: `Any \| None` 默认值: `{}`

YIELDS	描述
`AsyncIterator[Output]`	`Runnable` 的输出。

astream_log `异步` ¶

astream_log(
    input: Any,
    config: RunnableConfig | None = None,
    *,
    diff: bool = True,
    with_streamed_output_list: bool = True,
    include_names: Sequence[str] | None = None,
    include_types: Sequence[str] | None = None,
    include_tags: Sequence[str] | None = None,
    exclude_names: Sequence[str] | None = None,
    exclude_types: Sequence[str] | None = None,
    exclude_tags: Sequence[str] | None = None,
    **kwargs: Any,
) -> AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]

流式传输 Runnable 的所有输出，如回调系统所报告。

这包括 LLM、检索器、工具等的所有内部运行。

输出以 Log 对象的形式流式传输，其中包括一个 Jsonpatch 操作列表，描述了运行状态在每一步中如何变化，以及运行的最终状态。

可以按顺序应用 Jsonpatch 操作来构造状态。

参数	描述
`输入`	`Runnable` 的输入。类型： `Any`
`配置`	用于 `Runnable` 的配置。类型： `RunnableConfig \| None` 默认值： `None`
`差异`	是生成每一步之间的差异还是当前状态。类型: `bool` 默认值: `True`
`带流式输出列表`	是否生成 `streamed_output` 列表。类型: `bool` 默认值: `True`
`包含名称`	仅包含具有这些名称的日志。类型: `Sequence[str] \| None` 默认值: `None`
`包含类型`	仅包含具有这些类型的日志。类型: `Sequence[str] \| None` 默认值: `None`
`包含标签`	仅包含具有这些标签的日志。类型: `Sequence[str] \| None` 默认值: `None`
`排除名称`	排除具有这些名称的日志。类型: `Sequence[str] \| None` 默认值: `None`
`排除类型`	排除具有这些类型的日志。类型: `Sequence[str] \| None` 默认值: `None`
`排除标签`	排除具有这些标签的日志。类型: `Sequence[str] \| None` 默认值: `None`
`**kwargs`	要传递给 `Runnable` 的其他关键字参数。类型： `Any` 默认值： `{}`

YIELDS	描述
`AsyncIterator[RunLogPatch] \| AsyncIterator[RunLog]`	一个 `RunLogPatch` 或 `RunLog` 对象。

astream_events `异步` ¶

astream_events(
    input: Any,
    config: RunnableConfig | None = None,
    *,
    version: Literal["v1", "v2"] = "v2",
    include_names: Sequence[str] | None = None,
    include_types: Sequence[str] | None = None,
    include_tags: Sequence[str] | None = None,
    exclude_names: Sequence[str] | None = None,
    exclude_types: Sequence[str] | None = None,
    exclude_tags: Sequence[str] | None = None,
    **kwargs: Any,
) -> AsyncIterator[StreamEvent]

生成事件流。

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

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

event：事件名称的格式为：on_[runnable_type]_(start|stream|end)。
name：生成事件的 Runnable 的名称。
run_id：与发出事件的 Runnable 的给定执行相关联的随机生成的 ID。作为父 Runnable 执行的一部分被调用的子 Runnable 被分配其自己的唯一 ID。
parent_ids：生成事件的父可运行对象的 ID。根 Runnable 将有一个空列表。父 ID 的顺序是从根到直接父级。仅适用于 API 的 v2 版本。API 的 v1 版本将返回一个空列表。
tags：生成事件的 Runnable 的标签。
metadata：生成事件的 Runnable 的元数据。
data：与事件关联的数据。此字段的内容取决于事件的类型。有关更多详细信息，请参见下表。

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

注意

此参考表适用于模式的 v2 版本。

事件	name	chunk	输入	output
`on_chat_model_start`	`'[model name]'`		`{"messages": [[SystemMessage, HumanMessage]]}`
`on_chat_model_stream`	`'[model name]'`	`AIMessageChunk(content="hello")`
`on_chat_model_end`	`'[model name]'`		`{"messages": [[SystemMessage, HumanMessage]]}`	`AIMessageChunk(content="hello world")`
`on_llm_start`	`'[model name]'`		`{'input': 'hello'}`
`on_llm_stream`	`'[model name]'`	`'你好'`
`on_llm_end`	`'[model name]'`		`'你好，人类！'`
`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`	`'[retriever name]'`		`{"query": "hello"}`
`on_retriever_end`	`'[retriever name]'`		`{"query": "hello"}`	`[Document(...), ..]`
`on_prompt_start`	`'[template_name]'`		`{"question": "hello"}`
`on_prompt_end`	`'[template_name]'`		`{"question": "hello"}`	`ChatPromptValue(messages: [SystemMessage, ...])`

除了标准事件外，用户还可以分派自定义事件（见下例）。

自定义事件将仅在 API 的 v2 版本中出现！

自定义事件具有以下格式

属性	类型	描述
`name`	`str`	用户为事件定义的名称。
`data`	`任意`	与事件关联的数据。这可以是任何东西，但我们建议使其可 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)

参数	描述
`输入`	`Runnable` 的输入。类型： `Any`
`配置`	用于 `Runnable` 的配置。类型： `RunnableConfig \| None` 默认值： `None`
`版本`	要使用的模式版本，可以是 `'v2'` 或 `'v1'`。用户应使用 `'v2'`。`'v1'` 是为了向后兼容，并将在 `0.4.0` 中弃用。在 API 稳定之前不会分配默认值。自定义事件将仅在 `'v2'` 中出现。类型: `Literal['v1', 'v2']` 默认值: `'v2'`
`包含名称`	仅包括来自具有匹配名称的 `Runnable` 对象的事件。类型: `Sequence[str] \| None` 默认值: `None`
`包含类型`	仅包括来自具有匹配类型的 `Runnable` 对象的事件。类型: `Sequence[str] \| None` 默认值: `None`
`包含标签`	仅包括来自具有匹配标签的 `Runnable` 对象的事件。类型: `Sequence[str] \| None` 默认值: `None`
`排除名称`	排除来自具有匹配名称的 `Runnable` 对象的事件。类型: `Sequence[str] \| None` 默认值: `None`
`排除类型`	排除来自具有匹配类型的 `Runnable` 对象的事件。类型: `Sequence[str] \| None` 默认值: `None`
`排除标签`	排除来自具有匹配标签的 `Runnable` 对象的事件。类型: `Sequence[str] \| None` 默认值: `None`
`**kwargs`	要传递给 `Runnable` 的其他关键字参数。这些将传递给 `astream_log`，因为 `astream_events` 的此实现是建立在 `astream_log` 之上的。类型： `Any` 默认值： `{}`

YIELDS	描述
`AsyncIterator[StreamEvent]`	`StreamEvent` 的异步流。

引发	描述
`NotImplementedError`	如果版本不是 `'v1'` 或 `'v2'`。

transform ¶

transform(
    input: Iterator[Input], config: RunnableConfig | None = None, **kwargs: Any | None
) -> Iterator[Output]

将输入转换为输出。

transform 的默认实现，它缓冲输入并调用 astream。

如果子类可以在输入仍在生成时开始产生输出，则必须重写此方法。

参数	描述
`输入`	`Runnable` 输入的迭代器。类型: `Iterator[Input]`
`配置`	用于 `Runnable` 的配置。类型： `RunnableConfig \| None` 默认值： `None`
`**kwargs`	要传递给 `Runnable` 的其他关键字参数。类型: `Any \| None` 默认值: `{}`

YIELDS	描述
`输出`	`Runnable` 的输出。

atransform `异步` ¶

atransform(
    input: AsyncIterator[Input],
    config: RunnableConfig | None = None,
    **kwargs: Any | None,
) -> AsyncIterator[Output]

将输入转换为输出。

atransform 的默认实现，它缓冲输入并调用 astream。

如果子类可以在输入仍在生成时开始产生输出，则必须重写此方法。

参数	描述
`输入`	`Runnable` 输入的异步迭代器。类型： `AsyncIterator[Input]`
`配置`	用于 `Runnable` 的配置。类型： `RunnableConfig \| None` 默认值： `None`
`**kwargs`	要传递给 `Runnable` 的其他关键字参数。类型: `Any \| None` 默认值: `{}`

YIELDS	描述
`AsyncIterator[Output]`	`Runnable` 的输出。

bind ¶

bind(**kwargs: Any) -> Runnable[Input, Output]

将参数绑定到 Runnable，返回一个新的 Runnable。

当链中的 Runnable 需要一个不在前一个 Runnable 的输出中或未包含在用户输入中的参数时非常有用。

参数	描述
`**kwargs`	要绑定到 `Runnable` 的参数。类型： `Any` 默认值： `{}`

返回	描述
`Runnable[Input, Output]`	一个绑定了参数的新 `Runnable`。

示例

from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser

model = ChatOllama(model="llama3.1")

# Without bind
chain = model | StrOutputParser()

chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two three four five.'

# With bind
chain = model.bind(stop=["three"]) | StrOutputParser()

chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two'

with_config ¶

with_config(
    config: RunnableConfig | None = None, **kwargs: Any
) -> Runnable[Input, Output]

将配置绑定到 Runnable，返回一个新的 Runnable。

参数	描述
`配置`	要绑定到 `Runnable` 的配置。类型： `RunnableConfig \| None` 默认值： `None`
`**kwargs`	要传递给 `Runnable` 的其他关键字参数。类型： `Any` 默认值： `{}`

返回	描述
`Runnable[Input, Output]`	一个绑定了配置的新 `Runnable`。

with_listeners ¶

with_listeners(
    *,
    on_start: Callable[[Run], None]
    | Callable[[Run, RunnableConfig], None]
    | None = None,
    on_end: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None = None,
    on_error: Callable[[Run], None]
    | Callable[[Run, RunnableConfig], None]
    | None = None,
) -> Runnable[Input, Output]

将生命周期侦听器绑定到 Runnable，返回一个新的 Runnable。

Run 对象包含有关运行的信息，包括其 id、type、input、output、error、start_time、end_time 以及添加到运行中的任何标签或元数据。

参数	描述
`开始时`	在 `Runnable` 开始运行之前调用，并传入 `Run` 对象。类型： `Callable[[Run], None] \| Callable[[Run, RunnableConfig], None] \| None` 默认值： `None`
`结束时`	在 `Runnable` 完成运行后调用，并传入 `Run` 对象。类型： `Callable[[Run], None] \| Callable[[Run, RunnableConfig], None] \| None` 默认值： `None`
`出错时`	如果 `Runnable` 抛出错误，则调用此函数，并传入 `Run` 对象。类型： `Callable[[Run], None] \| Callable[[Run, RunnableConfig], None] \| None` 默认值： `None`

返回	描述
`Runnable[Input, Output]`	一个绑定了监听器的新 `Runnable`。

示例

from langchain_core.runnables import RunnableLambda
from langchain_core.tracers.schemas import Run

import time


def test_runnable(time_to_sleep: int):
    time.sleep(time_to_sleep)


def fn_start(run_obj: Run):
    print("start_time:", run_obj.start_time)


def fn_end(run_obj: Run):
    print("end_time:", run_obj.end_time)


chain = RunnableLambda(test_runnable).with_listeners(
    on_start=fn_start, on_end=fn_end
)
chain.invoke(2)

with_alisteners ¶

with_alisteners(
    *,
    on_start: AsyncListener | None = None,
    on_end: AsyncListener | None = None,
    on_error: AsyncListener | None = None,
) -> Runnable[Input, Output]

将异步生命周期侦听器绑定到 Runnable。

返回一个新的 Runnable。

Run 对象包含有关运行的信息，包括其 id、type、input、output、error、start_time、end_time 以及添加到运行中的任何标签或元数据。

参数	描述
`开始时`	在 `Runnable` 开始运行之前异步调用，并传入 `Run` 对象。类型： `AsyncListener \| None` 默认值： `None`
`结束时`	在 `Runnable` 完成运行后异步调用，并传入 `Run` 对象。类型： `AsyncListener \| None` 默认值： `None`
`出错时`	如果 `Runnable` 抛出错误，则异步调用此函数，并传入 `Run` 对象。类型： `AsyncListener \| None` 默认值： `None`

返回	描述
`Runnable[Input, Output]`	一个绑定了监听器的新 `Runnable`。

示例

from langchain_core.runnables import RunnableLambda, Runnable
from datetime import datetime, timezone
import time
import asyncio

def format_t(timestamp: float) -> str:
    return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat()

async def test_runnable(time_to_sleep: int):
    print(f"Runnable[{time_to_sleep}s]: starts at {format_t(time.time())}")
    await asyncio.sleep(time_to_sleep)
    print(f"Runnable[{time_to_sleep}s]: ends at {format_t(time.time())}")

async def fn_start(run_obj: Runnable):
    print(f"on start callback starts at {format_t(time.time())}")
    await asyncio.sleep(3)
    print(f"on start callback ends at {format_t(time.time())}")

async def fn_end(run_obj: Runnable):
    print(f"on end callback starts at {format_t(time.time())}")
    await asyncio.sleep(2)
    print(f"on end callback ends at {format_t(time.time())}")

runnable = RunnableLambda(test_runnable).with_alisteners(
    on_start=fn_start,
    on_end=fn_end
)
async def concurrent_runs():
    await asyncio.gather(runnable.ainvoke(2), runnable.ainvoke(3))

asyncio.run(concurrent_runs())
Result:
on start callback starts at 2025-03-01T07:05:22.875378+00:00
on start callback starts at 2025-03-01T07:05:22.875495+00:00
on start callback ends at 2025-03-01T07:05:25.878862+00:00
on start callback ends at 2025-03-01T07:05:25.878947+00:00
Runnable[2s]: starts at 2025-03-01T07:05:25.879392+00:00
Runnable[3s]: starts at 2025-03-01T07:05:25.879804+00:00
Runnable[2s]: ends at 2025-03-01T07:05:27.881998+00:00
on end callback starts at 2025-03-01T07:05:27.882360+00:00
Runnable[3s]: ends at 2025-03-01T07:05:28.881737+00:00
on end callback starts at 2025-03-01T07:05:28.882428+00:00
on end callback ends at 2025-03-01T07:05:29.883893+00:00
on end callback ends at 2025-03-01T07:05:30.884831+00:00

with_types ¶

with_types(
    *, input_type: type[Input] | None = None, output_type: type[Output] | None = None
) -> Runnable[Input, Output]

将输入和输出类型绑定到 Runnable，返回一个新的 Runnable。

参数	描述
`输入类型`	要绑定到 `Runnable` 的输入类型。类型： `type[Input] \| None` 默认值： `None`
`输出类型`	要绑定到 `Runnable` 的输出类型。类型： `type[Output] \| None` 默认值： `None`

返回	描述
`Runnable[Input, Output]`	一个绑定了类型的新 `Runnable`。

with_retry ¶

with_retry(
    *,
    retry_if_exception_type: tuple[type[BaseException], ...] = (Exception,),
    wait_exponential_jitter: bool = True,
    exponential_jitter_params: ExponentialJitterParams | None = None,
    stop_after_attempt: int = 3,
) -> Runnable[Input, Output]

创建一个新的 Runnable，它在发生异常时重试原始的 Runnable。

参数	描述
`如果异常类型则重试`	一个用于重试的异常类型元组。类型： `tuple[type[BaseException], ...]` 默认值： `(Exception,)`
`指数等待抖动`	是否在两次重试之间的等待时间中添加抖动。类型: `bool` 默认值: `True`
`尝试后停止`	放弃前尝试的最大次数。类型： `int` 默认值： `3`
`指数抖动参数`	`tenacity.wait_exponential_jitter` 的参数。即：`initial`、`max`、`exp_base` 和 `jitter`（均为 `float` 值）。类型： `ExponentialJitterParams \| None` 默认值： `None`

返回	描述
`Runnable[Input, Output]`	一个新的 Runnable，它会在发生异常时重试原始的 Runnable。

示例

from langchain_core.runnables import RunnableLambda

count = 0


def _lambda(x: int) -> None:
    global count
    count = count + 1
    if x == 1:
        raise ValueError("x is 1")
    else:
        pass


runnable = RunnableLambda(_lambda)
try:
    runnable.with_retry(
        stop_after_attempt=2,
        retry_if_exception_type=(ValueError,),
    ).invoke(1)
except ValueError:
    pass

assert count == 2

map ¶

map() -> Runnable[list[Input], list[Output]]

返回一个新的 Runnable，它将输入列表映射到输出列表。

使用每个输入调用 invoke。

返回	描述
`Runnable[list[Input], list[Output]]`	一个新的 `Runnable`，它将输入列表映射到输出列表。

示例

from langchain_core.runnables import RunnableLambda


def _lambda(x: int) -> int:
    return x + 1


runnable = RunnableLambda(_lambda)
print(runnable.map().invoke([1, 2, 3]))  # [2, 3, 4]

with_fallbacks ¶

with_fallbacks(
    fallbacks: Sequence[Runnable[Input, Output]],
    *,
    exceptions_to_handle: tuple[type[BaseException], ...] = (Exception,),
    exception_key: str | None = None,
) -> RunnableWithFallbacks[Input, Output]

向 Runnable 添加回退机制，返回一个新的 Runnable。

新的 Runnable 将在失败时先尝试原始的 Runnable，然后按顺序尝试每个备选方案。

参数	描述
`备用方案`	如果原始 `Runnable` 失败，将尝试的一系列 runnable。类型： `Sequence[Runnable[Input, Output]]`
`要处理的异常`	一个要处理的异常类型元组。类型： `tuple[type[BaseException], ...]` 默认值： `(Exception,)`
`异常键`	如果指定了 `string`，则已处理的异常将作为输入的一部分，在指定的键下传递给备选方案。如果为 `None`，异常将不会传递给备选方案。如果使用，基础 `Runnable` 及其备选方案必须接受一个字典作为输入。类型: `str \| None` 默认值: `None`

返回	描述
`RunnableWithFallbacks[Input, Output]`	一个新的 `Runnable`，它会在失败时先尝试原始的 `Runnable`，然后按顺序尝试每个备选方案。

示例

from typing import Iterator

from langchain_core.runnables import RunnableGenerator


def _generate_immediate_error(input: Iterator) -> Iterator[str]:
    raise ValueError()
    yield ""


def _generate(input: Iterator) -> Iterator[str]:
    yield from "foo bar"


runnable = RunnableGenerator(_generate_immediate_error).with_fallbacks(
    [RunnableGenerator(_generate)]
)
print("".join(runnable.stream({})))  # foo bar

参数	描述
`备用方案`	如果原始 `Runnable` 失败，将尝试的一系列 runnable。类型： `Sequence[Runnable[Input, Output]]`
`要处理的异常`	一个要处理的异常类型元组。类型： `tuple[type[BaseException], ...]` 默认值： `(Exception,)`
`异常键`	如果指定了 `string`，则已处理的异常将作为输入的一部分，在指定的键下传递给备选方案。如果为 `None`，异常将不会传递给备选方案。如果使用，基础 `Runnable` 及其备选方案必须接受一个字典作为输入。类型: `str \| None` 默认值: `None`

返回	描述
`RunnableWithFallbacks[Input, Output]`	一个新的 `Runnable`，它会在失败时先尝试原始的 `Runnable`，然后按顺序尝试每个备选方案。

as_tool ¶

as_tool(
    args_schema: type[BaseModel] | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    arg_types: dict[str, type] | None = None,
) -> BaseTool

从 Runnable 创建一个 BaseTool。

as_tool 将从一个 Runnable 实例化一个 BaseTool，该工具具有名称、描述和 args_schema。在可能的情况下，模式会从 runnable.get_input_schema 中推断。或者（例如，如果 Runnable 接受一个字典作为输入，并且特定的字典键没有类型），模式可以通过 args_schema 直接指定。你也可以传递 arg_types 来仅指定必需的参数及其类型。

参数	描述
`参数模式`	工具的模式。类型： `type[BaseModel] \| None` 默认值： `None`
`name`	工具的名称。类型: `str \| None` 默认值: `None`
`描述`	工具的描述。类型: `str \| None` 默认值: `None`
`参数类型`	一个从参数名称到类型的字典。类型： `dict[str, type] \| None` 默认值： `None`

返回	描述
`BaseTool`	一个 `BaseTool` 实例。

类型化字典输入

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
from pydantic 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
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")

init ¶

__init__(*args: Any, **kwargs: Any) -> None

lc_id `类方法` ¶

lc_id() -> list[str]

为此类返回一个用于序列化目的的唯一标识符。

唯一标识符是一个描述对象路径的字符串列表。

例如，对于类 langchain.llms.openai.OpenAI，id 是 ["langchain", "llms", "openai", "OpenAI"]。

to_json ¶

to_json() -> SerializedConstructor | SerializedNotImplemented

将 Runnable 序列化为 JSON。

返回	描述
`SerializedConstructor \| SerializedNotImplemented`	一个 `Runnable` 的 JSON 可序列化表示。

to_json_not_implemented ¶

to_json_not_implemented() -> SerializedNotImplemented

序列化一个“未实现”的对象。

返回	描述
`SerializedNotImplemented`	`SerializedNotImplemented`.

configurable_fields ¶

configurable_fields(
    **kwargs: AnyConfigurableField,
) -> RunnableSerializable[Input, Output]

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

参数	描述
`**kwargs`	一个要配置的 `ConfigurableField` 实例的字典。类型： `AnyConfigurableField` 默认值： `{}`

引发	描述
`ValueError`	如果在 `Runnable` 中找不到配置键。

返回	描述
`RunnableSerializable[Input, Output]`	一个配置了字段的新 `Runnable`。

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,
)

configurable_alternatives ¶

configurable_alternatives(
    which: ConfigurableField,
    *,
    default_key: str = "default",
    prefix_keys: bool = False,
    **kwargs: Runnable[Input, Output] | Callable[[], Runnable[Input, Output]],
) -> RunnableSerializable[Input, Output]

为可在运行时设置的 Runnable 对象配置备选项。

参数	描述
`哪个`	将用于选择备选项的 `ConfigurableField` 实例。类型： `ConfigurableField`
`默认键`	如果未选择备选项，则使用的默认键。类型： `str` 默认值： `'default'`
`前缀键`	是否用 `ConfigurableField` id 作为键的前缀。类型： `bool` 默认值： `False`
`**kwargs`	一个从键到 `Runnable` 实例或返回 `Runnable` 实例的可调用对象的字典。类型： `Runnable[Input, Output] \| Callable[[], Runnable[Input, Output]]` 默认值： `{}`

返回	描述
`RunnableSerializable[Input, Output]`	一个配置了备选项的新 `Runnable`。

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

model = ChatAnthropic(
    model_name="claude-sonnet-4-5-20250929"
).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
)

set_verbose ¶

set_verbose(verbose: bool | None) -> bool

如果 verbose 是 None，则设置它。

这允许用户传入 None 作为 verbose 来访问全局设置。

参数	描述
`verbose`	要使用的详细程度设置。类型： `bool \| None`

返回	描述
`bool`	要使用的详细程度设置。

generate_prompt ¶

generate_prompt(
    prompts: list[PromptValue],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    **kwargs: Any,
) -> LLMResult

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

对于提供批量 API 的模型，此方法应利用批量调用。

当你想要

利用批量调用，
需要从模型获得比最高生成值更多的输出，
正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
`prompts`	`PromptValue` 对象列表。`PromptValue` 是一个可以转换为匹配任何语言模型格式的对象（对于纯文本生成模型是字符串，对于聊天模型是 `BaseMessage` 对象）。类型： `list[PromptValue]`
`停止`	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。类型: `list[str] \| None` 默认值: `None`
`回调`	要传递的 `Callbacks`。用于在生成过程中执行额外的功能，如日志记录或流式处理。类型： `Callbacks` 默认值： `None`
`**kwargs`	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。类型： `Any` 默认值： `{}`

返回	描述
`LLMResult`	一个 `LLMResult`，其中包含每个输入提示的候选 `Generation` 对象列表以及额外的模型提供商特定的输出。

agenerate_prompt `异步` ¶

agenerate_prompt(
    prompts: list[PromptValue],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    **kwargs: Any,
) -> LLMResult

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

对于提供批量 API 的模型，此方法应利用批量调用。

当你想要

利用批量调用，
需要从模型获得比最高生成值更多的输出，
正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
`prompts`	`PromptValue` 对象列表。`PromptValue` 是一个可以转换为匹配任何语言模型格式的对象（对于纯文本生成模型是字符串，对于聊天模型是 `BaseMessage` 对象）。类型： `list[PromptValue]`
`停止`	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。类型: `list[str] \| None` 默认值: `None`
`回调`	要传递的 `Callbacks`。用于在生成过程中执行额外的功能，如日志记录或流式处理。类型： `Callbacks` 默认值： `None`
`**kwargs`	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。类型： `Any` 默认值： `{}`

返回	描述
`LLMResult`	一个 `LLMResult`，其中包含每个输入提示的候选 `Generation` 对象列表以及额外的模型提供商特定的输出。

get_token_ids ¶

get_token_ids(text: str) -> list[int]

返回文本中 token 的有序 ID。

参数	描述
`text`	要进行分词的字符串输入。类型： `str`

返回	描述
`list[int]`	与文本中的词元相对应的 ID 列表，按它们在文本中出现的顺序列出。

get_num_tokens ¶

get_num_tokens(text: str) -> int

获取文本中存在的 token 数量。

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

参数	描述
`text`	要进行分词的字符串输入。类型： `str`

返回	描述
`int`	文本中的词元整数数量。

generate ¶

generate(
    messages: list[list[BaseMessage]],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    *,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    run_name: str | None = None,
    run_id: UUID | None = None,
    **kwargs: Any,
) -> LLMResult

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

对于提供批量 API 的模型，此方法应利用批量调用。

当你想要

利用批量调用，
需要从模型获得比最高生成值更多的输出，
正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
`messages`	消息列表的列表。类型： `list[list[BaseMessage]]`
`停止`	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。类型: `list[str] \| None` 默认值: `None`
`回调`	要传递的 `Callbacks`。用于在生成过程中执行额外的功能，如日志记录或流式处理。类型： `Callbacks` 默认值： `None`
`tags`	要应用的标签。类型: `list[str] \| None` 默认值: `None`
`metadata`	要应用的元数据。类型： `dict[str, Any] \| None` 默认值： `None`
`运行名称`	运行的名称。类型: `str \| None` 默认值: `None`
`run_id`	运行的 ID。类型: `UUID \| None` 默认值: `None`
`**kwargs`	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。类型： `Any` 默认值： `{}`

返回	描述
`LLMResult`	一个 `LLMResult`，其中包含每个输入提示的候选 `Generations` 列表以及额外的模型提供商特定的输出。

agenerate `异步` ¶

agenerate(
    messages: list[list[BaseMessage]],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    *,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    run_name: str | None = None,
    run_id: UUID | None = None,
    **kwargs: Any,
) -> LLMResult

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

对于提供批量 API 的模型，此方法应利用批量调用。

当你想要

利用批量调用，
需要从模型获得比最高生成值更多的输出，
正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
`messages`	消息列表的列表。类型： `list[list[BaseMessage]]`
`停止`	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。类型: `list[str] \| None` 默认值: `None`
`回调`	要传递的 `Callbacks`。用于在生成过程中执行额外的功能，如日志记录或流式处理。类型： `Callbacks` 默认值： `None`
`tags`	要应用的标签。类型: `list[str] \| None` 默认值: `None`
`metadata`	要应用的元数据。类型： `dict[str, Any] \| None` 默认值： `None`
`运行名称`	运行的名称。类型: `str \| None` 默认值: `None`
`run_id`	运行的 ID。类型: `UUID \| None` 默认值: `None`
`**kwargs`	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。类型： `Any` 默认值： `{}`

返回	描述
`LLMResult`	一个 `LLMResult`，其中包含每个输入提示的候选 `Generations` 列表以及额外的模型提供商特定的输出。

dict ¶

dict(**kwargs: Any) -> dict

返回 LLM 的字典。

ChatAnthropic¶

langchain_anthropic.chat_models.ChatAnthropic ¶

model 类属性 实例属性 ¶

max_tokens 类属性 实例属性 ¶

temperature 类属性 实例属性 ¶

top_k 类属性 实例属性 ¶

top_p 类属性 实例属性 ¶

default_request_timeout 类属性 实例属性 ¶

max_retries 类属性 实例属性 ¶

stop_sequences 类属性 实例属性 ¶

anthropic_api_url 类属性 实例属性 ¶

anthropic_api_key 类属性 实例属性 ¶

anthropic_proxy 类属性 实例属性 ¶

default_headers 类属性 实例属性 ¶

betas 类属性 实例属性 ¶

streaming 类属性 实例属性 ¶

stream_usage 类属性 实例属性 ¶

thinking 类属性 实例属性 ¶

mcp_servers 类属性 实例属性 ¶

context_management 类属性 实例属性 ¶

lc_secrets 属性 ¶

name 类属性 实例属性 ¶

InputType 属性 ¶

OutputType 属性 ¶

input_schema 属性 ¶

output_schema 属性 ¶

config_specs 属性 ¶

lc_attributes 属性 ¶

cache 类属性 实例属性 ¶

verbose 类属性 实例属性 ¶

callbacks 类属性 实例属性 ¶

tags 类属性 实例属性 ¶

metadata 类属性 实例属性 ¶

custom_get_token_ids 类属性 实例属性 ¶

rate_limiter 类属性 实例属性 ¶

disable_streaming 类属性 实例属性 ¶

output_version 类属性 实例属性 ¶

profile 属性 ¶

is_lc_serializable 类方法 ¶

get_lc_namespace 类方法 ¶

set_default_max_tokens 类方法 ¶

build_extra 类方法 ¶

bind_tools ¶

with_structured_output ¶

get_num_tokens_from_messages ¶

get_name ¶

get_input_schema ¶

get_input_jsonschema ¶

get_output_schema ¶

get_output_jsonschema ¶

config_schema ¶

get_config_jsonschema ¶

get_graph ¶

get_prompts ¶

__or__ ¶

__ror__ ¶

pipe ¶

pick ¶

assign ¶

invoke ¶

ainvoke 异步 ¶

batch ¶

batch_as_completed ¶

abatch 异步 ¶

abatch_as_completed 异步 ¶

stream ¶

astream 异步 ¶

astream_log 异步 ¶

astream_events 异步 ¶

transform ¶

atransform 异步 ¶

bind ¶

with_config ¶

with_listeners ¶

with_alisteners ¶

with_types ¶

with_retry ¶

map ¶

with_fallbacks ¶

as_tool ¶

`ChatAnthropic`¶

model `类属性` `实例属性` ¶

max_tokens `类属性` `实例属性` ¶

temperature `类属性` `实例属性` ¶

top_k `类属性` `实例属性` ¶

top_p `类属性` `实例属性` ¶

default_request_timeout `类属性` `实例属性` ¶

max_retries `类属性` `实例属性` ¶

stop_sequences `类属性` `实例属性` ¶

anthropic_api_url `类属性` `实例属性` ¶

anthropic_api_key `类属性` `实例属性` ¶

anthropic_proxy `类属性` `实例属性` ¶

default_headers `类属性` `实例属性` ¶

betas `类属性` `实例属性` ¶

streaming `类属性` `实例属性` ¶

stream_usage `类属性` `实例属性` ¶

thinking `类属性` `实例属性` ¶

mcp_servers `类属性` `实例属性` ¶

context_management `类属性` `实例属性` ¶

lc_secrets `属性` ¶

name `类属性` `实例属性` ¶

InputType `属性` ¶

OutputType `属性` ¶

input_schema `属性` ¶

output_schema `属性` ¶

config_specs `属性` ¶

lc_attributes `属性` ¶

cache `类属性` `实例属性` ¶

verbose `类属性` `实例属性` ¶

callbacks `类属性` `实例属性` ¶

tags `类属性` `实例属性` ¶

metadata `类属性` `实例属性` ¶

custom_get_token_ids `类属性` `实例属性` ¶

rate_limiter `类属性` `实例属性` ¶

disable_streaming `类属性` `实例属性` ¶

output_version `类属性` `实例属性` ¶

profile `属性` ¶

is_lc_serializable `类方法` ¶

get_lc_namespace `类方法` ¶

set_default_max_tokens `类方法` ¶

build_extra `类方法` ¶

or ¶

ror ¶

ainvoke `异步` ¶

abatch `异步` ¶

abatch_as_completed `异步` ¶

astream `异步` ¶

astream_log `异步` ¶

astream_events `异步` ¶

atransform `异步` ¶

init ¶

lc_id `类方法` ¶

agenerate_prompt `异步` ¶

agenerate `异步` ¶