解锁 Langchain：构建强大工作流的 5 个关键 Runnables
Runnables：LangChain 的构建基石
Runnable 是 LangChain 中的一个基础接口。
它是一个协议，用于标准化组件（例如 LLM、输出解析器、检索器或提示模板）的定义和执行方式。该接口确保了与各种组件的一致交互，从而促进了复杂工作流程和链的创建。
上一章：
[## LangChain in Chains #49: 批量处理
LangChain 中的批量处理
blog.devgenius.io](https://blog.devgenius.io/langchain-in-chains-49-batch-processing-a36a3cb27d63)
Runnable 本质上是一个抽象，代表一个工作单元。一个 Runnable 必须实现以下方法：
invoke：处理单个输入以产生输出。batch：并行处理多个输入。stream：在生成时流式传输输出。ainvoke,abatch,astream：这些是相应 的异步方法。astream_log：流式传输回中间步骤。astream_events：流式传输事件。
我们使用 Runnables 创建组合。我们使用 LCEL 将它们组合起来形成复杂 的管道。
每个 Runnable 都会公开输入和输出模式，从而清晰地了解预期的数据结构，并促进验证和调试。
让我们通过一个简单的例子并分析模式。
import os
os.environ["OPENAI_API_KEY"] = ""
from langchain_openai import ChatOpenAI
from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import PromptTemplate
from pydantic import BaseModel, Field
首先，我们有一个 PromptTemplate。
prompt_string = """
Write a movie review in any language for the movie {movie}
Format instructions:
{format_instructions}
"""
prompt_template = PromptTemplate(
input_variables=["movie", "format_instructions"],
template=prompt_string
)
通过调用 prompt_template.input_schema.model_json_schema()，我们获得预期输入的 JSON 模式表示：
prompt_template.input_schema.model_json_schema()
"""
{'properties': {'format_instructions': {'title': 'Format Instructions',
'type': 'string'},
'movie': {'title': 'Movie', 'type': 'string'}},
'required': ['format_instructions', 'movie'],
'title': 'PromptInput',
'type': 'object'}
"""
此模式表明 format_instructions 和 movie 都是必需的字符串属性。
通过调用 prompt_template.output_schema.model_json_schema()，我们检索预期输出的 JSON 模式表示。
'type': {'const': 'function',
'default': 'function',
'title': 'Type',
'type': 'string'},
'name': {'title': 'Name', 'type': 'string'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'example': {'default': False, 'title': 'Example', 'type': 'boolean'},
'tool_calls': {'default': [],
'items': {'$ref': '#/$defs/ToolCall'},
'title': 'Tool Calls',
'type': 'array'},
'invalid_tool_calls': {'default': [],
'items': {'$ref': '#/$defs/InvalidToolCall'},
'title': 'Invalid Tool Calls',
'type': 'array'},
'usage_metadata': {'anyOf': [{'$ref': '#/$defs/UsageMetadata'},
{'type': 'null'}],
'default': None}},
'required': ['content', 'name'],
'title': 'FunctionMessage',
'type': 'object'},
'FunctionMessageChunk': {'additionalProperties': True,
'description': 'Message chunk for passing the result of executing a tool back to a model.',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'FunctionMessageChunk',
'default': 'FunctionMessageChunk',
'title': 'Type',
'type': 'string'},
'name': {'title': 'Name', 'type': 'string'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'example': {'default': False, 'title': 'Example', 'type': 'boolean'},
'tool_calls': {'default': [],
'items': {'$ref': '#/$defs/ToolCall'},
'title': 'Tool Calls',
'type': 'array'},
'invalid_tool_calls': {'default': [],
'items': {'$ref': '#/$defs/InvalidToolCall'},
'title': 'Invalid Tool Calls',
'type': 'array'},
'usage_metadata': {'anyOf': [{'$ref': '#/$defs/UsageMetadata'},
{'type': 'null'}],
'default': None},
'tool_call_chunks': {'default': [],
'items': {'$ref': '#/$defs/ToolCallChunk'},
'title': 'Tool Call Chunks',
'type': 'array'}},
'required': ['content', 'name'],
'title': 'FunctionMessageChunk',
'type': 'object'},
'HumanMessage': {'additionalProperties': True,
'description': 'Message from a human.',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'human',
'default': 'human',
'title': 'Type',
'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'example': {'default': False, 'title': 'Example', 'type': 'boolean'},
'tool_calls': {'default': [],
'items': {'$ref': '#/$defs/ToolCall'},
'title': 'Tool Calls',
'type': 'array'},
'invalid_tool_calls': {'default': [],
'items': {'$ref': '#/$defs/InvalidToolCall'},
'title': 'Invalid Tool Calls',
'type': 'array'},
'usage_metadata': {'anyOf': [{'$ref': '#/$defs/UsageMetadata'},
{'type': 'null'}],
'default': None}},
'required': ['content'],
'title': 'HumanMessage',
'type': 'object'},
'HumanMessageChunk': {'additionalProperties': True,
'description': 'Message chunk from a human.',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'HumanMessageChunk',
'default': 'HumanMessageChunk',
'title': 'Type',
'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'example': {'default': False, 'title': 'Example', 'type': 'boolean'},
'tool_calls': {'default': [],
'items': {'$ref': '#/$defs/ToolCall'},
'title': 'Tool Calls',
'type': 'array'},
'invalid_tool_calls': {'default': [],
'items': {'$ref': '#/$defs/InvalidToolCall'},
'title': 'Invalid Tool Calls',
'type': 'array'},
'usage_metadata': {'anyOf': [{'$ref': '#/$defs/UsageMetadata'},
{'type': 'null'}],
'default': None},
'tool_call_chunks': {'default': [],
'items': {'$ref': '#/$defs/ToolCallChunk'},
'title': 'Tool Call Chunks',
'type': 'array'}},
'required': ['content'],
'title': 'HumanMessageChunk',
'type': 'object'},
'InvalidToolCall': {'additionalProperties': True,
'description': 'Represents a tool call that failed.',
'properties': {'id': {'title': 'Id', 'type': 'string'},
'type': {'const': 'invalid_tool_call',
'default': 'invalid_tool_call',
'title': 'Type',
'type': 'string'},
'code': {'title': 'Code', 'type': 'string'},
'message': {'title': 'Message', 'type': 'string'},
'tool_call_id': {'title': 'Tool Call Id', 'type': 'string'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'}},
'required': ['id', 'code', 'message', 'tool_call_id'],
'title': 'InvalidToolCall',
'type': 'object'},
'SystemMessage': {'additionalProperties': True,
'description': 'Message from the system.',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'system',
'default': 'system',
'title': 'Type',
'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': '
```markdown
### langchain_core.messages
```python
from typing import Any, Dict, List, Optional, Union
from langchain_core.pydantic_v1 import BaseModel, Field, root_validator
from langchain_core.utils import (
convert_to_json_compatible,
get_type_display_name,
)
BaseMessage
class BaseMessage(BaseModel):
"""Base class for all messages."""
content: Union[str, List[Union[str, Dict]]] = Field(
..., description="The contents of the message."
)
"""Contents of the message."""
additional_kwargs: Dict = Field(
default_factory=dict,
description="Additional information about the message.",
)
"""Additional information about the message."""
response_metadata: Dict = Field(
default_factory=dict,
description="Metadata about the response.",
)
"""Metadata about the response."""
type: str
"""Type of the message."""
name: Optional[str] = Field(default=None, description="Optional name for the message.")
"""Optional name for the message."""
id: Optional[str] = Field(default=None, description="Optional ID for the message.")
"""Optional ID for the message."""
@root_validator(pre=True)
def _validate_content(cls, values: Dict) -> Dict:
content = values.get("content")
if content is not None:
values["content"] = convert_to_json_compatible(content)
return values
def dict(self, *args: Any, **kwargs: Any) -> Dict:
"""Return dictionary representation of message."""
kwargs["exclude"] = kwargs.get("exclude", set()).union({"response_metadata"})
return super().dict(*args, **kwargs)
def __repr__(self) -> str:
"""Return string representation of message."""
return (
f"{self.__class__.__name__}(content='{self.content[:20]}'"
f"{'...' if len(self.content) > 20 else ''}"
f"{f', name={self.name}' if self.name else ''}"
f")"
)
def __str__(self) -> str:
"""Return string representation of message."""
return self.content
def to_json(self) -> str:
"""Return JSON representation of message."""
return self.json()
@property
def lc_namespace(self) -> List[str]:
"""Return the namespace of the module."""
return ["langchain_core", "messages"]
@property
def lc_serializable(self) -> bool:
"""Return whether this class is serializable."""
return True
BaseMessageChunk
class BaseMessageChunk(BaseMessage):
"""Base class for all message chunks."""
@property
def lc_serializable(self) -> bool:
"""Return whether this class is serializable."""
return False
AIMessage
class AIMessage(BaseMessage):
"""Message from an AI."""
type: str = "ai"
"""Type of the message."""
name: Optional[str] = Field(default=None, description="Optional name for the message.")
"""Optional name for the message."""
@classmethod
def is_lc_serializable(cls) -> bool:
"""Return whether this class is serializable."""
return True
@property
def lc_serializable(self) -> bool:
"""Return whether this class is serializable."""
return True
AIMessageChunk
class AIMessageChunk(BaseMessageChunk):
"""AI Message chunk."""
type: str = "AIMessageChunk"
"""Type of the message."""
name: Optional[str] = Field(default=None, description="Optional name for the message.")
"""Optional name for the message."""
FunctionMessage
class FunctionMessage(BaseMessage):
"""Message from a function."""
content: str = Field(..., description="The contents of the message.")
"""Contents of the message."""
type: str = "function"
"""Type of the message."""
name: str = Field(..., description="The name of the function.")
"""The name of the function."""
id: Optional[str] = Field(default=None, description="Optional ID for the message.")
"""Optional ID for the message."""
@classmethod
def is_lc_serializable(cls) -> bool:
"""Return whether this class is serializable."""
return True
@property
def lc_serializable(self) -> bool:
"""Return whether this class is serializable."""
return True
FunctionMessageChunk
class FunctionMessageChunk(BaseMessageChunk):
"""Function Message chunk."""
content: Union[str, List[Union[str, Dict]]] = Field(
..., description="The contents of the message."
)
"""Contents of the message."""
type: str = "FunctionMessageChunk"
"""Type of the message."""
name: str = Field(..., description="The name of the function.")
"""The name of the function."""
id: Optional[str] = Field(default=None, description="Optional ID for the message.")
"""Optional ID for the message."""
HumanMessage
class HumanMessage(BaseMessage):
"""Message from a human.
HumanMessages are messages that are passed in from a human to the model.
Example:
.. code-block:: python
from langchain_core.messages import HumanMessage, SystemMessage
messages = [
SystemMessage(
content="You are a helpful assistant! Your name is Bob."
),
HumanMessage(
content="What is your name?"
)
]
# Instantiate a chat model and invoke it with the messages
model = ...
print(model.invoke(messages))
"""
type: str = "human"
"""Type of the message."""
name: Optional[str] = Field(default=None, description="Optional name for the message.")
"""Optional name for the message."""
example: bool = Field(default=False, description="Whether this is an example message.")
"""Whether this is an example message."""
@classmethod
def is_lc_serializable(cls) -> bool:
"""Return whether this class is serializable."""
return True
@property
def lc_serializable(self) -> bool:
"""Return whether this class is serializable."""
return True
HumanMessageChunk
class HumanMessageChunk(BaseMessageChunk):
"""Human Message chunk."""
type: str = "HumanMessageChunk"
"""Type of the message."""
name: Optional[str] = Field(default=None, description="Optional name for the message.")
"""Optional name for the message."""
example: bool = Field(default=False, description="Whether this is an example message.")
"""Whether this is an example message."""
InputTokenDetails
class InputTokenDetails(BaseModel):
"""Breakdown of input token counts.
Does *not* need to sum to full input token count. Does *not* need to have all keys.
Example:
.. code-block:: python
{
"audio": 10,
"cache_creation": 200,
"cache_read": 100,
}
.. versionadded:: 0.3.9
"""
audio: Optional[int] = None
cache_creation: Optional[int] = None
cache_read: Optional[int] = None
InvalidToolCall
class InvalidToolCall(BaseModel):
"""Allowance for errors made by LLM.
Here we add an `error` key to surface errors made during generation
(e.g., invalid JSON arguments.)
"""
name: Optional[str] = Field(default=None)
args: Optional[str] = Field(default=None)
id: str
error: str
type: str = "invalid_tool_call"
OutputTokenDetails
class OutputTokenDetails(BaseModel):
"""Breakdown of output token counts.
Does *not* need to sum to full output token count. Does *not* need to have all keys.
Example:
.. code-block:: python
{
```markdown
### 消息类型
本文档介绍了 LangChain 中使用的不同消息类型。
```python
from typing import Any, Dict, List, Optional, Union
from langchain_core.pydantic_v1 import BaseModel, Field
BaseMessage
所有消息的基类。
class BaseMessage(BaseModel):
"""Base class for all messages."""
content: Union[str, List[Any]] = Field(
"", description="Contents of the message. Can be a string or a list."
)
"""Contents of the message. Can be a string or a list."""
additional_kwargs: Dict = Field(
default_factory=dict,
description="Additional information to include in the message.",
)
"""Additional information to include in the message."""
response_metadata: Dict = Field(
default_factory=dict,
description="Metadata about the response.",
)
"""Metadata about the response."""
type: str
@property
def lc_namespace(self) -> List[str]:
"""Return the namespace of the module."""
return ["langchain_core", "messages"]
@property
def lc_serializable(self) -> bool:
"""Return whether this object is serializable."""
return True
def __add__(self, other: "BaseMessage") -> "BaseMessage":
"""Add two messages together.
This is useful for merging chunks of messages.
"""
raise NotImplementedError()
def dict(self, *args: Any, **kwargs: Any) -> Dict:
"""Return dictionary representation of the message."""
# We do not want to include the type in the dict
# as it is already known.
kwargs["exclude"] = kwargs.get("exclude", set()).union({"type"})
return super().dict(*args, **kwargs)
def json(self, *args: Any, **kwargs: Any) -> str:
"""Return JSON representation of the message."""
# We do not want to include the type in the json
# as it is already known.
kwargs["exclude"] = kwargs.get("exclude", set()).union({"type"})
return super().json(*args, **kwargs)
InputTokenDetails
class InputTokenDetails(BaseModel):
"""Input token details."""
audio: Optional[int] = Field(default=None, description="Number of audio input tokens.")
cache_creation: Optional[int] = Field(
default=None, description="Number of cache creation input tokens."
)
cache_read: Optional[int] = Field(
default=None, description="Number of cache read input tokens."
)
other: Optional[Dict[str, int]] = Field(
default=None, description="Other input token details."
)
OutputTokenDetails
class OutputTokenDetails(BaseModel):
"""Output token details."""
audio: Optional[int] = Field(default=None, description="Number of audio output tokens.")
reasoning: Optional[int] = Field(
default=None, description="Number of reasoning output tokens."
)
other: Optional[Dict[str, int]] = Field(
default=None, description="Other output token details."
)
AIMessage
来自 AI 的消息。
class AIMessage(BaseMessage):
"""Message from an AI."""
type: str = "ai"
name: Optional[str] = Field(default=None, title="Name")
id: Optional[str] = Field(default=None, title="Id")
AIMessageChunk
来自 AI 的消息块（例如，作为流的一部分）。
当合并 AIMessageChunks（例如，通过 AIMessageChunk.__add__）时，所有字符串属性都将被连接。
class AIMessageChunk(BaseMessage):
"""Chunk of a message from an AI (e.g., as part of a stream).
When merging `AIMessageChunks` (e.g., via `AIMessageChunk.__add__`),
all string attributes are concatenated.
Chunks are only merged if their values of `index` are equal and not None.
"""
type: str = "AIMessageChunk"
name: Optional[str] = Field(default=None, title="Name")
id: Optional[str] = Field(default=None, title="Id")
tool_call_chunks: Optional[List["ToolCallChunk"]] = Field(
default=None, title="Tool Call Chunks"
)
def __add__(self, other: "AIMessageChunk") -> "AIMessageChunk":
"""Add two messages together.
This is useful for merging chunks of messages.
"""
if not isinstance(other, AIMessageChunk):
raise TypeError(f"Cannot add {type(other)} to AIMessageChunk")
if self.tool_call_chunks is not None and other.tool_call_chunks is not None:
if len(self.tool_call_chunks) != len(other.tool_call_chunks):
raise ValueError(
"Cannot merge AIMessageChunks with different number of tool calls"
)
merged_tool_call_chunks: List[ToolCallChunk] = []
for i in range(len(self.tool_call_chunks)):
merged_tool_call_chunks.append(
self.tool_call_chunks[i] + other.tool_call_chunks[i]
)
elif self.tool_call_chunks is not None:
merged_tool_call_chunks = self.tool_call_chunks
elif other.tool_call_chunks is not None:
merged_tool_call_chunks = other.tool_call_chunks
else:
merged_tool_call_chunks = None
return AIMessageChunk(
content=self.content + other.content,
name=(self.name or "") + (other.name or ""),
id=(self.id or "") + (other.id or ""),
tool_call_chunks=merged_tool_call_chunks,
)
BaseMessageChunk
消息块的基类。
class BaseMessageChunk(BaseModel):
"""Base class for message chunks."""
content: Union[str, List[Any]] = Field(
"", description="Contents of the message. Can be a string or a list."
)
"""Contents of the message. Can be a string or a list."""
additional_kwargs: Dict = Field(
default_factory=dict,
description="Additional information to include in the message.",
)
"""Additional information to include in the message."""
response_metadata: Dict = Field(
default_factory=dict,
description="Metadata about the response.",
)
"""Metadata about the response."""
type: str
name: Optional[str] = Field(default=None, title="Name")
id: Optional[str] = Field(default=None, title="Id")
@property
def lc_namespace(self) -> List[str]:
"""Return the namespace of the module."""
return ["langchain_core", "messages"]
@property
def lc_serializable(self) -> bool:
"""Return whether this object is serializable."""
return True
def __add__(self, other: "BaseMessageChunk") -> "BaseMessageChunk":
"""Add two messages together.
This is useful for merging chunks of messages.
"""
raise NotImplementedError()
def dict(self, *args: Any, **kwargs: Any) -> Dict:
"""Return dictionary representation of the message."""
# We do not want to include the type in the dict
# as it is already known.
kwargs["exclude"] = kwargs.get("exclude", set()).union({"type"})
return super().dict(*args, **kwargs)
def json(self, *args: Any, **kwargs: Any) -> str:
"""Return JSON representation of the message."""
# We do not want to include the type in the json
# as it is already known.
kwargs["exclude"] = kwargs.get("exclude", set()).union({"type"})
return super().json(*args, **kwargs)
ChatMessage
来自聊天的通用消息。
class ChatMessage(BaseMessage):
"""A generic message from a chat."""
role: str = Field(description="The role of the agent.")
"""The role of the agent."""
type: str =
### PromptTemplateOutput
```json
"""
{
'description': 'Output of a prompt template.',
'oneOf': [{'$ref': '#/$defs/StringPromptValue'},
{'$ref': '#/$defs/ChatPromptValueConcrete'}],
'$defs': {
'StringPromptValue': {
'description': 'A single string prompt value.',
'type': 'string',
'title': 'StringPromptValue'},
'ChatPromptValueConcrete': {
'description': 'A chat prompt value.',
'properties': {
'messages': {
'items': {'$ref': '#/$defs/ChatMessage'},
'type': 'array',
'description': 'The messages in the chat prompt.',
'title': 'Messages'},
'type': {'const': 'chat', 'description': 'The type of the prompt value.', 'title': 'Type'}},
'required': ['messages', 'type'],
'type': 'object',
'title': 'ChatPromptValueConcrete'},
'ChatMessage': {
'description': 'A single chat message.',
'oneOf': [{'$ref': '#/$defs/HumanChatMessage'},
{'$ref': '#/$defs/AIChatMessage'},
{'$ref': '#/$defs/SystemChatMessage'},
{'$ref': '#/$defs/FunctionChatMessage'}],
'title': 'ChatMessage'},
'HumanChatMessage': {
'description': 'A human chat message.',
'properties': {
'content': {'type': 'string', 'description': 'The content of the message.', 'title': 'Content'},
'type': {'const': 'human', 'description': 'The type of the message.', 'title': 'Type'}},
'required': ['content', 'type'],
'type': 'object',
'title': 'HumanChatMessage'},
'AIChatMessage': {
'description': 'An AI chat message.',
'properties': {
'content': {'type': ['string', 'null'], 'description': 'The content of the message.', 'title': 'Content'},
'type': {'const': 'ai', 'description': 'The type of the message.', 'title': 'Type'},
'additional_kwargs': {
'type': 'object',
'description': 'Additional information about the message.',
'title': 'Additional Kwargs'}},
'required': ['type'],
'type': 'object',
'title': 'AIChatMessage'},
'SystemChatMessage': {
'description': 'A system chat message.',
'properties': {
'content': {'type': 'string', 'description': 'The content of the message.', 'title': 'Content'},
'type': {'const': 'system', 'description': 'The type of the message.', 'title': 'Type'}},
'required': ['content', 'type'],
'type': 'object',
'title': 'SystemChatMessage'},
'FunctionChatMessage': {
'description': 'A function chat message.',
'properties': {
'content': {'type': ['string', 'null'], 'description': 'The content of the message.', 'title': 'Content'},
'name': {'type': 'string', 'description': 'The name of the function.', 'title': 'Name'},
'type': {'const': 'function', 'description': 'The type of the message.', 'title': 'Type'}},
'required': ['name', 'type'],
'type': 'object',
'title': 'FunctionChatMessage'},
'UsageMetadata': {
'description': 'Metadata about the usage of the prompt template.',
'properties': {},
'type': 'object'}},
'anyOf': [{'$ref': '#/$defs/StringPromptValue'},
{'$ref': '#/$defs/ChatPromptValueConcrete'}],
'title': 'PromptTemplateOutput'}
"""
消息类型和 LLM 创建
提供的模式非常详尽，详细说明了各种消息类型，例如 AIMessage、HumanMessage、ChatMessage 等。这些定义概述了提示可能生成或与之交互的不同消息组件的结构和关系。
现在，让我们创建 LLM。
llm = ChatOpenAI(model="gpt-4o-mini")
llm.input_schema.model_json_schema()
'type': {'const': 'function',
'default': 'function',
'title': 'Type',
'type': 'string'},
'name': {'title': 'Name', 'type': 'string'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'example': {'default': False, 'title': 'Example', 'type': 'boolean'},
'usage_metadata': {'anyOf': [{'$ref': '#/$defs/UsageMetadata'},
{'type': 'null'}],
'default': None}},
'required': ['content', 'name'],
'title': 'FunctionMessage',
'type': 'object'},
'FunctionMessageChunk': {'additionalProperties': True,
'description': 'Message chunk for passing the result of executing a tool back to a model.',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'FunctionMessageChunk',
'default': 'FunctionMessageChunk',
'title': 'Type',
'type': 'string'},
'name': {'title': 'Name', 'type': 'string'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'example': {'default': False, 'title': 'Example', 'type': 'boolean'},
'usage_metadata': {'anyOf': [{'$ref': '#/$defs/UsageMetadata'},
{'type': 'null'}],
'default': None}},
'required': ['content', 'name'],
'title': 'FunctionMessageChunk',
'type': 'object'},
'HumanMessage': {'additionalProperties': True,
'description': 'Message from a human.',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'human',
'default': 'human',
'title': 'Type',
'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'example': {'default': False, 'title': 'Example', 'type': 'boolean'},
'usage_metadata': {'anyOf': [{'$ref': '#/$defs/UsageMetadata'},
{'type': 'null'}],
'default': None}},
'required': ['content'],
'title': 'HumanMessage',
'type': 'object'},
'HumanMessageChunk': {'additionalProperties': True,
'description': 'Message chunk from a human.',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'HumanMessageChunk',
'default': 'HumanMessageChunk',
'title': 'Type',
'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'example': {'default': False, 'title': 'Example', 'type': 'boolean'},
'usage_metadata': {'anyOf': [{'$ref': '#/$defs/UsageMetadata'},
{'type': 'null'}],
'default': None}},
'required': ['content'],
'title': 'HumanMessageChunk',
'type': 'object'},
'InvalidToolCall': {'additionalProperties': True,
'description': 'Represents an invalid tool call.',
'properties': {'id': {'title': 'Id', 'type': 'string'},
'name': {'title': 'Name', 'type': 'string'},
'args': {'title': 'Args', 'type': 'string'},
'message_log': {'items': {'$ref': '#/$defs/BaseMessage'},
'title': 'Message Log',
'type': 'array'},
'invalid_message': {'$ref': '#/$defs/BaseMessage',
'title': 'Invalid Message'}},
'required': ['id', 'name', 'args', 'message_log', 'invalid_message'],
'title': 'InvalidToolCall',
'type': 'object'},
'SystemMessage': {'additionalProperties': True,
'description': 'Message from the system.',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'system',
'default': 'system',
'title': 'Type',
'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'example': {'default': False, 'title': 'Example', 'type': 'boolean'},
'usage_metadata': {'anyOf': [{'$ref': '#/$defs/UsageMetadata'},
{'type': 'null'}],
'default': None}},
'required': ['content'],
'title': 'SystemMessage',
'type': 'object'},
'SystemMessageChunk': {'additionalProperties': True,
'description': 'Message chunk from the system.',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'SystemMessageChunk',
'default': 'SystemMessageChunk',
'title': 'Type',
'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'example': {'default': False, 'title': 'Example', 'type': 'boolean'},
'usage_metadata': {'anyOf': [{'$ref': '#/$defs/UsageMetadata'},
{'type': 'null'}],
'default': None}},
'required': ['content'],
'title': 'SystemMessageChunk',
'type': 'object'},
'ToolCall': {'additionalProperties': True,
'description': 'Represents a tool call.',
'properties': {'id': {'title': 'Id', 'type': 'string'},
```markdown
### langchain_core.messages
```python
from typing import Any, Dict, List, Optional, Union
from langchain_core.pydantic_v1 import BaseModel, Field
BaseMessage
class BaseMessage(BaseModel):
"""Base class for all messages."""
content: Union[str, List[Union[str, Dict]]] = Field(
..., description="Contents of the message."
)
"""Contents of the message."""
additional_kwargs: Dict = Field(
default_factory=dict,
description="Additional information to pass to the model.",
)
"""Additional information to pass to the model."""
response_metadata: Dict = Field(
default_factory=dict, description="Metadata about the response."
)
"""Metadata about the response."""
type: str
"""Type of the message."""
name: Optional[str] = Field(default=None)
"""Name of the message sender."""
id: Optional[str] = Field(default=None)
"""Unique ID of the message."""
def __init__(self, **data: Any):
"""Initialize a message."""
super().__init__(**data)
@property
def lc_namespace(self) -> List[str]:
"""Return the namespace of the module."""
return ["langchain_core", "messages"]
@property
def lc_serializable(self) -> bool:
"""Return whether this object is serializable."""
return True
AIMessage
class AIMessage(BaseMessage):
"""Message from an AI."""
type: str = "ai"
"""Type of the message."""
name: Optional[str] = Field(default=None)
"""Name of the message sender."""
id: Optional[str] = Field(default=None)
"""Unique ID of the message."""
def __init__(self, **data: Any):
"""Initialize a message."""
super().__init__(**data)
AIMessageChunk
class AIMessageChunk(BaseMessage):
"""AI Message chunk."""
type: str = "AIMessageChunk"
"""Type of the message."""
name: Optional[str] = Field(default=None)
"""Name of the message sender."""
id: Optional[str] = Field(default=None)
"""Unique ID of the message."""
def __init__(self, **data: Any):
"""Initialize a message."""
super().__init__(**data)
BaseMessageChunk
class BaseMessageChunk(BaseModel):
"""Base class for all message chunks."""
content: Union[str, List[Union[str, Dict]]] = Field(
..., description="Contents of the message."
)
"""Contents of the message."""
additional_kwargs: Dict = Field(
default_factory=dict,
description="Additional information to pass to the model.",
)
"""Additional information to pass to the model."""
response_metadata: Dict = Field(
default_factory=dict, description="Metadata about the response."
)
"""Metadata about the response."""
type: str
"""Type of the message."""
name: Optional[str] = Field(default=None)
"""Name of the message sender."""
id: Optional[str] = Field(default=None)
"""Unique ID of the message."""
def __init__(self, **data: Any):
"""Initialize a message."""
super().__init__(**data)
@property
def lc_namespace(self) -> List[str]:
"""Return the namespace of the module."""
return ["langchain_core", "messages"]
@property
def lc_serializable(self) -> bool:
"""Return whether this object is serializable."""
return True
ChatMessage
class ChatMessage(BaseMessage):
"""Message from any actor in a chat."""
role: str = Field(..., description="The role of the agent.")
"""The role of the agent."""
type: str = "chat"
"""Type of the message."""
name: Optional[str] = Field(default=None)
"""Name of the message sender."""
id: Optional[str] = Field(default=None)
"""Unique ID of the message."""
def __init__(self, **data: Any):
"""Initialize a message."""
super().__init__(**data)
ChatMessageChunk
class ChatMessageChunk(BaseMessageChunk):
"""Chat Message chunk."""
role: str = Field(..., description="The role of the agent.")
"""The role of the agent."""
type: str = "ChatMessageChunk"
"""Type of the message."""
name: Optional[str] = Field(default=None)
"""Name of the message sender."""
id: Optional[str] = Field(default=None)
"""Unique ID of the message."""
def __init__(self, **data: Any):
"""Initialize a message."""
super().__init__(**data)
FunctionMessage
class FunctionMessage(BaseMessage):
"""Message from a function call."""
name: str = Field(..., description="The name of the function.")
"""The name of the function."""
type: str = "function"
"""Type of the message."""
id: Optional[str] = Field(default=None)
"""Unique ID of the message."""
def __init__(self, **data: Any):
"""Initialize a message."""
super().__init__(**data)
FunctionMessageChunk
class FunctionMessageChunk(BaseMessageChunk):
"""Function Message chunk."""
name: str = Field(..., description="The name of the function.")
"""The name of the function."""
type: str = "FunctionMessageChunk"
"""Type of the message."""
id: Optional[str] = Field(default=None)
"""Unique ID of the message."""
def __init__(self, **data: Any):
"""Initialize a message."""
super().__init__(**data)
HumanMessage
class HumanMessage(BaseMessage):
"""Message from a human.
HumanMessages are messages that are passed in from a human to the model.
Example:
.. code-block:: python
from langchain_core.messages import HumanMessage, SystemMessage
messages = [
SystemMessage(
content="You are a helpful assistant! Your name is Bob."
),
HumanMessage(
content="What is your name?"
)
]
# Instantiate a chat model and invoke it with the messages
model = ...
print(model.invoke(messages))
"""
type: str = "human"
"""Type of the message."""
name: Optional[str] = Field(default=None)
"""Name of the message sender."""
id: Optional[str] = Field(default=None)
"""Unique ID of the message."""
example: bool = Field(default=False)
"""Whether the message is an example."""
def __init__(self, **data: Any):
"""Initialize a message."""
super().__init__(**data)
HumanMessageChunk
class HumanMessageChunk(BaseMessageChunk):
"""Human Message chunk."""
type: str = "HumanMessageChunk"
"""Type of the message."""
name: Optional[str] = Field(default=None)
"""Name of the message sender."""
id: Optional[str] = Field(default=None)
"""Unique ID of the message."""
example: bool = Field(default=False)
"""Whether the message is an example."""
def __init__(self, **data: Any):
"""Initialize a message."""
super().__init__(**data)
InputTokenDetails
class InputTokenDetails(BaseModel):
"""Breakdown of input token counts.
Does *not* need to sum to full input token count. Does *not* need to have all keys.
Example:
.. code-block:: python
{
"audio": 10,
"cache_creation": 200,
"cache_read": 100,
}
.. versionadded:: 0.3.9
"""
audio: Optional[int] = None
cache_creation: Optional[int] = None
cache_read:
```markdown
### 消息类型
```json
{
"$defs": {
"InputTokenDetails": {
"description": "Details about the input tokens.",
"additionalProperties": {
"type": "integer"
},
"title": "InputTokenDetails",
"type": "object"
},
"OutputTokenDetails": {
"description": "Details about the output tokens.",
"additionalProperties": {
"type": "integer"
},
"title": "OutputTokenDetails",
"type": "object"
}
},
"AIMessage": {
"description": "Message from an AI.",
"properties": {
"content": {
"anyOf": [
{
"type": "string"
},
{
"items": {
"anyOf": [
{
"type": "string"
},
{
"type": "object"
}
]
},
"type": "array"
}
],
"title": "Content"
},
"additional_kwargs": {
"title": "Additional Kwargs",
"type": "object"
},
"response_metadata": {
"title": "Response Metadata",
"type": "object"
},
"type": {
"const": "ai",
"default": "ai",
"title": "Type",
"type": "string"
},
"name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"title": "Name"
},
"id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"title": "Id"
},
"tool_calls": {
"items": {
"$ref": "#/$defs/ToolCall"
},
"title": "Tool Calls",
"type": "array"
},
"usage": {
"allOf": [
{
"$ref": "#/$defs/UsageMetadata"
},
{
"description": "Usage metadata for a message, such as token counts.",
"title": "Usage Metadata",
"type": "object"
}
],
"title": "Usage"
}
},
"required": [
"content"
],
"title": "AIMessage",
"type": "object"
},
"AIMessageChunk": {
"description": "Chunk of an AI message (e.g., as part of a stream).",
"properties": {
"content": {
"anyOf": [
{
"type": "string"
},
{
"items": {
"anyOf": [
{
"type": "string"
},
{
"type": "object"
}
]
},
"type": "array"
}
],
"title": "Content"
},
"additional_kwargs": {
"title": "Additional Kwargs",
"type": "object"
},
"response_metadata": {
"title": "Response Metadata",
"type": "object"
},
"type": {
"const": "AIMessageChunk",
"default": "AIMessageChunk",
"title": "Type",
"type": "string"
},
"name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"title": "Name"
},
"id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"title": "Id"
},
"tool_call_chunks": {
"items": {
"$ref": "#/$defs/ToolCallChunk"
},
"title": "Tool Call Chunks",
"type": "array"
},
"usage": {
"allOf": [
{
"$ref": "#/$defs/UsageMetadata"
},
{
"description": "Usage metadata for a message, such as token counts.",
"title": "Usage Metadata",
"type": "object"
}
],
"title": "Usage"
}
},
"required": [
"content"
],
"title": "AIMessageChunk",
"type": "object"
},
"BaseMessage": {
"description": "Base class for all messages.",
"properties": {
"content": {
"anyOf": [
{
"type": "string"
},
{
"items": {
"anyOf": [
{
"type": "string"
},
{
"type": "object"
}
]
},
"type": "array"
}
],
"title": "Content"
},
"additional_kwargs": {
"title": "Additional Kwargs",
"type": "object"
},
"response_metadata": {
"title": "Response Metadata",
"type": "object"
},
"type": {
"title": "Type",
"type": "string"
},
"name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"title": "Name"
},
"id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"title": "Id"
}
},
"required": [
"content"
],
"title": "BaseMessage",
"type": "object"
},
"ChatMessage": {
"description": "Message from a chat.",
"properties": {
"content": {
"anyOf": [
{
"type": "string"
},
{
"items": {
"anyOf": [
{
"type": "string"
},
{
"type": "object"
}
]
},
"type": "array"
}
],
"title": "Content"
},
"additional_kwargs": {
"title": "Additional Kwargs",
"type": "object"
},
"response_metadata": {
"title": "Response Metadata",
"type": "object"
},
"type": {
"const": "chat",
"default": "chat",
"title": "Type",
"type": "string"
},
"role": {
"title": "Role",
"type": "string"
},
"name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"title": "Name"
},
"id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"title": "Id"
}
},
"required": [
"content",
"role"
],
"title": "ChatMessage",
"type": "object"
},
"ChatMessageChunk": {
"description": "Chunk of a chat message (
ChatOpenAIInput Schema
{
'description': 'ChatOpenAI的输入',
'additionalProperties': False,
'properties': {
'model_name': {
'title': 'Model Name',
'type': 'string',
'description': '要使用的模型名称。'},
'temperature': {
'default': 0.0,
'description': '使用什么采样温度。较高的值意味着模型将承担更多风险。对于更具创造性的应用，尝试0.9，对于有明确答案的应用，尝试0（argmax采样）。',
'title': 'Temperature',
'type': 'number'},
'top_p': {
'default': 1.0,
'description': '一种替代温度采样的方案，称为核采样，其中模型考虑具有top_p概率质量的标记的结果。因此，0.1意味着只考虑包含前10％概率质量的标记。',
'title': 'Top P',
'type': 'number'},
'presence_penalty': {
'default': 0.0,
'description': '介于-2.0和2.0之间的数字。正值会根据新标记是否已出现在文本中来惩罚新标记，从而增加模型谈论新主题的可能性。',
'title': 'Presence Penalty',
'type': 'number'},
'frequency_penalty': {
'default': 0.0,
'description': '介于-2.0和2.0之间的数字。正值会根据新标记在文本中已有的频率来惩罚新标记，从而降低模型逐字重复同一行的可能性。',
'title': 'Frequency Penalty',
'type': 'number'},
'n': {
'default': 1,
'description': '为每个提示生成多少个补全。',
'title': 'N',
'type': 'integer'},
'max_tokens': {
'description': '在补全中生成的最大标记数。您的提示的标记计数加上max_tokens不能超过模型的上下文长度。大多数模型的上下文长度为2048个标记（最新的模型除外，它们支持4096个）。',
'title': 'Max Tokens',
'type': 'integer'},
'model_kwargs': {
'description': '保存任何对`model_name`有效的模型参数。',
'title': 'Model Kwargs',
'type': 'object'},
'stop': {
'anyOf': [{'type': 'string'}, {'items': {'type': 'string'}, 'type': 'array'}],
'description': 'API将停止生成更多标记的最多4个序列。',
'title': 'Stop'},
'streaming': {
'default': False,
'description': '是否流式传输结果。',
'title': 'Streaming',
'type': 'boolean'},
'callbacks': {
'anyOf': [{
'items': {'$ref': '#/$defs/BaseCallbackHandler'},
'type': 'array'},
{'$ref': '#/$defs/BaseCallbackHandler'}],
'description': '用于此链的回调。',
'title': 'Callbacks'},
'tags': {
'items': {'type': 'string'},
'description': '用于此链的标签。',
'title': 'Tags',
'type': 'array'},
'metadata': {
'description': '用于此链的元数据。',
'title': 'Metadata',
'type': 'object'},
'verbose': {
'default': False,
'description': '是否打印详细输出。',
'title': 'Verbose',
'type': 'boolean'},
'return_only_outputs': {
'default': False,
'description': '是否仅返回输出或完整响应。',
'title': 'Return Only Outputs',
'type': 'boolean'},
'handle_tokenization_error': {
'default': False,
'description': '是否处理标记化错误。',
'title': 'Handle Tokenization Error',
'type': 'boolean'},
'input_variables': {
'default': ['input'],
'description': '要使用的输入变量。',
'items': {'type': 'string'},
'title': 'Input Variables',
'type': 'array'},
'prompt': {
'anyOf': [{'$ref': '#/$defs/BasePromptTemplate'},
{'type': 'string'},
{'items': {'oneOf': [{'$ref': '#/$defs/AIMessage'},
{'$ref': '#/$defs/HumanMessage'},
{'$ref': '#/$defs/ChatMessage'},
{'$ref': '#/$defs/SystemMessage'},
{'$ref': '#/$defs/FunctionMessage'},
{'$ref': '#/$defs/ToolMessage'},
{'$ref': '#/$defs/AIMessageChunk'},
{'$ref': '#/$defs/HumanMessageChunk'},
{'$ref': '#/$defs/ChatMessageChunk'},
{'$ref': '#/$defs/SystemMessageChunk'},
{'$ref': '#/$defs/FunctionMessageChunk'},
{'$ref': '#/$defs/ToolMessageChunk'}]},
'type': 'array'}],
'title': 'Prompt'},
'messages': {
'description': '用于聊天提示的消息列表。',
'title': 'Messages',
'anyOf': [{'type': 'string'},
{'$ref': '#/$defs/StringPromptValue'},
{'$ref': '#/$defs/ChatPromptValueConcrete'},
{'items': {'oneOf': [{'$ref': '#/$defs/AIMessage'},
{'$ref': '#/$defs/HumanMessage'},
{'$ref': '#/$defs/ChatMessage'},
{'$ref': '#/$defs/SystemMessage'},
{'$ref': '#/$defs/FunctionMessage'},
{'$ref': '#/$defs/ToolMessage'},
{'$ref': '#/$defs/AIMessageChunk'},
{'$ref': '#/$defs/HumanMessageChunk'},
{'$ref': '#/$defs/ChatMessageChunk'},
{'$ref': '#/$defs/SystemMessageChunk'},
{'$ref': '#/$defs/FunctionMessageChunk'},
{'$ref': '#/$defs/ToolMessageChunk'}]},
'type': 'array'}],
'title': 'Messages'},
'extra_kwargs': {
'description': '已弃用。请改用model_kwargs。',
'title': 'Extra Kwargs',
'type': 'object'},
'api_key': {
'description': 'OpenAI API密钥。',
'title': 'Api Key',
'type': 'string'},
'openai_api_key': {
'description': 'OpenAI API密钥。',
'title': 'Openai Api Key',
'type': 'string'},
'openai_api_base': {
'description': 'OpenAI API的基本URL。',
'title': 'Openai Api Base',
'type': 'string'},
'openai_api_type': {
'description': 'OpenAI API的类型。',
'title': 'Openai Api Type',
'type': 'string'},
'openai_api_version': {
'description': 'OpenAI API的API版本。',
'title': 'Openai Api Version',
'type': 'string'},
'organization': {
'description': 'OpenAI组织ID。',
'title': 'Organization',
'type': 'string'},
'allowed_special': {
'default': ['<|file_separator|>'],
'description': '允许的特殊标记。',
'items': {'type': 'string'},
'title': 'Allowed Special',
'type': 'array'},
'client': {
'description': '要使用的OpenAI客户端。',
'title': 'Client',
'anyOf': [{'$ref': '#/$defs/OpenAI'}, {'type': 'null'}]},
'cache': {
'description': '用于此链的缓存。',
'title': 'Cache',
'anyOf': [{'$ref': '#/$defs/BaseCache'}, {'type': 'null'}]},
'request_timeout': {
'description': 'OpenAI API请求的超时时间。',
'title': 'Request Timeout',
'type': ['integer', 'null']},
}
}
### llm.output_schema.model_json_schema()
此函数返回一个 JSON 模式，描述了 LLM（大型语言模型）的输出结构。这对于验证 LLM 的输出是否符合预期格式非常有用。
#### 函数签名
```python
llm.output_schema.model_json_schema()
返回值
一个 JSON 模式（Python 字典），描述了 LLM 的输出结构。
示例
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import chain
from langchain_openai import ChatOpenAI
## 定义一个 Pydantic 模型来描述输出
class Person(BaseModel):
name: str = Field(description="The person's name")
age: int = Field(description="The person's age")
fav_food: str = Field(description="The person's favorite food")
## 创建一个 LLM 实例
llm = ChatOpenAI(model="gpt-3.5-turbo-0613", temperature=0)
## 创建一个提示模板
prompt = ChatPromptTemplate.from_messages(
[
("system", "You are a helpful assistant that extracts information."),
("user", "Extract the following information: {information}"),
]
)
## 创建一个 Runnable 序列
chain = (
prompt
| llm.with_structured_output(Person)
)
## 获取 JSON 模式
json_schema = Person.model_json_schema()
print(json_schema)
## 使用 JSON 模式来获取 LLM 的输出
output = chain.invoke({"information": "My name is John, I am 30 years old and I like pizza."})
print(output)
详细说明
llm.output_schema.model_json_schema() 函数主要用于定义 LLM 输出的结构，并将其作为提示的一部分传递给 LLM。这有助于确保 LLM 的输出符合预期的格式，从而更容易地进行后续处理。
该函数通常与 with_structured_output() 方法结合使用，用于指定 LLM 输出的结构。with_structured_output() 方法接受一个 Pydantic 模型作为参数，该模型定义了输出的字段和数据类型。model_json_schema() 方法用于生成该 Pydantic 模型的 JSON 模式，然后将其传递给 LLM，从而指导 LLM 生成符合该 JSON 模式的输出。
通过使用 llm.output_schema.model_json_schema()，可以有效地控制 LLM 的输出格式，提高输出的可预测性和可处理性。
}, ‘id’: {‘anyOf’: [{‘type’: ‘string’}, {‘type’: ‘null’}], ‘default’: None, ‘title’: ‘Id’}}, ‘required’: [‘content’, ‘name’], ‘title’: ‘FunctionMessageChunk’, ‘type’: ‘object’}, ‘HumanMessage’: {‘additionalProperties’: True, ‘description’: ‘Message from a human.’, ‘properties’: {‘content’: {‘anyOf’: [{‘type’: ‘string’}, {‘items’: {‘anyOf’: [{‘type’: ‘string’}, {‘type’: ‘object’}]}, ‘type’: ‘array’}], ‘title’: ‘Content’}, ‘additional_kwargs’: {‘title’: ‘Additional Kwargs’, ‘type’: ‘object’}, ‘response_metadata’: {‘title’: ‘Response Metadata’, ‘type’: ‘object’}, ‘type’: {‘const’: ‘human’, ‘default’: ‘human’, ‘title’: ‘Type’, ‘type’: ‘string’}, ‘name’: {‘anyOf’: [{‘type’: ‘string’}, {‘type’: ‘null’}], ‘default’: None, ‘title’: ‘Name’}, ‘id’: {‘anyOf’: [{‘type’: ‘string’}, {‘type’: ‘null’}], ‘default’: None, ‘title’: ‘Id’}, ‘example’: {‘default’: False, ‘title’: ‘Example’, ‘type’: ‘boolean’}}, ‘required’: [‘content’], ‘title’: ‘HumanMessage’, ‘type’: ‘object’}, ‘HumanMessageChunk’: {‘additionalProperties’: True, ‘description’: ‘Human message chunk.’, ‘properties’: {‘content’: {‘anyOf’: [{‘type’: ‘string’}, {‘items’: {‘anyOf’: [{‘type’: ‘string’}, {‘type’: ‘object’}]}, ‘type’: ‘array’}], ‘title’: ‘Content’}, ‘additional_kwargs’: {‘title’: ‘Additional Kwargs’, ‘type’: ‘object’}, ‘response_metadata’: {‘title’: ‘Response Metadata’, ‘type’: ‘object’}, ‘type’: {‘const’: ‘HumanMessageChunk’, ‘default’: ‘HumanMessageChunk’, ‘title’: ‘Type’, ‘type’: ‘string’}, ‘name’: {‘anyOf’: [{‘type’: ‘string’}, {‘type’: ‘null’}], ‘default’: None, ‘title’: ‘Name’}, ‘id’: {‘anyOf’: [{‘type’: ‘string’}, {‘type’: ‘null’}], ‘default’: None, ‘title’: ‘Id’}, ‘example’: {‘default’: False, ‘title’: ‘Example’, ‘type’: ‘boolean’}}, ‘required’: [‘content’], ‘title’: ‘HumanMessageChunk’, ‘type’: ‘object’}, ‘InvalidToolCall’: {‘additionalProperties’: True, ‘description’: ‘Represents an invalid tool call.’, ‘properties’: {‘id’: {‘title’: ‘Id’, ‘type’: ‘string’}, ‘name’: {‘title’: ‘Name’, ‘type’: ‘string’}, ‘type’: {‘const’: ‘invalid_tool_call’, ‘default’: ‘invalid_tool_call’, ‘title’: ‘Type’, ‘type’: ‘string’}, ‘invalid_tool_call_error’: {‘title’: ‘Invalid Tool Call Error’, ‘type’: ‘string’}}, ‘required’: [‘id’, ‘name’, ‘invalid_tool_call_error’], ‘title’: ‘InvalidToolCall’, ‘type’: ‘object’}, ‘InvalidToolCallChunk’: {‘additionalProperties’: True, ‘description’: ‘Represents an invalid tool call chunk.’, ‘properties’: {‘id’: {‘title’: ‘Id’, ‘type’: ‘string’}, ‘name’: {‘title’: ‘Name’, ‘type’: ‘string’}, ‘type’: {‘const’: ‘InvalidToolCallChunk’, ‘default’: ‘InvalidToolCallChunk’, ‘title’: ‘Type’, ‘type’: ‘string’}, ‘invalid_tool_call_error’: {‘title’: ‘Invalid Tool Call Error’, ‘type’: ‘string’}}, ‘required’: [‘id’, ‘name’, ‘invalid_tool_call_error’], ‘title’: ‘InvalidToolCallChunk’, ‘type’: ‘object’}, ‘SystemMessage’: {‘additionalProperties’: True, ‘description’: ‘Message from the system.’, ‘properties’: {‘content’: {‘anyOf’: [{‘type’: ‘string’}, {‘items’: {‘anyOf’: [{‘type’: ‘string’}, {‘type’: ‘object’}]}, ‘type’: ‘array’}], ‘title’: ‘Content’}, ‘additional_kwargs’: {‘title’: ‘Additional Kwargs’, ‘type’: ‘object’}, ‘response_metadata’: {‘title’: ‘Response Metadata’, ‘type’: ‘object’}, ‘type’: {‘const’: ‘system’, ‘default’: ‘system’, ‘title’: ‘Type’, ‘type’: ‘string’}, ‘name’: {‘anyOf’: [{‘type’: ‘string’}, {‘type’: ‘null’}], ‘default’: None, ‘title’: ‘Name’}, ‘id’: {‘anyOf’: [{‘type’: ‘string’}, {‘type’: ‘null’}], ‘default’: None, ‘title’: ‘Id’}, ‘example’: {‘default’: False, ‘title’: ‘Example’, ‘type’: ‘boolean’}}, ‘required’: [‘content’], ‘title’: ‘SystemMessage’, ‘type’: ‘object’}, ‘SystemMessageChunk’: {‘additionalProperties’: True, ‘description’: ‘System message chunk.’, ‘properties’: {‘content’: {‘anyOf’: [{‘type’: ‘string’}, {‘items’: {‘anyOf’: [{‘type’: ‘string’}, {‘type’: ‘object’}]}, ‘type’: ‘array’}], ‘title’: ‘Content’}, ‘additional_kwargs’: {‘title’: ‘Additional Kwargs’, ‘type’: ‘object’}, ‘response_metadata’: {‘title’: ‘Response Metadata’, ‘type’: ‘object’}, ‘type’: {‘const’: ‘SystemMessageChunk’, ‘default’: ‘SystemMessageChunk’, ‘title’: ‘Type’, ‘type’: ‘string’}, ‘name’: {‘anyOf’: [{‘type’: ‘string’}, {‘type’: ‘null’}], ‘default’: None, ‘title’: ‘Name’}, ‘id’: {‘anyOf’: [{‘type’: ‘string’}, {‘type’: ‘null’}], ‘default’: None, ‘title’: ‘Id’}, ‘example’: {‘default’: False, ‘title’: ‘Example’, ‘type’: ‘boolean’}}, ‘required’: [‘content’], ‘title’: ‘SystemMessageChunk’, ‘type’: ‘object’}, ‘ToolCall’: {‘additionalProperties’: True, ‘description’: ‘Represents a tool call.’, ‘properties’: {‘id’: {‘title’: ‘Id’, ‘type’: ‘string’}, ‘type’: {‘const’: ‘function’, ‘default’: ‘function’, ‘title’: ‘Type’, ‘type’: ‘string’}, ‘function’: {‘title’: ‘Function’, ‘type’: ‘object’, ‘properties’: {‘name’: {‘title’: ‘Name’, ‘type’: ‘string’}, ‘arguments’: {‘title’: ‘Arguments’, ‘type’: ‘string’}}, ‘required’: [‘name’, ‘arguments’]}}, ‘required’: [‘id’, ‘function’], ‘title’: ‘ToolCall’, ‘type’: ‘object’}, ‘ToolCallChunk’: {‘additionalProperties’: True, ‘description’: ‘Represents a tool call chunk.’, ‘properties’: {‘id’: {‘title’: ‘Id’, ‘type’: ‘string’}, ‘type’: {‘const’: ‘ToolCallChunk’, ‘default’: ‘ToolCallChunk’, ‘title’: ‘Type’, ‘type’: ‘string’}, ‘function’: {‘title’: ‘Function’, ‘type’: ‘object’, ‘properties’: {‘name’: {‘title’: ‘Name’, ‘type’: ‘string’}, ‘arguments’: {‘title’: ‘Arguments’, ‘type’: ‘string’}}, ‘required’: [‘name’, ‘arguments’]}}, ‘required’: [‘id’, ‘function’], ‘title’: ‘ToolCallChunk’, ‘type’: ‘object’}, ‘ToolMessage’: {‘additionalProperties’: True, ‘description’: ‘Message for passing the result of executing a tool back to a model.’, ‘properties’: {‘content’: {‘anyOf’: [{‘type’: ‘string’}, {‘items’: {‘anyOf’: [{‘type’: ‘string’}, {‘type’: ‘object’}]}, ‘type’: ‘array’}], ‘title’: ‘Content’}, ‘additional_kwargs’: {‘title’: ‘Additional Kw
消息类型
以下是 LangChain 中使用的各种消息类型的定义。
AIMessage
from langchain_core.messages import AIMessage
## Define an AI message
message = AIMessage(content="你好！")
print(message)
AIMessageChunk
from langchain_core.messages import AIMessageChunk
## Define an AI message chunk
message = AIMessageChunk(content="你好！")
print(message)
BaseMessage
from langchain_core.messages import BaseMessage
## Define a base message
message = BaseMessage(content="你好！", type="base")
print(message)
FunctionMessage
from langchain_core.messages import FunctionMessage
## Define a function message
message = FunctionMessage(content="你好！", name="my_function")
print(message)
FunctionMessageChunk
from langchain_core.messages import FunctionMessageChunk
## Define a function message chunk
message = FunctionMessageChunk(content="你好！", name="my_function")
print(message)
HumanMessage
from langchain_core.messages import HumanMessage
## Define a human message
message = HumanMessage(content="你好！")
print(message)
HumanMessageChunk
from langchain_core.messages import HumanMessageChunk
## Define a human message chunk
message = HumanMessageChunk(content="你好！")
print(message)
InputTokenDetails
from langchain_core.messages import InputTokenDetails
## Define input token details
message = InputTokenDetails(audio=10, cache_creation=200, cache_read=100)
print(message)
InvalidToolCall
from langchain_core.messages import InvalidToolCall
## Define an invalid tool call
message = InvalidToolCall(name="my_tool", args="{}", id="123", error="Invalid JSON")
print(message)
OutputTokenDetails
from langchain_core.messages import OutputTokenDetails
## Define output token details
message = OutputTokenDetails(reasoning=200)
print(message)
SystemMessage
from langchain_core.messages import SystemMessage
## Define a system message
message = SystemMessage(content="你是一个有用的助手！")
print(message)
from langchain_core.messages import HumanMessage, SystemMessage
messages = [
SystemMessage(
content="You are a helpful assistant! Your name is Bob."
),
HumanMessage(
content="What is your name?"
)
]
## Instantiate a chat model and invoke it with the messages
model = ...
print(model.invoke(messages))
SystemMessageChunk
from langchain_core.messages import SystemMessageChunk
## Define a system message chunk
message = SystemMessageChunk(content="你是一个有用的助手！")
print(message)
ToolCall
from langchain_core.messages import ToolCall
## Define a tool call
message = ToolCall(name="my_tool", args={"a": 1}, id="123")
print(message)
ToolCallChunk
from langchain_core.messages import ToolCallChunk
## Define a tool call chunk
message = ToolCallChunk(name="my_tool", args={"a": 1}, id="123")
print(message)
### ChatOpenAIOutput
```json
{
'description': 'ChatOpenAI 调用的输出。',
'definitions': {
'InputTokenDetails': {
'additionalProperties': {'type': 'integer'},
'description': '关于输入 token 的详细信息。',
'title': 'InputTokenDetails',
'type': 'object'
},
'OutputTokenDetails': {
'additionalProperties': {'type': 'integer'},
'description': '关于输出 token 的详细信息。',
'title': 'OutputTokenDetails',
'type': 'object'
},
'AIMessage': {
'additionalProperties': True,
'description': '来自 AI 的消息。',
'properties': {
'content': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'ai', 'default': 'ai', 'title': 'Type', 'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'tool_calls': {
'items': {'$ref': '#/$defs/ToolCall'},
'title': 'Tool Calls',
'type': 'array'
},
'function_call': {'$ref': '#/$defs/FunctionCall', 'title': 'Function Call'}
},
'required': ['content'],
'title': 'AIMessage',
'type': 'object'},
'AIMessageChunk': {
'additionalProperties': True,
'description': 'AI 消息的块。',
'properties': {
'content': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'AIMessageChunk',
'default': 'AIMessageChunk',
'title': 'Type',
'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'tool_call_chunks': {
'items': {'$ref': '#/$defs/ToolCallChunk'},
'title': 'Tool Call Chunks',
'type': 'array'
},
'function_call': {'$ref': '#/$defs/FunctionCall', 'title': 'Function Call'}
},
'required': ['content'],
'title': 'AIMessageChunk',
'type': 'object'},
'ChatMessage': {
'additionalProperties': True,
'description': '来自任何参与者的消息。',
'properties': {
'content': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'chat', 'default': 'chat', 'title': 'Type', 'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'}
},
'required': ['content', 'name'],
'title': 'ChatMessage',
'type': 'object'},
'ChatMessageChunk': {
'additionalProperties': True,
'description': '聊天消息的块。',
'properties': {
'content': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'ChatMessageChunk',
'default': 'ChatMessageChunk',
'title': 'Type',
'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'}
},
'required': ['content', 'name'],
'title': 'ChatMessageChunk',
'type': 'object'},
'FunctionCall': {
'description': '函数调用。',
'properties': {'name': {'type': 'string', 'title': 'Name'},
'arguments': {'type': 'string', 'title': 'Arguments'}},
'required': ['name', 'arguments'],
'title': 'FunctionCall',
'type': 'object'},
'FunctionMessage': {
'additionalProperties': True,
'description': '用于将函数输出传回模型的消息。',
'properties': {
'content': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'function',
'default': 'function',
'title': 'Type',
'type': 'string'},
'name': {'type': 'string', 'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'}
},
'required': ['content', 'name'],
'title': 'FunctionMessage',
'type': 'object'},
'FunctionMessageChunk': {
'additionalProperties': True,
'description': '函数消息的块。',
'properties': {
'content': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'FunctionMessageChunk',
'default': 'FunctionMessageChunk',
'title': 'Type',
'type': 'string'},
'name': {'type': 'string', 'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'}
},
'required': ['content', 'name'],
'title': 'FunctionMessageChunk',
'type': 'object'},
'HumanMessage': {
'additionalProperties': True,
'description': '来自人类的消息。',
'properties': {
'content': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'human',
'default': 'human',
'title': 'Type',
'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'}
},
'required': ['content'],
class MovieReview(BaseModel): movie_title: str = Field(description=“电影的标题”) review: str = Field(description=“电影的评论”) language: str = Field(description=“评论的语言”)
parser = PydanticOutputParser(pydantic_object=MovieReview) format_instructions = parser.get_format_instructions()
parser.input_schema.model_json_schema()
```markdown
### 消息类型
```json
{'$defs': {'AIMessage': {'additionalProperties': True,
'description': '来自 AI 的消息。\n\nAIMessage 作为对提示的响应从聊天模型返回。\n\n此消息表示模型的输出，由模型返回的原始输出以及 LangChain 框架添加的标准化字段（例如，工具调用、使用元数据）组成。',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'ai',
'default': 'ai',
'title': 'Type',
'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'example': {'default': False, 'title': 'Example', 'type': 'boolean'},
'tool_calls': {'default': [],
'items': {'$ref': '#/$defs/ToolCall'},
'title': 'Tool Calls',
'type': 'array'},
'invalid_tool_calls': {'default': [],
'items': {'$ref': '#/$defs/InvalidToolCall'},
'title': 'Invalid Tool Calls',
'type': 'array'},
'usage_metadata': {'anyOf': [{'$ref': '#/$defs/UsageMetadata'},
{'type': 'null'}],
'default': None}},
'required': ['content'],
'title': 'AIMessage',
'type': 'object'},
'AIMessageChunk': {'additionalProperties': True,
'description': '来自 AI 的消息块。',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'AIMessageChunk',
'default': 'AIMessageChunk',
'title': 'Type',
'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'example': {'default': False, 'title': 'Example', 'type': 'boolean'},
'tool_calls': {'default': [],
'items': {'$ref': '#/$defs/ToolCall'},
'title': 'Tool Calls',
'type': 'array'},
'invalid_tool_calls': {'default': [],
'items': {'$ref': '#/$defs/InvalidToolCall'},
'title': 'Invalid Tool Calls',
'type': 'array'},
'usage_metadata': {'anyOf': [{'$ref': '#/$defs/UsageMetadata'},
{'type': 'null'}],
'default': None},
'tool_call_chunks': {'default': [],
'items': {'$ref': '#/$defs/ToolCallChunk'},
'title': 'Tool Call Chunks',
'type': 'array'}},
'required': ['content'],
'title': 'AIMessageChunk',
'type': 'object'},
'ChatMessage': {'additionalProperties': True,
'description': '可以分配任意说话者（即角色）的消息。',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'chat',
'default': 'chat',
'title': 'Type',
'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'role': {'title': 'Role', 'type': 'string'}},
'required': ['content', 'role'],
'title': 'ChatMessage',
'type': 'object'},
'ChatMessageChunk': {'additionalProperties': True,
'description': '聊天消息块。',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'ChatMessageChunk',
'default': 'ChatMessageChunk',
'title': 'Type',
'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'role': {'title': 'Role', 'type': 'string'}},
'required': ['content', 'role'],
'title': 'ChatMessageChunk',
'type': 'object'},
'FunctionMessage': {'additionalProperties': True,
'description': '用于将执行工具的结果传递回模型的消息。\n\nFunctionMessage 是 ToolMessage 模式的旧版本，并且\n不包含 tool_call_id 字段。\n\ntool_call_id 字段用于将工具调用请求与\n工具调用响应关联起来。这在聊天模型能够\n并行请求多个工具调用的情况下很有用。',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'function',
'default': 'function',
'title': 'Type',
'type': 'string'},
'name': {'title': 'Name', 'type': 'string'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'}},
'required': ['content', 'name'],
'title': 'FunctionMessage',
'type': 'object'},
'FunctionMessageChunk': {'additionalProperties': True,
'description': '函数消息块。',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'FunctionMessageChunk',
'default': 'FunctionMessageChunk',
'title': 'Type',
'type': 'string'},
'name': {'title':
### 消息类型
LangChain 提供了多种消息类型，用于与大型语言模型 (LLM) 进行交互。这些消息类型定义了不同类型的输入和输出，以及它们各自的属性。
以下是 LangChain 中定义的消息类型的详细信息：
#### HumanMessage
`HumanMessage` 代表来自人类用户的消息。
```python
from langchain_core.messages import HumanMessage, SystemMessage
messages = [
SystemMessage(
content="You are a helpful assistant! Your name is Bob."
),
HumanMessage(
content="What is your name?"
)
]
## Instantiate a chat model and invoke it with the messages
model = ...
print(model.invoke(messages))
属性:
content: 消息的内容。可以是字符串或字符串列表。additional_kwargs: 额外的关键字参数。response_metadata: 响应元数据。type: 消息类型，固定为 “human”。name: 消息的发送者名称。id: 消息的 ID。example: 一个布尔值，指示该消息是否为示例。
HumanMessageChunk
HumanMessageChunk 是 HumanMessage 的一个分块版本。
属性:
content: 消息的内容。可以是字符串或字符串列表。additional_kwargs: 额外的关键字参数。response_metadata: 响应元数据。type: 消息类型，固定为 “HumanMessageChunk”。name: 消息的发送者名称。id: 消息的 ID。example: 一个布尔值，指示该消息是否为示例。
SystemMessage
SystemMessage 用于指示 AI 行为。通常作为消息序列的第一个消息传递。
from langchain_core.messages import HumanMessage, SystemMessage
messages = [
SystemMessage(
content="You are a helpful assistant! Your name is Bob."
),
HumanMessage(
content="What is your name?"
)
]
## Define a chat model and invoke it with the messages
print(model.invoke(messages))
属性:
content: 消息的内容。可以是字符串或字符串列表。additional_kwargs: 额外的关键字参数。response_metadata: 响应元数据。type: 消息类型，固定为 “system”。name: 消息的发送者名称。id: 消息的 ID。
SystemMessageChunk
SystemMessageChunk 是 SystemMessage 的一个分块版本。
属性:
content: 消息的内容。可以是字符串或字符串列表。additional_kwargs: 额外的关键字参数。response_metadata: 响应元数据。type: 消息类型，固定为 “SystemMessageChunk”。name: 消息的发送者名称。id: 消息的 ID。
ToolCall
ToolCall 代表调用工具的请求。
示例:
{
"name": "foo",
"args": {"a": 1},
"id": "123"
}
这代表请求调用名为 “foo” 的工具，参数为 {"a": 1}，标识符为 “123”。
属性:
name: 工具的名称。args: 工具的参数。id: 工具调用的 ID。type: 消息类型，固定为 “tool_call”。
ToolCallChunk
ToolCallChunk 是工具调用的一个分块 (例如，作为流的一部分)。
属性:
name: 工具的名称。args: 工具的参数。id: 工具调用的 ID。type: 消息类型，固定为 “ToolCallChunk”。
InvalidToolCall
InvalidToolCall 用于处理 LLM 产生的错误。
这里，我们在生成过程中添加一个 error 键来显示错误 (例如，无效的 JSON 参数)。
属性:
name: 工具的名称。args: 工具的参数。id: 工具调用的 ID。error: 错误信息。type: 消息类型，固定为 “invalid_tool_call”。
InputTokenDetails
输入令牌计数的细分。
不需要 累加到完整的输入令牌计数。 不需要 包含所有键。
示例:
{
"audio": 10,
"cache_creation": 200,
"cache_read": 100,
}
属性:
audio: 音频令牌的数量。cache_creation: 缓存创建的令牌数量。cache_read: 缓存读取的令牌数量。
OutputTokenDetails
输出令牌计数的细分。
不需要 累加到完整的输出令牌计数。 不需要 包含所有键。
示例:
{
"audio": 10,
"reasoning": 200,
}
属性:
audio: 音频令牌的数量。reasoning: 推理令牌的数量。
```python
{
'description': 'Output parser for the chat model.',
'definitions': {
'InputTokenDetails': {'additionalProperties': {'type': 'integer'},
'description': 'Details about the input tokens used.',
'title': 'InputTokenDetails',
'type': 'object'},
'OutputTokenDetails': {'additionalProperties': {'type': 'integer'},
'description': 'Details about the output tokens used.',
'title': 'OutputTokenDetails',
'type': 'object'},
'AIMessage': {'additionalProperties': True,
'description': 'Message from an AI.',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'ai', 'default': 'ai', 'title': 'Type', 'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'tool_call_chunks': {'items': {'$ref': '#/$defs/ToolCallChunk'},
'title': 'Tool Call Chunks',
'type': 'array'},
'usage': {'$ref': '#/$defs/UsageMetadata', 'title': 'Usage'}},
'required': ['content'],
'title': 'AIMessage',
'type': 'object'},
'AIMessageChunk': {'additionalProperties': True,
'description': 'Message from an AI, in chunks.',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'AIMessageChunk',
'default': 'AIMessageChunk',
'title': 'Type',
'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'tool_call_chunks': {'items': {'$ref': '#/$defs/ToolCallChunk'},
'title': 'Tool Call Chunks',
'type': 'array'},
'usage': {'$ref': '#/$defs/UsageMetadata', 'title': 'Usage'}},
'required': ['content'],
'title': 'AIMessageChunk',
'type': 'object'},
'ChatMessage': {'additionalProperties': True,
'description': 'Message from any actor.',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'chat', 'default': 'chat', 'title': 'Type', 'type': 'string'},
'role': {'title': 'Role', 'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'usage': {'$ref': '#/$defs/UsageMetadata', 'title': 'Usage'}},
'required': ['content', 'role'],
'title': 'ChatMessage',
'type': 'object'},
'ChatMessageChunk': {'additionalProperties': True,
'description': 'Message from any actor, in chunks.',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'ChatMessageChunk',
'default': 'ChatMessageChunk',
'title': 'Type',
'type': 'string'},
'role': {'title': 'Role', 'type': 'string'},
'name': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Name'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'usage': {'$ref': '#/$defs/UsageMetadata', 'title': 'Usage'}},
'required': ['content', 'role'],
'title': 'ChatMessageChunk',
'type': 'object'},
'FunctionMessage': {'additionalProperties': True,
'description': 'Message for passing the result of a function back to a model.',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'function',
'default': 'function',
'title': 'Type',
'type': 'string'},
'name': {'title': 'Name', 'type': 'string'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'usage': {'$ref': '#/$defs/UsageMetadata', 'title': 'Usage'}},
'required': ['content', 'name'],
'title': 'FunctionMessage',
'type': 'object'},
'FunctionMessageChunk': {'additionalProperties': True,
'description': 'Message for passing the result of a function back to a model, in chunks.',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type': 'object'},
'response_metadata': {'title': 'Response Metadata', 'type': 'object'},
'type': {'const': 'FunctionMessageChunk',
'default': 'FunctionMessageChunk',
'title': 'Type',
'type': 'string'},
'name': {'title': 'Name', 'type': 'string'},
'id': {'anyOf': [{'type': 'string'}, {'type': 'null'}],
'default': None,
'title': 'Id'},
'usage': {'$ref': '#/$defs/UsageMetadata', 'title': 'Usage'}},
'required': ['content', 'name'],
'title': 'FunctionMessageChunk',
'type': 'object'},
'HumanMessage': {'additionalProperties': True,
'description': 'Message from a human.',
'properties': {'content': {'anyOf': [{'type': 'string'},
{'items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]},
'type': 'array'}],
'title': 'Content'},
'additional_kwargs': {'title': 'Additional Kwargs', 'type':
### Runnables
Runnables 是 Langchain 中的一个核心概念，它提供了一个统一的接口来调用 LLMs、chains 和 agents。它们提供了一种灵活且可组合的方式来构建复杂的应用程序。
#### 基本用法
Runnables 的基本用法包括创建和调用它们。
```python
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.models import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
prompt = ChatPromptTemplate.from_template("Tell me a short joke about {topic}")
model = ChatOpenAI()
output_parser = StrOutputParser()
chain = prompt | model | output_parser
在这里，我们创建了一个 chain，它：
- 将一个主题作为输入，并使用 prompt template 格式化它。
- 将格式化的 prompt 传递给语言模型。
- 将模型的输出解析为字符串。
然后，我们可以使用一个输入主题来调用该 chain：
response = chain.invoke({"topic": "ice cream"})
print(response)
输入和输出 Schema
Runnables 支持输入和输出 schema，它们定义了预期的输入和输出格式。这允许更好的类型检查和错误处理。
from langchain_core.pydantic_v1 import BaseModel, Field
class MovieReview(BaseModel):
movie_title: str = Field(description="The title of the movie")
review: str = Field(description="The review of the movie")
language: str = Field(description="The language of the review")
我们可以使用这些 schema 来定义 Runnables 的输入和输出格式。
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.models import ChatOpenAI
from langchain_core.output_parsers import JsonOutputParser
prompt_template = ChatPromptTemplate.from_messages(
[
("system", "You are a helpful assistant that translates {input_language} to {output_language}."),
("human", "{input}"),
MessagesPlaceholder(variable_name="history"),
]
)
model = ChatOpenAI()
parser = JsonOutputParser(pydantic_schema=MovieReview)
然后，我们可以访问 Runnables 的输入和输出 schema。
prompt_template.input_schema.model_json_schema()
"""
{'properties': {'input_language': {'title': 'Input Language', 'type': 'string'},
'output_language': {'title': 'Output Language', 'type': 'string'},
'input': {'title': 'Input', 'type': 'string'},
'history': {'items': {'$ref': '#/definitions/BaseMessage'},
'title': 'History',
'type': 'array'}},
'required': ['input_language', 'output_language', 'input'],
'title': 'PromptInput',
'type': 'object',
'definitions': {'BaseMessage': {'properties': {'type': {'title': 'Type',
'type': 'string'},
'content': {'title': 'Content', 'type': 'string'}},
'required': ['type', 'content'],
'title': 'BaseMessage',
'type': 'object'}}}
"""
model.input_schema.model_json_schema()
"""
{'properties': {'messages': {'items': {'$ref': '#/definitions/BaseMessage'},
'title': 'Messages',
'type': 'array'},
'functions': {'items': {'$ref': '#/definitions/FunctionSchema'},
'title': 'Functions',
'type': 'array'},
'function_call': {'anyOf': [{'type': 'string'},
{'type': 'object',
'properties': {'name': {'title': 'Name', 'type': 'string'}},
'required': ['name']}],
'title': 'Function Call'}},
'required': ['messages'],
'title': 'ChatInput',
'type': 'object',
'definitions': {'BaseMessage': {'properties': {'type': {'title': 'Type',
'type': 'string'},
'content': {'title': 'Content', 'type': 'string'}},
'required': ['type', 'content'],
'title': 'BaseMessage',
'type': 'object'},
'FunctionSchema': {'properties': {'name': {'title': 'Name', 'type': 'string'},
'description': {'title': 'Description', 'type': 'string'},
'parameters': {'title': 'Parameters', 'type': 'object'}},
'required': ['name', 'parameters'],
'title': 'FunctionSchema',
'type': 'object'}}}
"""
parser.output_schema.model_json_schema()
"""
{'properties': {'movie_title': {'description': 'The title of the movie',
'title': 'Movie Title',
'type': 'string'},
'review': {'description': 'The review of the movie',
'title': 'Review',
'type': 'string'},
'language': {'description': 'The language of the review',
'title': 'Language',
'type': 'string'}},
'required': ['movie_title', 'review', 'language'],
'title': 'MovieReview',
'type': 'object'}
"""
最后，我们可以构建我们的 chain。我们使用 | 管道运算符来链接每个组件。
chain = prompt_template | model | parser
chain.input_schema.model_json_schema()
"""
{'properties': {'format_instructions': {'title': 'Format Instructions',
'type': 'string'},
'movie': {'title': 'Movie', 'type': 'string'}},
'required': ['format_instructions', 'movie'],
'title': 'PromptInput',
'type': 'object'}
"""
chain.output_schema.model_json_schema()
"""
{'properties': {'movie_title': {'description': 'The title of the movie',
'title': 'Movie Title',
'type': 'string'},
'review': {'description': 'The review of the movie',
'title': 'Review',
'type': 'string'},
'language': {'description': 'The language of the review',
'title': 'Language',
'type': 'string'}},
'required': ['movie_title', 'review', 'language'],
'title': 'MovieReview',
'type': 'object'}
"""
我们使用 Runnable 的 invoke 方法并获取响应。
使用提示调用链
result = runnable.invoke('hello')
print(result)
### Output: {'llm1': 'completion', 'llm2': 'completion', 'total_chars': 20}
fake_llm 函数模拟一个 LLM。我们使用 | 运算符将字典链接到 RunnablePassthrough.assign()。
RunnablePassthrough.assign() 是一种方法，它接受关键字参数，其中每个键是要添加的新字段，其值是一个函数，该函数根据输入计算该字段的值。
在这种情况下，它向输出添加一个新键 'total_chars'。该值由一个 lambda 函数计算，该函数：
- 接受
inputs（一个包含'llm1'和'llm2'的输出的字典）。 - 连接来自
inputs['llm1']和inputs['llm2']的字符串。 - 使用
len()计算连接字符串的长度。
result = runnable.invoke('hello')
- 使用输入
'hello'调用invoke方法。 - 此输入传递给
'llm1'和'llm2'函数（它们都是fake_llm）。 - 每个
fake_llm函数返回'completion'。 - 输出被收集到一个字典中：
{'llm1': 'completion', 'llm2': 'completion'}。 - 在获得初始输出后，
RunnablePassthrough.assign()添加'total_chars'键。
让我们创建一个 RAG 示例。
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
template = """仅根据以下上下文回答问题：
{context}
问题：{question}
"""
prompt = ChatPromptTemplate.from_template(template)
llm = ChatOpenAI(model="gpt-4o-mini")
vectorstore = FAISS.from_texts(
["harrison worked at kensho"], embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()
retrieval_chain = (
{
"context": retriever.with_config(run_name="Docs"),
"question": RunnablePassthrough(),
}
| prompt
| llm.with_config(run_name="my_llm")
| StrOutputParser()
)
response = retrieval_chain.invoke("Where did harrison work?")
print(response)
"""
'Harrison worked at Kensho.'
"""
让我们仔细看看链部分，在这里我们组合了 Runnables。
retrieval_chain = (
{
"context": retriever.with_config(run_name="Docs"),
"question": RunnablePassthrough(),
}
| prompt
| llm.with_config(run_name="my_llm")
| StrOutputParser()
)
response = retrieval_chain.invoke("Where did harrison work?")
此链旨在通过检索相关上下文并生成适当的答案来处理用户的问题。
首先，可运行的是带有上下文和问题的字典。
{
"context": retriever.with_config(run_name="Docs"),
"question": RunnablePassthrough(),
}
这是一个字典，它为链构建了初始输入结构。它是一个 RunnableParallel 结构，这意味着它并行处理其键（上下文和问题），并生成一个包含这些键的字典输出。
context：此键与retriever对象关联，该对象配置了"Docs"的run_name。retriever负责根据输入问题获取相关文档或上下文。question：此键使用RunnablePassthrough()，它只是将输入（用户的问题）不变地传递到下一步。
用户问题
|
v
+-----------------+
| Retriever | --> 从向量存储中检索上下文
+-----------------+
|
v
+-----------------+
| RunnablePassthrough | --> 不变地传递问题
+-----------------+
|
v
+-----------------+
| 提示模板 | --> 将上下文和问题格式化为提示
+-----------------+
|
v
+-----------------+
| LLM (GPT-4) | --> 根据提示生成响应
+-----------------+
|
v
+-----------------+
| StrOutputParser | --> 将 LLM 的输出解析为字符串
+-----------------+
|
v
最终答案："Harrison worked at Kensho."
另一个例子：
from operator import itemgetter
template = """仅根据以下上下文回答问题：
{context}
问题：{question}
像一个 {character_style} 角色一样回答。
"""
prompt = ChatPromptTemplate.from_template(template)
chain = (
{
"context": itemgetter("question") | retriever,
"question": itemgetter("question"),
"character_style": itemgetter("character_style"),
}
| prompt
| llm
| StrOutputParser()
)
response = chain.invoke({"question": "Where did harrison work?", "character_style": "funny"})
print(response)
"""
好吧，好吧，好吧！如果这不是我们的朋友哈里森，数据方面的大师！我的朋友，他正在 Kensho 炫耀他的东西！没错！Kensho！听起来很花哨，不是吗？我敢打赌，他正在那里处理数字，像计算器商店里的巫师一样创造奇迹！✨🧙♂️
"""
首先，我们有一个输入字典。此字典中的每个键都对应于我们的提示模板所需的字段。
{
"context": itemgetter("question") | retriever,
"question": itemgetter("question"),
"character_style": itemgetter("character_style"),
}
itemgetter 是 Python operator 模块中的一个实用程序，它创建一个可调用对象，用于从集合（如字典）中提取特定项目。在这种情况下，itemgetter("question") 返回一个函数，该函数在给定一个字典时，检索与键 "question" 关联的值。
因此，首先，我们从输入字典中提取用户的问题，然后，| 运算符将提取的问题传递到 retriever 中。
retriever 在向量存储中搜索与问题相关的文档。
检索到的文档被分配到输出字典中的 context 键。
RunnableSequence
RunnableSequence 允许我们将多个 runnables 链接在一起，其中一个 runnable 的输出用作下一个的输入。
它可以通过使用 RunnableSequence 或（更常见的是）使用管道 | 运算符来创建。
实际上，这就是我们在本博客的演示中所做的事情。
RunnableParallel
RunnableParallel 是一个组合原语，它允许我们并发地执行多个 runnable，为每个 runnable 提供相同的输入，并返回它们输出的映射。当有几个可以同时执行的独立任务时，这特别有用，从而提高了效率并减少了总处理时间。
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
runnable = RunnableParallel(
passed=RunnablePassthrough(),
extra=RunnablePassthrough.assign(muliplied=lambda x: x["num"] * 2),
modified=lambda x: x["num"] + 1
)
runnable.invoke({"num": 1})
"""
{'passed': {'num': 1}, 'extra': {'num': 1, 'muliplied': 2}, 'modified': 2}
"""
我们创建了三个并行分支：
passed：使用RunnablePassthrough简单地原样传递输入。extra：使用RunnablePassthrough.assign()向输入添加一个新字段multiplied。modified：使用 lambda 函数仅转换num值。
输入被并发地馈送到 RunnableParallel 中的每个组件（passed、extra 和 `modified）。
RunnableParallel 收集所有组件的输出并将它们组装成一个字典：
'passed'：{"num": 1}'extra'：{"num": 1, "multiplied": 2}'modified'：2
{
'passed': {'num': 1},
'extra': {'num': 1, 'multiplied': 2},
'modified': 2
}
上面使用 character_style 的 RAG 示例实际上是一个 RunnableParallel。
{
"context": itemgetter("question") | retriever,
"question": itemgetter("question"),
"character_style": itemgetter("character_style"),
}
另一个例子：
joke_chain = ChatPromptTemplate.from_template("tell me a joke about {topic}") | llm
poem_chain = (
ChatPromptTemplate.from_template("write a 2-line poem about {topic}") | llm
)
map_chain = RunnableParallel(joke=joke_chain, poem=poem_chain)
map_chain.invoke({"topic": "marriage"})
RunnableLambda
RunnableLambda 是一种实用程序，它允许我们将一个 Python 函数（或任何可调用对象）封装到一个 runnable 组件中。
from langchain_core.runnables import RunnableLambda
### Step 1: Define a custom function
def add_greeting(input_text):
return f"Hello, {input_text}!"
### Step 2: Wrap the function in RunnableLambda
greet_runnable = RunnableLambda(add_greeting)
### Step 3: Use the RunnableLambda in a pipeline
result = greet_runnable.invoke("world")
print(result) # Output: "Hello, world!"
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
### Step 1: Define custom functions
def add_greeting(input_text):
return f"Hello, {input_text}!"
def to_uppercase(input_text):
return input_text.upper()
### Step 2: Wrap the functions in RunnableLambda
greet_runnable = RunnableLambda(add_greeting)
uppercase_runnable = RunnableLambda(to_uppercase)
### Step 3: Create a pipeline
pipeline = (
RunnablePassthrough() # Passes input through unchanged
| greet_runnable # Adds a greeting
| uppercase_runnable # Converts to uppercase
)
### Step 4: Invoke the pipeline
result = pipeline.invoke("world")
print(result) # Output: "HELLO, WORLD!"
Input: "world"
|
v
+-----------------+
| RunnablePassthrough | --> Output: "world"
+-----------------+
|
v
+-----------------+
| greet_runnable | --> Output: "Hello, world!"
+-----------------+
|
v
+-----------------+
| uppercase_runnable | --> Output: "HELLO, WORLD!"
+-----------------+
|
v
Final Output: "HELLO, WORLD!"
封装在 RunnableLambda 中的函数只能接受一个参数。如果需要多个参数，请使用字典。
RunnableConfig
简单来说，它是 runnable 的配置。它可以包括元数据，如标签或回调（在调用 LLM 链时使用）。
在此示例中，我们定义了一个函数 parse_or_fix，它尝试解析 JSON 字符串。如果解析失败，它会利用一个重试机制，该机制涉及一个 LLM 来更正 JSON 格式。
import json
from langchain_core.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnableConfig
from langchain.callbacks import get_openai_callback
def parse_or_fix(text: str, config: RunnableConfig):
fixing_chain = (
ChatPromptTemplate.from_template(
"Fix the following text:\n\n```text\n{input}\n```\nError: {error}"
" Don't narrate, just respond with the fixed data."
)
| ChatOpenAI()
| StrOutputParser()
)
for _ in range(3):
try:
return json.loads(text)
except Exception as e:
text = fixing_chain.invoke({"input": text, "error": e}, config)
return "Failed to parse"
with get_openai_callback() as cb:
output = RunnableLambda(parse_or_fix).invoke(
"{foo: bar}", {"tags": ["my-tag"], "callbacks": [cb]}
)
print(output)
print(cb)
"""
{'foo': 'bar'}
Tokens Used: 63
Prompt Tokens: 56
Prompt Tokens Cached: 0
Completion Tokens: 7
Reasoning Tokens: 0
Successful Requests: 1
Total Cost (USD): $3.85e-0
"""
该函数最多尝试解析 JSON 字符串三次。如果解析失败，则调用 fixing_chain 来更正文本。然后，更正后的文本将在下一次迭代中使用。
fixing_chain 是一个 RunnableSequence（通过 | 运算符）。
get_openai_callback() 创建一个上下文管理器，该管理器通过回调对象 cb 跟踪 OpenAI API 使用情况（令牌、成本、请求）。
RunnableLambda(parse_or_fix).invoke 将 parse_or_fix 函数封装在 RunnableLambda 中，并使用格式错误的 JSON 字符串和包含标签和回调的配置来调用它。
RunnableBranch
RunnableBranch 允许我们在工作流程中定义多个“分支”。 每个分支都与一个特定条件相关联。
from langchain_core.runnables import RunnableBranch
branch = RunnableBranch(
(lambda x: isinstance(x, str), lambda x: x.upper()),
(lambda x: isinstance(x, int), lambda x: x + 1),
(lambda x: isinstance(x, float), lambda x: x * 2),
lambda x: "goodbye",
)
branch.invoke("hello") # "HELLO"
branch.invoke(None) # "goodbye"
另一个例子：此链将用户的问题分类为预定义的类别之一：Technology、Health 或 Other。
classification_chain = (
PromptTemplate.from_template(
"""将以下问题分类为以下类别之一：`Technology`、`Health` 或 `Other`。
问题：{question}
类别："""
)
| ChatOpenAI()
| StrOutputParser()
)
technology_chain 回答与技术相关的问题。
technology_chain = (
PromptTemplate.from_template(
"""你是一位技术专家。 始终以“作为技术大师，我可以告诉你”开头你的回答。 回答以下问题：
问题：{question}
答案："""
)
| ChatOpenAI()
)
health_chain 回答与健康相关的问题。
health_chain = (
PromptTemplate.from_template(
"""你是一位健康专家。 始终以“作为健康专家，我建议”开头你的回答。 回答以下问题：
问题：{question}
答案："""
)
| ChatOpenAI()
)
general_chain 处理不属于指定类别的问题。
general_chain = (
PromptTemplate.from_template(
"""我很乐意帮助你解决问题：
问题：{question}
答案："""
)
| ChatOpenAI()
)
RunnableBranch 根据分类结果将输入定向到适当的链。
branch = RunnableBranch(
(lambda x: "technology" in x["category"].lower(), technology_chain),
(lambda x: "health" in x["category"].lower(), health_chain),
general_chain, # 如果没有满足的条件，则为默认链
)
此管道首先对问题进行分类，然后将其路由到相应的响应链。
full_chain = {"category": classification_chain,
"question": lambda x: x["question"]} | branch
response = full_chain.invoke({"question": "What are the latest advancements in artificial intelligence?"})
print(response)
"""
content="As a tech guru, I can tell you that the latest advancements in artificial intelligence include the development of more advanced deep learning algorithms, the integration of AI into various industries such as healthcare and finance, the emergence of autonomous systems, and the progress in natural language processing and computer vision technologies. Additionally, researchers are also focusing on enhancing AI's ability to understand and interpret human emotions, which opens up new possibilities for AI applications in mental health and social interactions." additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 91, 'prompt_tokens': 51, 'total_tokens': 142, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-24ab7153-c697-41c0-a5c0-c1a03cf5522a-0' usage_metadata={'input_tokens': 51, 'output_tokens': 91, 'total_tokens': 142, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}
"""
自定义 Runnable
我们可以通过实现 Runnable 类来创建自定义的 runnable。 让我们创建一个 Runnable，它在失败时多次重试任务，并可以选择延迟。
import time
import random
from langchain_core.runnables import Runnable
from langchain_core.runnables.utils import Input, Output
from typing import Any, Optional
class TaskRetrier(Runnable):
"""一个 Runnable，它在失败时多次重试任务，并可以选择延迟。"""
def __init__(self, task: Runnable[Input, Output], max_attempts: int, delay_seconds: float = 0.0):
"""
初始化 TaskRetrier。
参数：
task: 要执行具有重试逻辑的 Runnable。
max_attempts: 尝试任务的最大次数。
delay_seconds: 重试之间的延迟（以秒为单位）（默认值：0.0）。
"""
self.task = task
self.max_attempts = max_attempts
self.delay_seconds = delay_seconds
def invoke(self, input_data: Input, config: Optional[Any] = None) -> Output:
"""
执行具有重试逻辑的任务。
参数：
input_data: 传递给任务的输入。
config: 任务的可选配置（例如，回调）。
返回：
如果成功，则返回任务的输出。
引发：
Exception: 如果所有尝试都失败，则引发遇到的最后一个异常。
"""
attempt_count = 0
while attempt_count < self.max_attempts:
try:
return self.task.invoke(input_data, config)
except Exception as error:
attempt_count += 1
print(f"重试 {attempt_count}/{self.max_attempts} 失败：{error}")
if attempt_count >= self.max_attempts:
raise error
if self.delay_seconds > 0:
time.sleep(self.delay_seconds)
def occasionally_fails(text: str) -> str:
"""
一个将输入文本大写的函数，但有一半的时间会随机失败。
参数：
text: 要处理的输入字符串。
返回：
如果成功，则返回文本的大写版本。
引发：
ValueError: 随机引发以模拟失败。
"""
if random.random() < 0.5:
raise ValueError("模拟处理错误")
return text.upper()
### 创建一个启用重试的 Runnable
retry_handler = TaskRetrier(
task=RunnableLambda(occasionally_fails),
max_attempts=3,
delay_seconds=1.0,
)
### 测试重试逻辑
try:
result = retry_handler.invoke("hello")
print(f"成功：{result}")
except Exception as final_error:
print(f"所有重试已用尽：{final_error}")
"""
Retry 1/3 failed: Simulated processing error
Success: HELLO
"""
我们可以使用 Runnable 的 with_retry 方法实现相同的逻辑！
import time
import random
from langchain_core.runnables import RunnableLambda
### 定义一个模拟偶尔失败的函数
def unreliable_function(input_value: str) -> str:
if random.random() < 0.5:
raise ValueError("发生随机故障")
return input_value.upper()
### 将该函数包装在 RunnableLambda 中
runnable = RunnableLambda(unreliable_function)
为可运行对象添加重试逻辑
runnable_with_retries = runnable.with_retry(
retry_if_exception_type=(ValueError,), # Retry only on ValueError
wait_exponential_jitter=True, # Add jitter to the exponential backoff
stop_after_attempt=3 # Retry up to 3 times
)
### Invoke the runnable with retries
try:
result = runnable_with_retries.invoke("hello")
print(f"Success: {result}")
except Exception as e:
print(f"Operation failed after retries: {e}")
总结
LangChain 中的可运行对象是模块化的、可执行的构建块，允许您通过链式、分支或并行运行任务来创建动态的 AI 工作流程。它们对于创建灵活、可重用的管道至关重要，这些管道通过自定义逻辑、LLM 或检索器等工具处理数据。
从本质上讲，可运行对象标准化了输入到输出的流程，使其易于适应各种场景——想想聊天机器人、数据解析或错误处理。在任何需要对复杂任务进行可组合、逐步控制的地方使用它们。
最重要的是，请记住这一点：可运行对象是您构建智能、自适应系统的乐高积木——掌握它们，您将解锁无限的可能性。
了解更多
[## LangChain in Chains #45: Web Scraping with LLMs
Integrating LLMs into Web Scraping Workflows Using LangChain
blog.devgenius.io](https://blog.devgenius.io/langchain-in-chains-45-web-scraping-with-llms-8c4192bde030)
[## Mastering CrewAI: Chapter 1 — Your First Intelligent Workflow
The Basics of Agents and Tasks
ai.plainenglish.io](https://ai.plainenglish.io/mastering-crewai-chapter-1-your-first-intelligent-workflow-06b58aed1b3d)
[## Inside AutoGen: Chapter 1 –Introducing AgentChat
An Overview of AgentChat’s Features
pub.aimind.so](https://pub.aimind.so/inside-autogen-chapter-1-introducing-agentchat-1ffbe4489bfe)
[## MobileNet and Depthwise Separable Convolutions: Faster, Smaller, Smarter
MobileNet Explained: The Secret to Efficient Deep Learning on Mobile Devices
medium.com](https://readmedium.com/mobilenet-and-depthwise-separable-convolutions-faster-smaller-smarter-798f5a3920d3)
来源
https://python.langchain.com/v0.1/docs/expression_language/interface/
https://python.langchain.com/docs/concepts/runnables/
https://python.langchain.com/v0.1/docs/expression_language/primitives/passthrough/
https://www.youtube.com/watch?v=moJRxxEddzU
来自 AI Mind 的消息
