langchain-azure-ai¶
Reference docs
This page contains reference documentation for Azure AI. See the docs for conceptual guides, tutorials, and examples on using Azure AI.
langchain_azure_ai.agents
¶
Agents integrated with LangChain and LangGraph.
AgentServiceFactory
¶
Bases: BaseModel
Factory to create and manage prompt-based agents in Azure AI Foundry.
To create a simple echo agent:
from langchain_azure_ai.agents import AgentServiceFactory
from langchain_core.messages import HumanMessage
from azure.identity import DefaultAzureCredential
factory = AgentServiceFactory(
project_endpoint=(
"https://resource.services.ai.azure.com/api/projects/demo-project",
),
credential=DefaultAzureCredential()
)
agent = factory.create_prompt_agent(
name="my-echo-agent",
model="gpt-4.1",
instructions="You are a helpful AI assistant that always replies back
"saying the opposite of what the user says.",
)
messages = [HumanMessage(content="I'm a genius and I love programming!")]
state = agent.invoke({"messages": messages})
for m in state['messages']:
m.pretty_print()
Note
You can also create AgentServiceFactory without passing any parameters
if you have set the AZURE_AI_PROJECT_ENDPOINT environment variable and
are using DefaultAzureCredential for authentication.
Agents can also be created with tools. For example, to create an agent that can perform arithmetic using a calculator tool:
# add, multiply, divide are simple functions defined elsewhere
# those functions are documented and with proper type hints
tools = [add, multiply, divide]
agent = factory.create_prompt_agent(
name="math-agent",
model="gpt-4.1",
instructions="You are a helpful assistant tasked with performing "
"arithmetic on a set of inputs.",
tools=tools,
)
You can also use the built-in tools in the Agent Service. Those tools only work with agents created in Azure AI Foundry. For example, to create an agent that can use Code Interpreter.
from langchain_azure_ai.tools.agent_service import CodeInterpreterTool
document_parser_agent = factory.create_prompt_agent(
name="code-interpreter-agent",
model="gpt-4.1",
instructions="You are a helpful assistant that can run complex "
"mathematical functions precisely via tools.",
tools=[CodeInterpreterTool()],
)
| METHOD | DESCRIPTION |
|---|---|
validate_environment |
Validate that required values are present in the environment. |
delete_agent |
Delete an agent created with create_prompt_agent. |
get_agents_id_from_graph |
Get the Azure AI Foundry agent associated with a state graph. |
create_prompt_agent_node |
Create a prompt-based agent node in Azure AI Foundry. |
create_prompt_agent |
Create a prompt-based agent in Azure AI Foundry. |
project_endpoint
class-attribute
instance-attribute
¶
project_endpoint: str | None = None
The project endpoint associated with the AI project. If this is specified,
then the endpoint parameter becomes optional and credential has to be of type
TokenCredential.
credential
class-attribute
instance-attribute
¶
The API key or credential to use to connect to the service. If using a project
endpoint, this must be of type TokenCredential since only Microsoft EntraID is
supported.
api_version
class-attribute
instance-attribute
¶
api_version: str | None = None
The API version to use with Azure. If None, the default version is used.
client_kwargs
class-attribute
instance-attribute
¶
Additional keyword arguments to pass to the client.
validate_environment
¶
Validate that required values are present in the environment.
delete_agent
¶
delete_agent(agent: CompiledStateGraph | PromptBasedAgentNode) -> None
Delete an agent created with create_prompt_agent.
| PARAMETER | DESCRIPTION |
|---|---|
agent
|
The CompiledStateGraph representing the agent to delete.
TYPE:
|
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If the agent ID cannot be found in the graph metadata. |
get_agents_id_from_graph
¶
get_agents_id_from_graph(graph: CompiledStateGraph) -> set[str]
Get the Azure AI Foundry agent associated with a state graph.
create_prompt_agent_node
¶
create_prompt_agent_node(
name: str,
model: str,
description: str | None = None,
tools: Sequence[AgentServiceBaseTool | BaseTool | Callable]
| ToolNode
| None = None,
instructions: Prompt | None = None,
temperature: float | None = None,
top_p: float | None = None,
response_format: dict[str, Any] | None = None,
trace: bool = False,
) -> PromptBasedAgentNode
Create a prompt-based agent node in Azure AI Foundry.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the agent.
TYPE:
|
model
|
The model to use for the agent.
TYPE:
|
description
|
An optional description of the agent.
TYPE:
|
tools
|
The tools to use with the agent. This can be a list of BaseTools callables, or tool definitions, or a ToolNode.
TYPE:
|
instructions
|
The prompt instructions to use for the agent.
TYPE:
|
temperature
|
The temperature to use for the agent.
TYPE:
|
top_p
|
The top_p to use for the agent.
TYPE:
|
response_format
|
The response format to use for the agent. |
trace
|
Whether to enable tracing.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
PromptBasedAgentNode
|
A DeclarativeChatAgentNode representing the agent. |
create_prompt_agent
¶
create_prompt_agent(
model: str,
name: str,
description: str | None = None,
tools: Sequence[AgentServiceBaseTool | BaseTool | Callable]
| ToolNode
| None = None,
instructions: Prompt | None = None,
temperature: float | None = None,
top_p: float | None = None,
response_format: dict[str, Any] | None = None,
state_schema: StateSchemaType | None = None,
context_schema: Type[Any] | None = None,
checkpointer: Checkpointer | None = None,
store: BaseStore | None = None,
interrupt_before: list[str] | None = None,
interrupt_after: list[str] | None = None,
trace: bool = False,
debug: bool = False,
) -> CompiledStateGraph
Create a prompt-based agent in Azure AI Foundry.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the agent.
TYPE:
|
description
|
An optional description of the agent.
TYPE:
|
model
|
The model to use for the agent.
TYPE:
|
tools
|
The tools to use with the agent. This can be a list of BaseTools, callables, or tool definitions, or a ToolNode.
TYPE:
|
instructions
|
The prompt instructions to use for the agent.
TYPE:
|
temperature
|
The temperature to use for the agent.
TYPE:
|
top_p
|
The top_p to use for the agent.
TYPE:
|
response_format
|
The response format to use for the agent. |
state_schema
|
The schema for the state to pass to the agent. If None, AgentStateWithStructuredResponse is used if response_format is specified, otherwise AgentState is used.
TYPE:
|
context_schema
|
The schema for the context to pass to the agent. |
checkpointer
|
The checkpointer to use for the agent.
TYPE:
|
store
|
The store to use for the agent.
TYPE:
|
interrupt_before
|
A list of node names to interrupt before. |
interrupt_after
|
A list of node names to interrupt after. |
trace
|
Whether to enable tracing. When enabled, an OpenTelemetry tracer will be created using the project endpoint and credential provided to the factory.
TYPE:
|
debug
|
Whether to enable debug mode.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
CompiledStateGraph
|
A CompiledStateGraph representing the agent workflow. |
langchain_azure_ai.callbacks.tracers
¶
Tracing capabilities for Azure AI Foundry.
AzureAIOpenTelemetryTracer
¶
Bases: BaseCallbackHandler
LangChain callback handler that emits OpenTelemetry GenAI spans.
| METHOD | DESCRIPTION |
|---|---|
on_text |
Run on an arbitrary text. |
on_retry |
Run on a retry event. |
on_custom_event |
Override to define a handler for a custom event. |
on_llm_new_token |
Run on new output token. Only available when streaming is enabled. |
__init__ |
Initialize tracer state and configure Azure Monitor if needed. |
on_chain_start |
Handle start of a chain/agent invocation. |
on_chain_end |
Handle completion of a chain/agent invocation. |
on_chain_error |
Handle errors raised during chain execution. |
on_chat_model_start |
Record chat model start metadata. |
on_llm_start |
Record LLM start metadata. |
on_llm_end |
Record LLM response attributes and finish the span. |
on_llm_error |
Mark the LLM span as errored. |
on_tool_start |
Create a span representing tool execution. |
on_tool_end |
Finalize a tool span with results. |
on_tool_error |
Mark a tool span as errored. |
on_agent_action |
Cache tool context emitted from agent actions. |
on_agent_finish |
Close an agent span and record outputs. |
on_retriever_start |
Start a retriever span. |
on_retriever_end |
Record retriever results and close the span. |
on_retriever_error |
Mark a retriever span as errored. |
raise_error
class-attribute
instance-attribute
¶
raise_error: bool = False
Whether to raise an error if an exception occurs.
run_inline
class-attribute
instance-attribute
¶
run_inline: bool = False
Whether to run the callback inline.
on_text
¶
on_retry
¶
on_retry(
retry_state: RetryCallState,
*,
run_id: UUID,
parent_run_id: UUID | None = None,
**kwargs: Any,
) -> Any
Run on a retry event.
| PARAMETER | DESCRIPTION |
|---|---|
retry_state
|
The retry state.
TYPE:
|
run_id
|
The run ID. This is the ID of the current run.
TYPE:
|
parent_run_id
|
The parent run ID. This is the ID of the parent run.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
on_custom_event
¶
on_custom_event(
name: str,
data: Any,
*,
run_id: UUID,
tags: list[str] | None = None,
metadata: dict[str, Any] | None = None,
**kwargs: Any,
) -> Any
Override to define a handler for a custom event.
| PARAMETER | DESCRIPTION |
|---|---|
name
|
The name of the custom event.
TYPE:
|
data
|
The data for the custom event. Format will match the format specified by the user.
TYPE:
|
run_id
|
The ID of the run.
TYPE:
|
tags
|
The tags associated with the custom event (includes inherited tags). |
metadata
|
The metadata associated with the custom event (includes inherited metadata). |
on_llm_new_token
¶
on_llm_new_token(
token: str,
*,
chunk: GenerationChunk | ChatGenerationChunk | None = None,
run_id: UUID,
parent_run_id: UUID | None = None,
**kwargs: Any,
) -> Any
Run on new output token. Only available when streaming is enabled.
For both chat models and non-chat models (legacy LLMs).
| PARAMETER | DESCRIPTION |
|---|---|
token
|
The new token.
TYPE:
|
chunk
|
The new generated chunk, containing content and other information.
TYPE:
|
run_id
|
The run ID. This is the ID of the current run.
TYPE:
|
parent_run_id
|
The parent run ID. This is the ID of the parent run.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
__init__
¶
__init__(
*,
connection_string: str | None = None,
enable_content_recording: bool = True,
project_endpoint: str | None = None,
credential: Any | None = None,
name: str = "AzureAIOpenTelemetryTracer",
agent_id: str | None = None,
provider_name: str | None = None,
) -> None
Initialize tracer state and configure Azure Monitor if needed.
on_chain_start
¶
on_chain_start(
serialized: dict[str, Any],
inputs: dict[str, Any],
*,
run_id: UUID,
parent_run_id: UUID | None = None,
tags: list[str] | None = None,
metadata: dict[str, Any] | None = None,
**kwargs: Any,
) -> Any
Handle start of a chain/agent invocation.
on_chain_end
¶
on_chain_end(
outputs: dict[str, Any],
*,
run_id: UUID,
parent_run_id: UUID | None = None,
**kwargs: Any,
) -> Any
Handle completion of a chain/agent invocation.
on_chain_error
¶
on_chain_error(
error: BaseException,
*,
run_id: UUID,
parent_run_id: UUID | None = None,
**kwargs: Any,
) -> Any
Handle errors raised during chain execution.
on_chat_model_start
¶
on_chat_model_start(
serialized: dict[str, Any],
messages: list[list[BaseMessage]],
*,
run_id: UUID,
parent_run_id: UUID | None = None,
tags: list[str] | None = None,
metadata: dict[str, Any] | None = None,
**kwargs: Any,
) -> Any
Record chat model start metadata.
on_llm_start
¶
on_llm_start(
serialized: dict[str, Any],
prompts: list[str],
*,
run_id: UUID,
parent_run_id: UUID | None = None,
tags: list[str] | None = None,
metadata: dict[str, Any] | None = None,
**kwargs: Any,
) -> Any
Record LLM start metadata.
on_llm_end
¶
on_llm_end(
response: LLMResult,
*,
run_id: UUID,
parent_run_id: UUID | None = None,
**kwargs: Any,
) -> Any
Record LLM response attributes and finish the span.
on_llm_error
¶
on_llm_error(
error: BaseException,
*,
run_id: UUID,
parent_run_id: UUID | None = None,
**kwargs: Any,
) -> Any
Mark the LLM span as errored.
on_tool_start
¶
on_tool_start(
serialized: dict[str, Any],
input_str: str,
*,
run_id: UUID,
parent_run_id: UUID | None = None,
metadata: dict[str, Any] | None = None,
inputs: dict[str, Any] | None = None,
**kwargs: Any,
) -> Any
Create a span representing tool execution.
on_tool_end
¶
on_tool_end(
output: Any, *, run_id: UUID, parent_run_id: UUID | None = None, **kwargs: Any
) -> Any
Finalize a tool span with results.
on_tool_error
¶
on_tool_error(
error: BaseException,
*,
run_id: UUID,
parent_run_id: UUID | None = None,
**kwargs: Any,
) -> Any
Mark a tool span as errored.
on_agent_action
¶
on_agent_action(
action: AgentAction,
*,
run_id: UUID,
parent_run_id: UUID | None = None,
**kwargs: Any,
) -> Any
Cache tool context emitted from agent actions.
on_agent_finish
¶
on_agent_finish(
finish: AgentFinish,
*,
run_id: UUID,
parent_run_id: UUID | None = None,
**kwargs: Any,
) -> Any
Close an agent span and record outputs.
on_retriever_start
¶
on_retriever_start(
serialized: dict[str, Any],
query: str,
*,
run_id: UUID,
parent_run_id: UUID | None = None,
metadata: dict[str, Any] | None = None,
**kwargs: Any,
) -> Any
Start a retriever span.
langchain_azure_ai.chat_message_histories
¶
Chat message history stores a history of the message interactions in a chat.
Class hierarchy:
Main helpers:
CosmosDBChatMessageHistory
¶
Bases: BaseChatMessageHistory
Chat message history backed by Azure CosmosDB.
| METHOD | DESCRIPTION |
|---|---|
aget_messages |
Async version of getting messages. |
add_user_message |
Convenience method for adding a human message string to the store. |
add_ai_message |
Convenience method for adding an |
add_messages |
Add a list of messages. |
aadd_messages |
Async add a list of messages. |
aclear |
Async remove all messages from the store. |
__str__ |
Return a string representation of the chat history. |
__init__ |
Initializes a new instance of the CosmosDBChatMessageHistory class. |
prepare_cosmos |
Prepare the CosmosDB client. |
__enter__ |
Context manager entry point. |
__exit__ |
Context manager exit. |
load_messages |
Retrieve the messages from Cosmos. |
add_message |
Add a self-created message to the store. |
upsert_messages |
Update the cosmosdb item. |
clear |
Clear session memory from this memory and cosmos. |
messages
instance-attribute
¶
messages: list[BaseMessage] = []
A property or attribute that returns a list of messages.
In general, getting the messages may involve IO to the underlying persistence layer, so this operation is expected to incur some latency.
aget_messages
async
¶
aget_messages() -> list[BaseMessage]
Async version of getting messages.
Can over-ride this method to provide an efficient async implementation.
In general, fetching messages may involve IO to the underlying persistence layer.
| RETURNS | DESCRIPTION |
|---|---|
list[BaseMessage]
|
The messages. |
add_user_message
¶
add_user_message(message: HumanMessage | str) -> None
Convenience method for adding a human message string to the store.
Note
This is a convenience method. Code should favor the bulk add_messages
interface instead to save on round-trips to the persistence layer.
This method may be deprecated in a future release.
| PARAMETER | DESCRIPTION |
|---|---|
message
|
The
TYPE:
|
add_ai_message
¶
Convenience method for adding an AIMessage string to the store.
Note
This is a convenience method. Code should favor the bulk add_messages
interface instead to save on round-trips to the persistence layer.
This method may be deprecated in a future release.
| PARAMETER | DESCRIPTION |
|---|---|
message
|
The |
add_messages
¶
add_messages(messages: Sequence[BaseMessage]) -> None
Add a list of messages.
Implementations should over-ride this method to handle bulk addition of messages in an efficient manner to avoid unnecessary round-trips to the underlying store.
| PARAMETER | DESCRIPTION |
|---|---|
messages
|
A sequence of
TYPE:
|
aadd_messages
async
¶
aadd_messages(messages: Sequence[BaseMessage]) -> None
Async add a list of messages.
| PARAMETER | DESCRIPTION |
|---|---|
messages
|
A sequence of
TYPE:
|
__init__
¶
__init__(
cosmos_endpoint: str,
cosmos_database: str,
cosmos_container: str,
session_id: str,
user_id: str,
credential: Any = None,
connection_string: str | None = None,
ttl: int | None = None,
cosmos_client_kwargs: dict | None = None,
)
Initializes a new instance of the CosmosDBChatMessageHistory class.
Make sure to call prepare_cosmos or use the context manager to make sure your database is ready.
Either a credential or a connection string must be provided.
:param cosmos_endpoint: The connection endpoint for the Azure Cosmos DB account. :param cosmos_database: The name of the database to use. :param cosmos_container: The name of the container to use. :param session_id: The session ID to use, can be overwritten while loading. :param user_id: The user ID to use, can be overwritten while loading. :param credential: The credential to use to authenticate to Azure Cosmos DB. :param connection_string: The connection string to use to authenticate. :param ttl: The time to live (in seconds) to use for documents in the container. :param cosmos_client_kwargs: Additional kwargs to pass to the CosmosClient.
prepare_cosmos
¶
Prepare the CosmosDB client.
Use this function or the context manager to make sure your database is ready.
__exit__
¶
__exit__(
exc_type: Type[BaseException] | None,
exc_val: BaseException | None,
traceback: TracebackType | None,
) -> None
Context manager exit.
langchain_azure_ai.chat_models
¶
Chat completions model for Azure AI.
AzureAIChatCompletionsModel
¶
Bases: BaseChatModel, ModelInferenceService
Azure AI Chat Completions Model.
The Azure AI model inference API (https://aka.ms/azureai/modelinference) provides a common layer to talk with most models deployed to Azure AI. This class providers inference for chat completions models supporting it. See documentation for the list of models supporting the API.
Examples:
from langchain_azure_ai.chat_models import AzureAIChatCompletionsModel
from langchain_core.messages import HumanMessage, SystemMessage
model = AzureAIChatCompletionsModel(
endpoint="https://[your-service].services.ai.azure.com/models",
credential="your-api-key",
model="mistral-large-2407",
)
messages = [
SystemMessage(
content="Translate the following from English into Italian"
),
HumanMessage(content="hi!"),
]
model.invoke(messages)
For serverless endpoints running a single model, the model_name parameter
can be omitted:
from langchain_azure_ai.chat_models import AzureAIChatCompletionsModel
from langchain_core.messages import HumanMessage, SystemMessage
model = AzureAIChatCompletionsModel(
endpoint="https://[your-service].inference.ai.azure.com",
credential="your-api-key",
)
messages = [
SystemMessage(
content="Translate the following from English into Italian"
),
HumanMessage(content="hi!"),
]
model.invoke(messages)
You can pass additional properties to the underlying model, including
temperature, top_p, presence_penalty, etc.
model = AzureAIChatCompletionsModel(
endpoint="https://[your-service].services.ai.azure.com/models",
credential="your-api-key",
model="mistral-large-2407",
temperature=0.5,
top_p=0.9,
)
Azure OpenAI models require to pass the route `openai/v1`.
```python
model = AzureAIChatCompletionsModel(
endpoint="https://[your-service].services.ai.azure.com/openai/v1",
model="gpt-4.1",
credential="your-api-key",
)
Structured Output:
To use structured output with Azure AI models, you can use the
with_structured_output method. This method supports the same methods
as the base class, including function_calling, json_mode, and
json_schema.
from langchain_azure_ai.chat_models import AzureAIChatCompletionsModel
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.messages import HumanMessage
class Joke(BaseModel):
joke: str
model = AzureAIChatCompletionsModel(
endpoint="https://[your-service].services.ai.azure.com/models",
credential="your-api-key",
model="mistral-large-2407",
).with_structured_output(Joke, method="json_schema")
!!! note
Using `method="function_calling"` requires the model to support
function calling and `tool_choice". Use "json_mode" or
"json_schema" for best support.
**Troubleshooting:**
To diagnostic issues with the model, you can enable debug logging:
```python
import sys
import logging
from langchain_azure_ai.chat_models import AzureAIChatCompletionsModel
logger = logging.getLogger("azure")
# Set the desired logging level. logging.
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler(stream=sys.stdout)
logger.addHandler(handler)
model = AzureAIChatCompletionsModel(
endpoint="https://[your-service].services.ai.azure.com/models",
credential="your-api-key",
model="mistral-large-2407",
client_kwargs={ "logging_enable": True }
)
| METHOD | DESCRIPTION |
|---|---|
validate_environment |
Validate that required values are present in the environment. |
get_name |
Get the name of the |
get_input_schema |
Get a Pydantic model that can be used to validate input to the |
get_input_jsonschema |
Get a JSON schema that represents the input to the |
get_output_schema |
Get a Pydantic model that can be used to validate output to the |
get_output_jsonschema |
Get a JSON schema that represents the output of the |
config_schema |
The type of config this |
get_config_jsonschema |
Get a JSON schema that represents the config of the |
get_graph |
Return a graph representation of this |
get_prompts |
Return a list of prompts used by this |
__or__ |
Runnable "or" operator. |
__ror__ |
Runnable "reverse-or" operator. |
pipe |
Pipe |
pick |
Pick keys from the output |
assign |
Assigns new fields to the |
invoke |
Transform a single input into an output. |
ainvoke |
Transform a single input into an output. |
batch |
Default implementation runs invoke in parallel using a thread pool executor. |
batch_as_completed |
Run |
abatch |
Default implementation runs |
abatch_as_completed |
Run |
stream |
Default implementation of |
astream |
Default implementation of |
astream_log |
Stream all output from a |
astream_events |
Generate a stream of events. |
transform |
Transform inputs to outputs. |
atransform |
Transform inputs to outputs. |
bind |
Bind arguments to a |
with_config |
Bind config to a |
with_listeners |
Bind lifecycle listeners to a |
with_alisteners |
Bind async lifecycle listeners to a |
with_types |
Bind input and output types to a |
with_retry |
Create a new |
map |
Return a new |
with_fallbacks |
Add fallbacks to a |
as_tool |
Create a |
__init__ |
|
is_lc_serializable |
Is this class serializable? |
lc_id |
Return a unique identifier for this class for serialization purposes. |
to_json |
Serialize the |
to_json_not_implemented |
Serialize a "not implemented" object. |
configurable_fields |
Configure particular |
configurable_alternatives |
Configure alternatives for |
set_verbose |
If verbose is |
generate_prompt |
Pass a sequence of prompts to the model and return model generations. |
agenerate_prompt |
Asynchronously pass a sequence of prompts and return model generations. |
get_token_ids |
Return the ordered IDs of the tokens in a text. |
get_num_tokens |
Get the number of tokens present in the text. |
get_num_tokens_from_messages |
Get the number of tokens in the messages. |
generate |
Pass a sequence of prompts to the model and return model generations. |
agenerate |
Asynchronously pass a sequence of prompts to a model and return generations. |
dict |
Return a dictionary of the LLM. |
initialize_client |
Initialize the Azure AI model inference client. |
bind_tools |
Bind tool-like objects to this chat model. |
with_structured_output |
Model wrapper that returns outputs formatted to match the given schema. |
get_lc_namespace |
Get the namespace of the langchain object. |
aclose |
Close the async client to prevent unclosed session warnings. |
project_endpoint
class-attribute
instance-attribute
¶
project_endpoint: str | None = None
The project endpoint associated with the AI project. If this is specified,
then the endpoint parameter becomes optional and credential has to be of type
TokenCredential.
endpoint
class-attribute
instance-attribute
¶
endpoint: str | None = None
The endpoint of the specific service to connect to. If you are connecting to a model, use the URL of the model deployment.
credential
class-attribute
instance-attribute
¶
credential: str | AzureKeyCredential | TokenCredential | None = None
The API key or credential to use to connect to the service. If using a project
endpoint, this must be of type TokenCredential since only Microsoft EntraID is
supported.
api_version
class-attribute
instance-attribute
¶
api_version: str | None = None
The API version to use with Azure. If None, the default version is used.
client_kwargs
class-attribute
instance-attribute
¶
Additional keyword arguments to pass to the client.
service
class-attribute
instance-attribute
¶
service: Literal['inference'] = 'inference'
The type of service to connect to. For Inference Services, use 'inference'.
name
class-attribute
instance-attribute
¶
name: str | None = None
The name of the Runnable. Used for debugging and tracing.
input_schema
property
¶
The type of input this Runnable accepts specified as a Pydantic model.
output_schema
property
¶
Output schema.
The type of output this Runnable produces specified as a Pydantic model.
config_specs
property
¶
config_specs: list[ConfigurableFieldSpec]
List configurable fields for this Runnable.
lc_secrets
property
¶
A map of constructor argument names to secret ids.
For example, {"openai_api_key": "OPENAI_API_KEY"}
lc_attributes
property
¶
lc_attributes: dict
List of attribute names that should be included in the serialized kwargs.
These attributes must be accepted by the constructor.
Default is an empty dictionary.
cache
class-attribute
instance-attribute
¶
Whether to cache the response.
- If
True, will use the global cache. - If
False, will not use a cache - If
None, will use the global cache if it's set, otherwise no cache. - If instance of
BaseCache, will use the provided cache.
Caching is not currently supported for streaming methods of models.
verbose
class-attribute
instance-attribute
¶
Whether to print out response text.
callbacks
class-attribute
instance-attribute
¶
callbacks: Callbacks = Field(default=None, exclude=True)
Callbacks to add to the run trace.
tags
class-attribute
instance-attribute
¶
Tags to add to the run trace.
metadata
class-attribute
instance-attribute
¶
Metadata to add to the run trace.
custom_get_token_ids
class-attribute
instance-attribute
¶
Optional encoder to use for counting tokens.
rate_limiter
class-attribute
instance-attribute
¶
rate_limiter: BaseRateLimiter | None = Field(default=None, exclude=True)
An optional rate limiter to use for limiting the number of requests.
disable_streaming
class-attribute
instance-attribute
¶
Whether to disable streaming for this model.
If streaming is bypassed, then stream/astream/astream_events will
defer to invoke/ainvoke.
- If
True, will always bypass streaming case. - If
'tool_calling', will bypass streaming case only when the model is called with atoolskeyword argument. In other words, LangChain will automatically switch to non-streaming behavior (invoke) only when the tools argument is provided. This offers the best of both worlds. - If
False(Default), will always use streaming case if available.
The main reason for this flag is that code might be written using stream and
a user may want to swap out a given model for another model whose the implementation
does not properly support streaming.
output_version
class-attribute
instance-attribute
¶
Version of AIMessage output format to store in message content.
AIMessage.content_blocks will lazily parse the contents of content into a
standard format. This flag can be used to additionally store the standard format
in message content, e.g., for serialization purposes.
Supported values:
'v0': provider-specific format in content (can lazily-parse withcontent_blocks)'v1': standardized format in content (consistent withcontent_blocks)
Partner packages (e.g.,
langchain-openai) can also use this
field to roll out new content formats in a backward-compatible way.
Added in langchain-core 1.0
profile
property
¶
profile: ModelProfile
Return profiling information for the model.
This property relies on the langchain-model-profiles package to retrieve chat
model capabilities, such as context window sizes and supported features.
| RAISES | DESCRIPTION |
|---|---|
ImportError
|
If |
| RETURNS | DESCRIPTION |
|---|---|
ModelProfile
|
A |
model_name
class-attribute
instance-attribute
¶
The name of the model to use for inference, if the endpoint is running more than one model. If not, this parameter is ignored.
max_tokens
class-attribute
instance-attribute
¶
max_tokens: int | None = None
The maximum number of tokens to generate in the response. If None, the default maximum tokens is used.
temperature
class-attribute
instance-attribute
¶
temperature: float | None = None
The temperature to use for sampling from the model. If None, the default temperature is used.
top_p
class-attribute
instance-attribute
¶
top_p: float | None = None
The top-p value to use for sampling from the model. If None, the default top-p value is used.
presence_penalty
class-attribute
instance-attribute
¶
presence_penalty: float | None = None
The presence penalty to use for sampling from the model. If None, the default presence penalty is used.
frequency_penalty
class-attribute
instance-attribute
¶
frequency_penalty: float | None = None
The frequency penalty to use for sampling from the model. If None, the default frequency penalty is used.
stop
class-attribute
instance-attribute
¶
stop: str | None = None
The stop token to use for stopping generation. If None, the default stop token is used.
seed
class-attribute
instance-attribute
¶
seed: int | None = None
The seed to use for random number generation. If None, the default seed is used.
model_kwargs
class-attribute
instance-attribute
¶
Additional kwargs model parameters.
validate_environment
¶
Validate that required values are present in the environment.
get_name
¶
get_input_schema
¶
get_input_schema(config: RunnableConfig | None = None) -> type[BaseModel]
Get a Pydantic model that can be used to validate input to the Runnable.
Runnable objects that leverage the configurable_fields and
configurable_alternatives methods will have a dynamic input schema that
depends on which configuration the Runnable is invoked with.
This method allows to get an input schema for a specific configuration.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
A config to use when generating the schema.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
type[BaseModel]
|
A Pydantic model that can be used to validate input. |
get_input_jsonschema
¶
get_input_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]
Get a JSON schema that represents the input to the Runnable.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
A config to use when generating the schema.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
dict[str, Any]
|
A JSON schema that represents the input to the |
Example
Added in langchain-core 0.3.0
get_output_schema
¶
get_output_schema(config: RunnableConfig | None = None) -> type[BaseModel]
Get a Pydantic model that can be used to validate output to the Runnable.
Runnable objects that leverage the configurable_fields and
configurable_alternatives methods will have a dynamic output schema that
depends on which configuration the Runnable is invoked with.
This method allows to get an output schema for a specific configuration.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
A config to use when generating the schema.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
type[BaseModel]
|
A Pydantic model that can be used to validate output. |
get_output_jsonschema
¶
get_output_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]
Get a JSON schema that represents the output of the Runnable.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
A config to use when generating the schema.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
dict[str, Any]
|
A JSON schema that represents the output of the |
Example
Added in langchain-core 0.3.0
config_schema
¶
The type of config this Runnable accepts specified as a Pydantic model.
To mark a field as configurable, see the configurable_fields
and configurable_alternatives methods.
| PARAMETER | DESCRIPTION |
|---|---|
include
|
A list of fields to include in the config schema. |
| RETURNS | DESCRIPTION |
|---|---|
type[BaseModel]
|
A Pydantic model that can be used to validate config. |
get_config_jsonschema
¶
get_graph
¶
get_graph(config: RunnableConfig | None = None) -> Graph
Return a graph representation of this Runnable.
get_prompts
¶
get_prompts(config: RunnableConfig | None = None) -> list[BasePromptTemplate]
Return a list of prompts used by this Runnable.
__or__
¶
__or__(
other: Runnable[Any, Other]
| Callable[[Iterator[Any]], Iterator[Other]]
| Callable[[AsyncIterator[Any]], AsyncIterator[Other]]
| Callable[[Any], Other]
| Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any],
) -> RunnableSerializable[Input, Other]
Runnable "or" operator.
Compose this Runnable with another object to create a
RunnableSequence.
| PARAMETER | DESCRIPTION |
|---|---|
other
|
Another
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Input, Other]
|
A new |
__ror__
¶
__ror__(
other: Runnable[Other, Any]
| Callable[[Iterator[Other]], Iterator[Any]]
| Callable[[AsyncIterator[Other]], AsyncIterator[Any]]
| Callable[[Other], Any]
| Mapping[str, Runnable[Other, Any] | Callable[[Other], Any] | Any],
) -> RunnableSerializable[Other, Output]
Runnable "reverse-or" operator.
Compose this Runnable with another object to create a
RunnableSequence.
| PARAMETER | DESCRIPTION |
|---|---|
other
|
Another
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Other, Output]
|
A new |
pipe
¶
pipe(
*others: Runnable[Any, Other] | Callable[[Any], Other], name: str | None = None
) -> RunnableSerializable[Input, Other]
Pipe Runnable objects.
Compose this Runnable with Runnable-like objects to make a
RunnableSequence.
Equivalent to RunnableSequence(self, *others) or self | others[0] | ...
Example
from langchain_core.runnables import RunnableLambda
def add_one(x: int) -> int:
return x + 1
def mul_two(x: int) -> int:
return x * 2
runnable_1 = RunnableLambda(add_one)
runnable_2 = RunnableLambda(mul_two)
sequence = runnable_1.pipe(runnable_2)
# Or equivalently:
# sequence = runnable_1 | runnable_2
# sequence = RunnableSequence(first=runnable_1, last=runnable_2)
sequence.invoke(1)
await sequence.ainvoke(1)
# -> 4
sequence.batch([1, 2, 3])
await sequence.abatch([1, 2, 3])
# -> [4, 6, 8]
| PARAMETER | DESCRIPTION |
|---|---|
*others
|
Other
TYPE:
|
name
|
An optional name for the resulting
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Input, Other]
|
A new |
pick
¶
Pick keys from the output dict of this Runnable.
Pick a single key:
import json
from langchain_core.runnables import RunnableLambda, RunnableMap
as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)
chain = RunnableMap(str=as_str, json=as_json)
chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3]}
json_only_chain = chain.pick("json")
json_only_chain.invoke("[1, 2, 3]")
# -> [1, 2, 3]
Pick a list of keys:
from typing import Any
import json
from langchain_core.runnables import RunnableLambda, RunnableMap
as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)
def as_bytes(x: Any) -> bytes:
return bytes(x, "utf-8")
chain = RunnableMap(str=as_str, json=as_json, bytes=RunnableLambda(as_bytes))
chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3], "bytes": b"[1, 2, 3]"}
json_and_bytes_chain = chain.pick(["json", "bytes"])
json_and_bytes_chain.invoke("[1, 2, 3]")
# -> {"json": [1, 2, 3], "bytes": b"[1, 2, 3]"}
| PARAMETER | DESCRIPTION |
|---|---|
keys
|
A key or list of keys to pick from the output dict. |
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Any, Any]
|
a new |
assign
¶
assign(
**kwargs: Runnable[dict[str, Any], Any]
| Callable[[dict[str, Any]], Any]
| Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]],
) -> RunnableSerializable[Any, Any]
Assigns new fields to the dict output of this Runnable.
from langchain_core.language_models.fake import FakeStreamingListLLM
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import SystemMessagePromptTemplate
from langchain_core.runnables import Runnable
from operator import itemgetter
prompt = (
SystemMessagePromptTemplate.from_template("You are a nice assistant.")
+ "{question}"
)
model = FakeStreamingListLLM(responses=["foo-lish"])
chain: Runnable = prompt | model | {"str": StrOutputParser()}
chain_with_assign = chain.assign(hello=itemgetter("str") | model)
print(chain_with_assign.input_schema.model_json_schema())
# {'title': 'PromptInput', 'type': 'object', 'properties':
{'question': {'title': 'Question', 'type': 'string'}}}
print(chain_with_assign.output_schema.model_json_schema())
# {'title': 'RunnableSequenceOutput', 'type': 'object', 'properties':
{'str': {'title': 'Str',
'type': 'string'}, 'hello': {'title': 'Hello', 'type': 'string'}}}
| PARAMETER | DESCRIPTION |
|---|---|
**kwargs
|
A mapping of keys to
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Any, Any]
|
A new |
invoke
¶
invoke(
input: LanguageModelInput,
config: RunnableConfig | None = None,
*,
stop: list[str] | None = None,
**kwargs: Any,
) -> AIMessage
Transform a single input into an output.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
A config to use when invoking the The config supports standard keys like Please refer to
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Output
|
The output of the |
ainvoke
async
¶
ainvoke(
input: LanguageModelInput,
config: RunnableConfig | None = None,
*,
stop: list[str] | None = None,
**kwargs: Any,
) -> AIMessage
Transform a single input into an output.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
A config to use when invoking the The config supports standard keys like Please refer to
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Output
|
The output of the |
batch
¶
batch(
inputs: list[Input],
config: RunnableConfig | list[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any | None,
) -> list[Output]
Default implementation runs invoke in parallel using a thread pool executor.
The default implementation of batch works well for IO bound runnables.
Subclasses must override this method if they can batch more efficiently;
e.g., if the underlying Runnable uses an API which supports a batch mode.
| PARAMETER | DESCRIPTION |
|---|---|
inputs
|
A list of inputs to the
TYPE:
|
config
|
A config to use when invoking the Please refer to
TYPE:
|
return_exceptions
|
Whether to return exceptions instead of raising them.
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Output]
|
A list of outputs from the |
batch_as_completed
¶
batch_as_completed(
inputs: Sequence[Input],
config: RunnableConfig | Sequence[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any | None,
) -> Iterator[tuple[int, Output | Exception]]
Run invoke in parallel on a list of inputs.
Yields results as they complete.
| PARAMETER | DESCRIPTION |
|---|---|
inputs
|
A list of inputs to the
TYPE:
|
config
|
A config to use when invoking the The config supports standard keys like Please refer to
TYPE:
|
return_exceptions
|
Whether to return exceptions instead of raising them.
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
tuple[int, Output | Exception]
|
Tuples of the index of the input and the output from the |
abatch
async
¶
abatch(
inputs: list[Input],
config: RunnableConfig | list[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any | None,
) -> list[Output]
Default implementation runs ainvoke in parallel using asyncio.gather.
The default implementation of batch works well for IO bound runnables.
Subclasses must override this method if they can batch more efficiently;
e.g., if the underlying Runnable uses an API which supports a batch mode.
| PARAMETER | DESCRIPTION |
|---|---|
inputs
|
A list of inputs to the
TYPE:
|
config
|
A config to use when invoking the The config supports standard keys like Please refer to
TYPE:
|
return_exceptions
|
Whether to return exceptions instead of raising them.
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Output]
|
A list of outputs from the |
abatch_as_completed
async
¶
abatch_as_completed(
inputs: Sequence[Input],
config: RunnableConfig | Sequence[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any | None,
) -> AsyncIterator[tuple[int, Output | Exception]]
Run ainvoke in parallel on a list of inputs.
Yields results as they complete.
| PARAMETER | DESCRIPTION |
|---|---|
inputs
|
A list of inputs to the
TYPE:
|
config
|
A config to use when invoking the The config supports standard keys like Please refer to
TYPE:
|
return_exceptions
|
Whether to return exceptions instead of raising them.
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[tuple[int, Output | Exception]]
|
A tuple of the index of the input and the output from the |
stream
¶
stream(
input: LanguageModelInput,
config: RunnableConfig | None = None,
*,
stop: list[str] | None = None,
**kwargs: Any,
) -> Iterator[AIMessageChunk]
Default implementation of stream, which calls invoke.
Subclasses must override this method if they support streaming output.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
Output
|
The output of the |
astream
async
¶
astream(
input: LanguageModelInput,
config: RunnableConfig | None = None,
*,
stop: list[str] | None = None,
**kwargs: Any,
) -> AsyncIterator[AIMessageChunk]
Default implementation of astream, which calls ainvoke.
Subclasses must override this method if they support streaming output.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[Output]
|
The output of the |
astream_log
async
¶
astream_log(
input: Any,
config: RunnableConfig | None = None,
*,
diff: bool = True,
with_streamed_output_list: bool = True,
include_names: Sequence[str] | None = None,
include_types: Sequence[str] | None = None,
include_tags: Sequence[str] | None = None,
exclude_names: Sequence[str] | None = None,
exclude_types: Sequence[str] | None = None,
exclude_tags: Sequence[str] | None = None,
**kwargs: Any,
) -> AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]
Stream all output from a Runnable, as reported to the callback system.
This includes all inner runs of LLMs, Retrievers, Tools, etc.
Output is streamed as Log objects, which include a list of Jsonpatch ops that describe how the state of the run has changed in each step, and the final state of the run.
The Jsonpatch ops can be applied in order to construct state.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
diff
|
Whether to yield diffs between each step or the current state.
TYPE:
|
with_streamed_output_list
|
Whether to yield the
TYPE:
|
include_names
|
Only include logs with these names. |
include_types
|
Only include logs with these types. |
include_tags
|
Only include logs with these tags. |
exclude_names
|
Exclude logs with these names. |
exclude_types
|
Exclude logs with these types. |
exclude_tags
|
Exclude logs with these tags. |
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]
|
A |
astream_events
async
¶
astream_events(
input: Any,
config: RunnableConfig | None = None,
*,
version: Literal["v1", "v2"] = "v2",
include_names: Sequence[str] | None = None,
include_types: Sequence[str] | None = None,
include_tags: Sequence[str] | None = None,
exclude_names: Sequence[str] | None = None,
exclude_types: Sequence[str] | None = None,
exclude_tags: Sequence[str] | None = None,
**kwargs: Any,
) -> AsyncIterator[StreamEvent]
Generate a stream of events.
Use to create an iterator over StreamEvent that provide real-time information
about the progress of the Runnable, including StreamEvent from intermediate
results.
A StreamEvent is a dictionary with the following schema:
event: Event names are of the format:on_[runnable_type]_(start|stream|end).name: The name of theRunnablethat generated the event.run_id: Randomly generated ID associated with the given execution of theRunnablethat emitted the event. A childRunnablethat gets invoked as part of the execution of a parentRunnableis assigned its own unique ID.parent_ids: The IDs of the parent runnables that generated the event. The rootRunnablewill have an empty list. The order of the parent IDs is from the root to the immediate parent. Only available for v2 version of the API. The v1 version of the API will return an empty list.tags: The tags of theRunnablethat generated the event.metadata: The metadata of theRunnablethat generated the event.data: The data associated with the event. The contents of this field depend on the type of event. See the table below for more details.
Below is a table that illustrates some events that might be emitted by various chains. Metadata fields have been omitted from the table for brevity. Chain definitions have been included after the table.
Note
This reference table is for the v2 version of the schema.
| event | name | chunk | input | output |
|---|---|---|---|---|
on_chat_model_start |
'[model name]' |
{"messages": [[SystemMessage, HumanMessage]]} |
||
on_chat_model_stream |
'[model name]' |
AIMessageChunk(content="hello") |
||
on_chat_model_end |
'[model name]' |
{"messages": [[SystemMessage, HumanMessage]]} |
AIMessageChunk(content="hello world") |
|
on_llm_start |
'[model name]' |
{'input': 'hello'} |
||
on_llm_stream |
'[model name]' |
'Hello' |
||
on_llm_end |
'[model name]' |
'Hello human!' |
||
on_chain_start |
'format_docs' |
|||
on_chain_stream |
'format_docs' |
'hello world!, goodbye world!' |
||
on_chain_end |
'format_docs' |
[Document(...)] |
'hello world!, goodbye world!' |
|
on_tool_start |
'some_tool' |
{"x": 1, "y": "2"} |
||
on_tool_end |
'some_tool' |
{"x": 1, "y": "2"} |
||
on_retriever_start |
'[retriever name]' |
{"query": "hello"} |
||
on_retriever_end |
'[retriever name]' |
{"query": "hello"} |
[Document(...), ..] |
|
on_prompt_start |
'[template_name]' |
{"question": "hello"} |
||
on_prompt_end |
'[template_name]' |
{"question": "hello"} |
ChatPromptValue(messages: [SystemMessage, ...]) |
In addition to the standard events, users can also dispatch custom events (see example below).
Custom events will be only be surfaced with in the v2 version of the API!
A custom event has following format:
| Attribute | Type | Description |
|---|---|---|
name |
str |
A user defined name for the event. |
data |
Any |
The data associated with the event. This can be anything, though we suggest making it JSON serializable. |
Here are declarations associated with the standard events shown above:
format_docs:
def format_docs(docs: list[Document]) -> str:
'''Format the docs.'''
return ", ".join([doc.page_content for doc in docs])
format_docs = RunnableLambda(format_docs)
some_tool:
prompt:
template = ChatPromptTemplate.from_messages(
[
("system", "You are Cat Agent 007"),
("human", "{question}"),
]
).with_config({"run_name": "my_template", "tags": ["my_template"]})
For instance:
from langchain_core.runnables import RunnableLambda
async def reverse(s: str) -> str:
return s[::-1]
chain = RunnableLambda(func=reverse)
events = [event async for event in chain.astream_events("hello", version="v2")]
# Will produce the following events
# (run_id, and parent_ids has been omitted for brevity):
[
{
"data": {"input": "hello"},
"event": "on_chain_start",
"metadata": {},
"name": "reverse",
"tags": [],
},
{
"data": {"chunk": "olleh"},
"event": "on_chain_stream",
"metadata": {},
"name": "reverse",
"tags": [],
},
{
"data": {"output": "olleh"},
"event": "on_chain_end",
"metadata": {},
"name": "reverse",
"tags": [],
},
]
from langchain_core.callbacks.manager import (
adispatch_custom_event,
)
from langchain_core.runnables import RunnableLambda, RunnableConfig
import asyncio
async def slow_thing(some_input: str, config: RunnableConfig) -> str:
"""Do something that takes a long time."""
await asyncio.sleep(1) # Placeholder for some slow operation
await adispatch_custom_event(
"progress_event",
{"message": "Finished step 1 of 3"},
config=config # Must be included for python < 3.10
)
await asyncio.sleep(1) # Placeholder for some slow operation
await adispatch_custom_event(
"progress_event",
{"message": "Finished step 2 of 3"},
config=config # Must be included for python < 3.10
)
await asyncio.sleep(1) # Placeholder for some slow operation
return "Done"
slow_thing = RunnableLambda(slow_thing)
async for event in slow_thing.astream_events("some_input", version="v2"):
print(event)
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
version
|
The version of the schema to use either
TYPE:
|
include_names
|
Only include events from |
include_types
|
Only include events from |
include_tags
|
Only include events from |
exclude_names
|
Exclude events from |
exclude_types
|
Exclude events from |
exclude_tags
|
Exclude events from |
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[StreamEvent]
|
An async stream of |
| RAISES | DESCRIPTION |
|---|---|
NotImplementedError
|
If the version is not |
transform
¶
transform(
input: Iterator[Input], config: RunnableConfig | None = None, **kwargs: Any | None
) -> Iterator[Output]
Transform inputs to outputs.
Default implementation of transform, which buffers input and calls astream.
Subclasses must override this method if they can start producing output while input is still being generated.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
An iterator of inputs to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
Output
|
The output of the |
atransform
async
¶
atransform(
input: AsyncIterator[Input],
config: RunnableConfig | None = None,
**kwargs: Any | None,
) -> AsyncIterator[Output]
Transform inputs to outputs.
Default implementation of atransform, which buffers input and calls astream.
Subclasses must override this method if they can start producing output while input is still being generated.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
An async iterator of inputs to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[Output]
|
The output of the |
bind
¶
Bind arguments to a Runnable, returning a new Runnable.
Useful when a Runnable in a chain requires an argument that is not
in the output of the previous Runnable or included in the user input.
| PARAMETER | DESCRIPTION |
|---|---|
**kwargs
|
The arguments to bind to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser
model = ChatOllama(model="llama3.1")
# Without bind
chain = model | StrOutputParser()
chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two three four five.'
# With bind
chain = model.bind(stop=["three"]) | StrOutputParser()
chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two'
with_config
¶
with_config(
config: RunnableConfig | None = None, **kwargs: Any
) -> Runnable[Input, Output]
Bind config to a Runnable, returning a new Runnable.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
The config to bind to the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
with_listeners
¶
with_listeners(
*,
on_start: Callable[[Run], None]
| Callable[[Run, RunnableConfig], None]
| None = None,
on_end: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None = None,
on_error: Callable[[Run], None]
| Callable[[Run, RunnableConfig], None]
| None = None,
) -> Runnable[Input, Output]
Bind lifecycle listeners to a Runnable, returning a new Runnable.
The Run object contains information about the run, including its id,
type, input, output, error, start_time, end_time, and
any tags or metadata added to the run.
| PARAMETER | DESCRIPTION |
|---|---|
on_start
|
Called before the
TYPE:
|
on_end
|
Called after the
TYPE:
|
on_error
|
Called if the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_core.runnables import RunnableLambda
from langchain_core.tracers.schemas import Run
import time
def test_runnable(time_to_sleep: int):
time.sleep(time_to_sleep)
def fn_start(run_obj: Run):
print("start_time:", run_obj.start_time)
def fn_end(run_obj: Run):
print("end_time:", run_obj.end_time)
chain = RunnableLambda(test_runnable).with_listeners(
on_start=fn_start, on_end=fn_end
)
chain.invoke(2)
with_alisteners
¶
with_alisteners(
*,
on_start: AsyncListener | None = None,
on_end: AsyncListener | None = None,
on_error: AsyncListener | None = None,
) -> Runnable[Input, Output]
Bind async lifecycle listeners to a Runnable.
Returns a new Runnable.
The Run object contains information about the run, including its id,
type, input, output, error, start_time, end_time, and
any tags or metadata added to the run.
| PARAMETER | DESCRIPTION |
|---|---|
on_start
|
Called asynchronously before the
TYPE:
|
on_end
|
Called asynchronously after the
TYPE:
|
on_error
|
Called asynchronously if the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_core.runnables import RunnableLambda, Runnable
from datetime import datetime, timezone
import time
import asyncio
def format_t(timestamp: float) -> str:
return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat()
async def test_runnable(time_to_sleep: int):
print(f"Runnable[{time_to_sleep}s]: starts at {format_t(time.time())}")
await asyncio.sleep(time_to_sleep)
print(f"Runnable[{time_to_sleep}s]: ends at {format_t(time.time())}")
async def fn_start(run_obj: Runnable):
print(f"on start callback starts at {format_t(time.time())}")
await asyncio.sleep(3)
print(f"on start callback ends at {format_t(time.time())}")
async def fn_end(run_obj: Runnable):
print(f"on end callback starts at {format_t(time.time())}")
await asyncio.sleep(2)
print(f"on end callback ends at {format_t(time.time())}")
runnable = RunnableLambda(test_runnable).with_alisteners(
on_start=fn_start, on_end=fn_end
)
async def concurrent_runs():
await asyncio.gather(runnable.ainvoke(2), runnable.ainvoke(3))
asyncio.run(concurrent_runs())
# Result:
# on start callback starts at 2025-03-01T07:05:22.875378+00:00
# on start callback starts at 2025-03-01T07:05:22.875495+00:00
# on start callback ends at 2025-03-01T07:05:25.878862+00:00
# on start callback ends at 2025-03-01T07:05:25.878947+00:00
# Runnable[2s]: starts at 2025-03-01T07:05:25.879392+00:00
# Runnable[3s]: starts at 2025-03-01T07:05:25.879804+00:00
# Runnable[2s]: ends at 2025-03-01T07:05:27.881998+00:00
# on end callback starts at 2025-03-01T07:05:27.882360+00:00
# Runnable[3s]: ends at 2025-03-01T07:05:28.881737+00:00
# on end callback starts at 2025-03-01T07:05:28.882428+00:00
# on end callback ends at 2025-03-01T07:05:29.883893+00:00
# on end callback ends at 2025-03-01T07:05:30.884831+00:00
with_types
¶
with_types(
*, input_type: type[Input] | None = None, output_type: type[Output] | None = None
) -> Runnable[Input, Output]
Bind input and output types to a Runnable, returning a new Runnable.
| PARAMETER | DESCRIPTION |
|---|---|
input_type
|
The input type to bind to the
TYPE:
|
output_type
|
The output type to bind to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
with_retry
¶
with_retry(
*,
retry_if_exception_type: tuple[type[BaseException], ...] = (Exception,),
wait_exponential_jitter: bool = True,
exponential_jitter_params: ExponentialJitterParams | None = None,
stop_after_attempt: int = 3,
) -> Runnable[Input, Output]
Create a new Runnable that retries the original Runnable on exceptions.
| PARAMETER | DESCRIPTION |
|---|---|
retry_if_exception_type
|
A tuple of exception types to retry on.
TYPE:
|
wait_exponential_jitter
|
Whether to add jitter to the wait time between retries.
TYPE:
|
stop_after_attempt
|
The maximum number of attempts to make before giving up.
TYPE:
|
exponential_jitter_params
|
Parameters for
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_core.runnables import RunnableLambda
count = 0
def _lambda(x: int) -> None:
global count
count = count + 1
if x == 1:
raise ValueError("x is 1")
else:
pass
runnable = RunnableLambda(_lambda)
try:
runnable.with_retry(
stop_after_attempt=2,
retry_if_exception_type=(ValueError,),
).invoke(1)
except ValueError:
pass
assert count == 2
map
¶
with_fallbacks
¶
with_fallbacks(
fallbacks: Sequence[Runnable[Input, Output]],
*,
exceptions_to_handle: tuple[type[BaseException], ...] = (Exception,),
exception_key: str | None = None,
) -> RunnableWithFallbacks[Input, Output]
Add fallbacks to a Runnable, returning a new Runnable.
The new Runnable will try the original Runnable, and then each fallback
in order, upon failures.
| PARAMETER | DESCRIPTION |
|---|---|
fallbacks
|
A sequence of runnables to try if the original |
exceptions_to_handle
|
A tuple of exception types to handle.
TYPE:
|
exception_key
|
If If If used, the base
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableWithFallbacks[Input, Output]
|
A new |
Example
from typing import Iterator
from langchain_core.runnables import RunnableGenerator
def _generate_immediate_error(input: Iterator) -> Iterator[str]:
raise ValueError()
yield ""
def _generate(input: Iterator) -> Iterator[str]:
yield from "foo bar"
runnable = RunnableGenerator(_generate_immediate_error).with_fallbacks(
[RunnableGenerator(_generate)]
)
print("".join(runnable.stream({}))) # foo bar
| PARAMETER | DESCRIPTION |
|---|---|
fallbacks
|
A sequence of runnables to try if the original |
exceptions_to_handle
|
A tuple of exception types to handle.
TYPE:
|
exception_key
|
If If If used, the base
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableWithFallbacks[Input, Output]
|
A new |
as_tool
¶
as_tool(
args_schema: type[BaseModel] | None = None,
*,
name: str | None = None,
description: str | None = None,
arg_types: dict[str, type] | None = None,
) -> BaseTool
Create a BaseTool from a Runnable.
as_tool will instantiate a BaseTool with a name, description, and
args_schema from a Runnable. Where possible, schemas are inferred
from runnable.get_input_schema.
Alternatively (e.g., if the Runnable takes a dict as input and the specific
dict keys are not typed), the schema can be specified directly with
args_schema.
You can also pass arg_types to just specify the required arguments and their
types.
| PARAMETER | DESCRIPTION |
|---|---|
args_schema
|
The schema for the tool. |
name
|
The name of the tool.
TYPE:
|
description
|
The description of the tool.
TYPE:
|
arg_types
|
A dictionary of argument names to types. |
| RETURNS | DESCRIPTION |
|---|---|
BaseTool
|
A |
Typed dict input:
from typing_extensions import TypedDict
from langchain_core.runnables import RunnableLambda
class Args(TypedDict):
a: int
b: list[int]
def f(x: Args) -> str:
return str(x["a"] * max(x["b"]))
runnable = RunnableLambda(f)
as_tool = runnable.as_tool()
as_tool.invoke({"a": 3, "b": [1, 2]})
dict input, specifying schema via args_schema:
from typing import Any
from pydantic import BaseModel, Field
from langchain_core.runnables import RunnableLambda
def f(x: dict[str, Any]) -> str:
return str(x["a"] * max(x["b"]))
class FSchema(BaseModel):
"""Apply a function to an integer and list of integers."""
a: int = Field(..., description="Integer")
b: list[int] = Field(..., description="List of ints")
runnable = RunnableLambda(f)
as_tool = runnable.as_tool(FSchema)
as_tool.invoke({"a": 3, "b": [1, 2]})
dict input, specifying schema via arg_types:
from typing import Any
from langchain_core.runnables import RunnableLambda
def f(x: dict[str, Any]) -> str:
return str(x["a"] * max(x["b"]))
runnable = RunnableLambda(f)
as_tool = runnable.as_tool(arg_types={"a": int, "b": list[int]})
as_tool.invoke({"a": 3, "b": [1, 2]})
str input:
is_lc_serializable
classmethod
¶
is_lc_serializable() -> bool
Is this class serializable?
By design, even if a class inherits from Serializable, it is not serializable
by default. This is to prevent accidental serialization of objects that should
not be serialized.
| RETURNS | DESCRIPTION |
|---|---|
bool
|
Whether the class is serializable. Default is |
lc_id
classmethod
¶
Return a unique identifier for this class for serialization purposes.
The unique identifier is a list of strings that describes the path to the object.
For example, for the class langchain.llms.openai.OpenAI, the id is
["langchain", "llms", "openai", "OpenAI"].
to_json
¶
Serialize the Runnable to JSON.
| RETURNS | DESCRIPTION |
|---|---|
SerializedConstructor | SerializedNotImplemented
|
A JSON-serializable representation of the |
to_json_not_implemented
¶
Serialize a "not implemented" object.
| RETURNS | DESCRIPTION |
|---|---|
SerializedNotImplemented
|
|
configurable_fields
¶
configurable_fields(
**kwargs: AnyConfigurableField,
) -> RunnableSerializable[Input, Output]
Configure particular Runnable fields at runtime.
| PARAMETER | DESCRIPTION |
|---|---|
**kwargs
|
A dictionary of
TYPE:
|
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If a configuration key is not found in the |
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Input, Output]
|
A new |
from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI
model = ChatOpenAI(max_tokens=20).configurable_fields(
max_tokens=ConfigurableField(
id="output_token_number",
name="Max tokens in the output",
description="The maximum number of tokens in the output",
)
)
# max_tokens = 20
print("max_tokens_20: ", model.invoke("tell me something about chess").content)
# max_tokens = 200
print(
"max_tokens_200: ",
model.with_config(configurable={"output_token_number": 200})
.invoke("tell me something about chess")
.content,
)
configurable_alternatives
¶
configurable_alternatives(
which: ConfigurableField,
*,
default_key: str = "default",
prefix_keys: bool = False,
**kwargs: Runnable[Input, Output] | Callable[[], Runnable[Input, Output]],
) -> RunnableSerializable[Input, Output]
Configure alternatives for Runnable objects that can be set at runtime.
| PARAMETER | DESCRIPTION |
|---|---|
which
|
The
TYPE:
|
default_key
|
The default key to use if no alternative is selected.
TYPE:
|
prefix_keys
|
Whether to prefix the keys with the
TYPE:
|
**kwargs
|
A dictionary of keys to
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Input, Output]
|
A new |
from langchain_anthropic import ChatAnthropic
from langchain_core.runnables.utils import ConfigurableField
from langchain_openai import ChatOpenAI
model = ChatAnthropic(
model_name="claude-sonnet-4-5-20250929"
).configurable_alternatives(
ConfigurableField(id="llm"),
default_key="anthropic",
openai=ChatOpenAI(),
)
# uses the default model ChatAnthropic
print(model.invoke("which organization created you?").content)
# uses ChatOpenAI
print(
model.with_config(configurable={"llm": "openai"})
.invoke("which organization created you?")
.content
)
set_verbose
¶
generate_prompt
¶
generate_prompt(
prompts: list[PromptValue],
stop: list[str] | None = None,
callbacks: Callbacks = None,
**kwargs: Any,
) -> LLMResult
Pass a sequence of prompts to the model and return model generations.
This method should make use of batched calls for models that expose a batched API.
Use this method when you want to:
- Take advantage of batched calls,
- Need more output from the model than just the top generated value,
- Are building chains that are agnostic to the underlying language model type (e.g., pure text completion models vs chat models).
| PARAMETER | DESCRIPTION |
|---|---|
prompts
|
List of A
TYPE:
|
stop
|
Stop words to use when generating. Model output is cut off at the first occurrence of any of these substrings. |
callbacks
|
Used for executing additional functionality, such as logging or streaming, throughout generation.
TYPE:
|
**kwargs
|
Arbitrary additional keyword arguments. These are usually passed to the model provider API call.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
LLMResult
|
An |
agenerate_prompt
async
¶
agenerate_prompt(
prompts: list[PromptValue],
stop: list[str] | None = None,
callbacks: Callbacks = None,
**kwargs: Any,
) -> LLMResult
Asynchronously pass a sequence of prompts and return model generations.
This method should make use of batched calls for models that expose a batched API.
Use this method when you want to:
- Take advantage of batched calls,
- Need more output from the model than just the top generated value,
- Are building chains that are agnostic to the underlying language model type (e.g., pure text completion models vs chat models).
| PARAMETER | DESCRIPTION |
|---|---|
prompts
|
List of A
TYPE:
|
stop
|
Stop words to use when generating. Model output is cut off at the first occurrence of any of these substrings. |
callbacks
|
Used for executing additional functionality, such as logging or streaming, throughout generation.
TYPE:
|
**kwargs
|
Arbitrary additional keyword arguments. These are usually passed to the model provider API call.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
LLMResult
|
An |
get_token_ids
¶
get_num_tokens
¶
Get the number of tokens present in the text.
Useful for checking if an input fits in a model's context window.
This should be overridden by model-specific implementations to provide accurate token counts via model-specific tokenizers.
| PARAMETER | DESCRIPTION |
|---|---|
text
|
The string input to tokenize.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
int
|
The integer number of tokens in the text. |
get_num_tokens_from_messages
¶
get_num_tokens_from_messages(
messages: list[BaseMessage], tools: Sequence | None = None
) -> int
Get the number of tokens in the messages.
Useful for checking if an input fits in a model's context window.
This should be overridden by model-specific implementations to provide accurate token counts via model-specific tokenizers.
Note
- The base implementation of
get_num_tokens_from_messagesignores tool schemas. - The base implementation of
get_num_tokens_from_messagesadds additional prefixes to messages in represent user roles, which will add to the overall token count. Model-specific implementations may choose to handle this differently.
| PARAMETER | DESCRIPTION |
|---|---|
messages
|
The message inputs to tokenize.
TYPE:
|
tools
|
If provided, sequence of dict,
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
int
|
The sum of the number of tokens across the messages. |
generate
¶
generate(
messages: list[list[BaseMessage]],
stop: list[str] | None = None,
callbacks: Callbacks = None,
*,
tags: list[str] | None = None,
metadata: dict[str, Any] | None = None,
run_name: str | None = None,
run_id: UUID | None = None,
**kwargs: Any,
) -> LLMResult
Pass a sequence of prompts to the model and return model generations.
This method should make use of batched calls for models that expose a batched API.
Use this method when you want to:
- Take advantage of batched calls,
- Need more output from the model than just the top generated value,
- Are building chains that are agnostic to the underlying language model type (e.g., pure text completion models vs chat models).
| PARAMETER | DESCRIPTION |
|---|---|
messages
|
List of list of messages.
TYPE:
|
stop
|
Stop words to use when generating. Model output is cut off at the first occurrence of any of these substrings. |
callbacks
|
Used for executing additional functionality, such as logging or streaming, throughout generation.
TYPE:
|
tags
|
The tags to apply. |
metadata
|
The metadata to apply. |
run_name
|
The name of the run.
TYPE:
|
run_id
|
The ID of the run.
TYPE:
|
**kwargs
|
Arbitrary additional keyword arguments. These are usually passed to the model provider API call.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
LLMResult
|
An |
agenerate
async
¶
agenerate(
messages: list[list[BaseMessage]],
stop: list[str] | None = None,
callbacks: Callbacks = None,
*,
tags: list[str] | None = None,
metadata: dict[str, Any] | None = None,
run_name: str | None = None,
run_id: UUID | None = None,
**kwargs: Any,
) -> LLMResult
Asynchronously pass a sequence of prompts to a model and return generations.
This method should make use of batched calls for models that expose a batched API.
Use this method when you want to:
- Take advantage of batched calls,
- Need more output from the model than just the top generated value,
- Are building chains that are agnostic to the underlying language model type (e.g., pure text completion models vs chat models).
| PARAMETER | DESCRIPTION |
|---|---|
messages
|
List of list of messages.
TYPE:
|
stop
|
Stop words to use when generating. Model output is cut off at the first occurrence of any of these substrings. |
callbacks
|
Used for executing additional functionality, such as logging or streaming, throughout generation.
TYPE:
|
tags
|
The tags to apply. |
metadata
|
The metadata to apply. |
run_name
|
The name of the run.
TYPE:
|
run_id
|
The ID of the run.
TYPE:
|
**kwargs
|
Arbitrary additional keyword arguments. These are usually passed to the model provider API call.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
LLMResult
|
An |
initialize_client
¶
initialize_client() -> AzureAIChatCompletionsModel
Initialize the Azure AI model inference client.
bind_tools
¶
bind_tools(
tools: Sequence[Dict[str, Any] | type | Callable | BaseTool],
*,
tool_choice: str | None = None,
**kwargs: Any,
) -> Runnable[LanguageModelInput, AIMessage]
Bind tool-like objects to this chat model.
| PARAMETER | DESCRIPTION |
|---|---|
tools
|
A list of tool definitions to bind to this chat model.
Supports any tool definition handled by
:meth: |
tool_choice
|
Which tool to require the model to call.
Must be the name of the single provided function or
"auto" to automatically determine which function to call
(if any), or a dict of the form:
{"type": "function", "function": {"name": <
TYPE:
|
kwargs
|
Any additional parameters are passed directly to
TYPE:
|
with_structured_output
¶
with_structured_output(
schema: dict | type,
method: Literal[
"function_calling", "json_mode", "json_schema"
] = "function_calling",
strict: bool | None = None,
*,
include_raw: bool = False,
**kwargs: Any,
) -> Runnable[LanguageModelInput, dict | BaseModel]
Model wrapper that returns outputs formatted to match the given schema.
| PARAMETER | DESCRIPTION |
|---|---|
schema
|
The schema to use for the output. If a pydantic model is provided, it will be used as the output type. If a dict is provided, it will be used as the schema for the output. |
method
|
The method to use for structured output. Can be "function_calling", "json_mode", or "json_schema".
TYPE:
|
strict
|
Whether to enforce strict mode for "json_schema".
TYPE:
|
include_raw
|
Whether to include the raw response from the model in the output.
TYPE:
|
kwargs
|
Any additional parameters are passed directly to
TYPE:
|
get_lc_namespace
classmethod
¶
Get the namespace of the langchain object.
aclose
async
¶
Close the async client to prevent unclosed session warnings.
This method should be called to properly clean up HTTP connections when using async operations.
langchain_azure_ai.embeddings
¶
Embedding model for Azure AI.
AzureAIEmbeddingsModel
¶
Bases: ModelInferenceService, Embeddings
Azure AI model inference for embeddings.
Examples:
from langchain_azure_ai.embeddings import AzureAIEmbeddingsModel
embed_model = AzureAIEmbeddingsModel(
endpoint="https://[your-endpoint].inference.ai.azure.com",
credential="your-api-key",
)
If your endpoint supports multiple models, indicate the parameter model_name:
from langchain_azure_ai.embeddings import AzureAIEmbeddingsModel
embed_model = AzureAIEmbeddingsModel(
endpoint="https://[your-service].services.ai.azure.com/models",
credential="your-api-key",
model="cohere-embed-v3-multilingual"
)
Troubleshooting:
To diagnostic issues with the model, you can enable debug logging:
import sys
import logging
from langchain_azure_ai.embeddings import AzureAIEmbeddingsModel
logger = logging.getLogger("azure")
# Set the desired logging level.
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler(stream=sys.stdout)
logger.addHandler(handler)
model = AzureAIEmbeddingsModel(
endpoint="https://[your-service].services.ai.azure.com/models",
credential="your-api-key",
model="cohere-embed-v3-multilingual",
client_kwargs={ "logging_enable": True }
)
| METHOD | DESCRIPTION |
|---|---|
validate_environment |
Validate that required values are present in the environment. |
initialize_client |
Initialize the Azure AI model inference client. |
embed_documents |
Embed search docs. |
embed_query |
Embed query text. |
aembed_documents |
Asynchronous Embed search docs. |
aembed_query |
Asynchronous Embed query text. |
project_endpoint
class-attribute
instance-attribute
¶
project_endpoint: str | None = None
The project endpoint associated with the AI project. If this is specified,
then the endpoint parameter becomes optional and credential has to be of type
TokenCredential.
endpoint
class-attribute
instance-attribute
¶
endpoint: str | None = None
The endpoint of the specific service to connect to. If you are connecting to a model, use the URL of the model deployment.
credential
class-attribute
instance-attribute
¶
credential: str | AzureKeyCredential | TokenCredential | None = None
The API key or credential to use to connect to the service. If using a project
endpoint, this must be of type TokenCredential since only Microsoft EntraID is
supported.
api_version
class-attribute
instance-attribute
¶
api_version: str | None = None
The API version to use with Azure. If None, the default version is used.
client_kwargs
class-attribute
instance-attribute
¶
Additional keyword arguments to pass to the client.
service
class-attribute
instance-attribute
¶
service: Literal['inference'] = 'inference'
The type of service to connect to. For Inference Services, use 'inference'.
model_name
class-attribute
instance-attribute
¶
The name of the model to use for inference, if the endpoint is running more than one model. If not, this parameter is ignored.
embed_batch_size
class-attribute
instance-attribute
¶
embed_batch_size: int = 1024
The batch size for embedding requests. The default is 1024.
dimensions
class-attribute
instance-attribute
¶
dimensions: int | None = None
The number of dimensions in the embeddings to generate. If None, the model's default is used.
model_kwargs
class-attribute
instance-attribute
¶
Additional kwargs model parameters.
validate_environment
¶
Validate that required values are present in the environment.
initialize_client
¶
initialize_client() -> AzureAIEmbeddingsModel
Initialize the Azure AI model inference client.
embed_documents
¶
embed_query
¶
aembed_documents
async
¶
langchain_azure_ai.retrievers
¶
Retrievers provide an interface to search and retrieve relevant documents from a data source.
Retrievers abstract the logic of querying underlying data stores (such as vector stores, search engines, or databases) and returning documents most relevant to a user's query. They are commonly used to power search, question answering, and RAG (Retrieval-Augmented Generation) workflows.
Class hierarchy:
Main helpers:
AzureAISearchRetriever
¶
Bases: BaseRetriever
Azure AI Search service retriever.
Setup:
See here for more detail: https://python.langchain.com/docs/integrations/retrievers/azure_ai_search/
We will need to install the below dependencies and set the required environment variables:
pip install -U azure-search-documents
export AZURE_AI_SEARCH_SERVICE_NAME="<YOUR_SEARCH_SERVICE_NAME>"
export AZURE_AI_SEARCH_INDEX_NAME="<YOUR_SEARCH_INDEX_NAME>"
export AZURE_AI_SEARCH_API_KEY="<YOUR_API_KEY>"
or
Key init args
content_key: str top_k: int index_name: str
Instantiate:
from langchain_community.retrievers import AzureAISearchRetriever
retriever = AzureAISearchRetriever(
content_key="content", top_k=1, index_name="langchain-vector-demo"
)
Usage:
Use within a chain:
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import AzureChatOpenAI
prompt = ChatPromptTemplate.from_template(
\"\"\"Answer the question based only on the context provided.
Context: {context}
Question: {question}\"\"\"
)
llm = AzureChatOpenAI(azure_deployment="gpt-35-turbo")
def format_docs(docs):
return "\\n\\n".join(doc.page_content for doc in docs)
chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
chain.invoke("...")
| METHOD | DESCRIPTION |
|---|---|
get_name |
Get the name of the |
get_input_schema |
Get a Pydantic model that can be used to validate input to the |
get_input_jsonschema |
Get a JSON schema that represents the input to the |
get_output_schema |
Get a Pydantic model that can be used to validate output to the |
get_output_jsonschema |
Get a JSON schema that represents the output of the |
config_schema |
The type of config this |
get_config_jsonschema |
Get a JSON schema that represents the config of the |
get_graph |
Return a graph representation of this |
get_prompts |
Return a list of prompts used by this |
__or__ |
Runnable "or" operator. |
__ror__ |
Runnable "reverse-or" operator. |
pipe |
Pipe |
pick |
Pick keys from the output |
assign |
Assigns new fields to the |
invoke |
Invoke the retriever to get relevant documents. |
ainvoke |
Asynchronously invoke the retriever to get relevant documents. |
batch |
Default implementation runs invoke in parallel using a thread pool executor. |
batch_as_completed |
Run |
abatch |
Default implementation runs |
abatch_as_completed |
Run |
stream |
Default implementation of |
astream |
Default implementation of |
astream_log |
Stream all output from a |
astream_events |
Generate a stream of events. |
transform |
Transform inputs to outputs. |
atransform |
Transform inputs to outputs. |
bind |
Bind arguments to a |
with_config |
Bind config to a |
with_listeners |
Bind lifecycle listeners to a |
with_alisteners |
Bind async lifecycle listeners to a |
with_types |
Bind input and output types to a |
with_retry |
Create a new |
map |
Return a new |
with_fallbacks |
Add fallbacks to a |
as_tool |
Create a |
__init__ |
|
is_lc_serializable |
Is this class serializable? |
get_lc_namespace |
Get the namespace of the LangChain object. |
lc_id |
Return a unique identifier for this class for serialization purposes. |
to_json |
Serialize the |
to_json_not_implemented |
Serialize a "not implemented" object. |
configurable_fields |
Configure particular |
configurable_alternatives |
Configure alternatives for |
validate_environment |
Validate that service name, index name and api key exists in environment. |
name
class-attribute
instance-attribute
¶
name: str | None = None
The name of the Runnable. Used for debugging and tracing.
InputType
property
¶
InputType: type[Input]
Input type.
The type of input this Runnable accepts specified as a type annotation.
| RAISES | DESCRIPTION |
|---|---|
TypeError
|
If the input type cannot be inferred. |
OutputType
property
¶
OutputType: type[Output]
Output Type.
The type of output this Runnable produces specified as a type annotation.
| RAISES | DESCRIPTION |
|---|---|
TypeError
|
If the output type cannot be inferred. |
input_schema
property
¶
The type of input this Runnable accepts specified as a Pydantic model.
output_schema
property
¶
Output schema.
The type of output this Runnable produces specified as a Pydantic model.
config_specs
property
¶
config_specs: list[ConfigurableFieldSpec]
List configurable fields for this Runnable.
lc_secrets
property
¶
A map of constructor argument names to secret ids.
For example, {"openai_api_key": "OPENAI_API_KEY"}
lc_attributes
property
¶
lc_attributes: dict
List of attribute names that should be included in the serialized kwargs.
These attributes must be accepted by the constructor.
Default is an empty dictionary.
tags
class-attribute
instance-attribute
¶
Optional list of tags associated with the retriever.
These tags will be associated with each call to this retriever,
and passed as arguments to the handlers defined in callbacks.
You can use these to eg identify a specific instance of a retriever with its use case.
metadata
class-attribute
instance-attribute
¶
Optional metadata associated with the retriever.
This metadata will be associated with each call to this retriever,
and passed as arguments to the handlers defined in callbacks.
You can use these to eg identify a specific instance of a retriever with its use case.
service_name
class-attribute
instance-attribute
¶
service_name: str = ''
Name of Azure AI Search service
index_name
class-attribute
instance-attribute
¶
index_name: str = ''
Name of Index inside Azure AI Search service
api_key
class-attribute
instance-attribute
¶
api_key: str = ''
API Key. Both Admin and Query keys work, but for reading data it's recommended to use a Query key.
aiosession
class-attribute
instance-attribute
¶
ClientSession, in case we want to reuse connection for better performance.
azure_ad_token
class-attribute
instance-attribute
¶
azure_ad_token: str = ''
Your Azure Active Directory token.
Automatically inferred from env var AZURE_AI_SEARCH_AD_TOKEN if not provided.
For more: https://www.microsoft.com/en-us/security/business/identity-access/microsoft-entra-id.
content_key
class-attribute
instance-attribute
¶
content_key: str = 'content'
Key in a retrieved result to set as the Document page_content.
top_k
class-attribute
instance-attribute
¶
top_k: int | None = None
Number of results to retrieve. Set to None to retrieve all results.
filter
class-attribute
instance-attribute
¶
filter: str | None = None
OData $filter expression to apply to the search query.
get_name
¶
get_input_schema
¶
get_input_schema(config: RunnableConfig | None = None) -> type[BaseModel]
Get a Pydantic model that can be used to validate input to the Runnable.
Runnable objects that leverage the configurable_fields and
configurable_alternatives methods will have a dynamic input schema that
depends on which configuration the Runnable is invoked with.
This method allows to get an input schema for a specific configuration.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
A config to use when generating the schema.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
type[BaseModel]
|
A Pydantic model that can be used to validate input. |
get_input_jsonschema
¶
get_input_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]
Get a JSON schema that represents the input to the Runnable.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
A config to use when generating the schema.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
dict[str, Any]
|
A JSON schema that represents the input to the |
Example
Added in langchain-core 0.3.0
get_output_schema
¶
get_output_schema(config: RunnableConfig | None = None) -> type[BaseModel]
Get a Pydantic model that can be used to validate output to the Runnable.
Runnable objects that leverage the configurable_fields and
configurable_alternatives methods will have a dynamic output schema that
depends on which configuration the Runnable is invoked with.
This method allows to get an output schema for a specific configuration.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
A config to use when generating the schema.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
type[BaseModel]
|
A Pydantic model that can be used to validate output. |
get_output_jsonschema
¶
get_output_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]
Get a JSON schema that represents the output of the Runnable.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
A config to use when generating the schema.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
dict[str, Any]
|
A JSON schema that represents the output of the |
Example
Added in langchain-core 0.3.0
config_schema
¶
The type of config this Runnable accepts specified as a Pydantic model.
To mark a field as configurable, see the configurable_fields
and configurable_alternatives methods.
| PARAMETER | DESCRIPTION |
|---|---|
include
|
A list of fields to include in the config schema. |
| RETURNS | DESCRIPTION |
|---|---|
type[BaseModel]
|
A Pydantic model that can be used to validate config. |
get_config_jsonschema
¶
get_graph
¶
get_graph(config: RunnableConfig | None = None) -> Graph
Return a graph representation of this Runnable.
get_prompts
¶
get_prompts(config: RunnableConfig | None = None) -> list[BasePromptTemplate]
Return a list of prompts used by this Runnable.
__or__
¶
__or__(
other: Runnable[Any, Other]
| Callable[[Iterator[Any]], Iterator[Other]]
| Callable[[AsyncIterator[Any]], AsyncIterator[Other]]
| Callable[[Any], Other]
| Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any],
) -> RunnableSerializable[Input, Other]
Runnable "or" operator.
Compose this Runnable with another object to create a
RunnableSequence.
| PARAMETER | DESCRIPTION |
|---|---|
other
|
Another
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Input, Other]
|
A new |
__ror__
¶
__ror__(
other: Runnable[Other, Any]
| Callable[[Iterator[Other]], Iterator[Any]]
| Callable[[AsyncIterator[Other]], AsyncIterator[Any]]
| Callable[[Other], Any]
| Mapping[str, Runnable[Other, Any] | Callable[[Other], Any] | Any],
) -> RunnableSerializable[Other, Output]
Runnable "reverse-or" operator.
Compose this Runnable with another object to create a
RunnableSequence.
| PARAMETER | DESCRIPTION |
|---|---|
other
|
Another
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Other, Output]
|
A new |
pipe
¶
pipe(
*others: Runnable[Any, Other] | Callable[[Any], Other], name: str | None = None
) -> RunnableSerializable[Input, Other]
Pipe Runnable objects.
Compose this Runnable with Runnable-like objects to make a
RunnableSequence.
Equivalent to RunnableSequence(self, *others) or self | others[0] | ...
Example
from langchain_core.runnables import RunnableLambda
def add_one(x: int) -> int:
return x + 1
def mul_two(x: int) -> int:
return x * 2
runnable_1 = RunnableLambda(add_one)
runnable_2 = RunnableLambda(mul_two)
sequence = runnable_1.pipe(runnable_2)
# Or equivalently:
# sequence = runnable_1 | runnable_2
# sequence = RunnableSequence(first=runnable_1, last=runnable_2)
sequence.invoke(1)
await sequence.ainvoke(1)
# -> 4
sequence.batch([1, 2, 3])
await sequence.abatch([1, 2, 3])
# -> [4, 6, 8]
| PARAMETER | DESCRIPTION |
|---|---|
*others
|
Other
TYPE:
|
name
|
An optional name for the resulting
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Input, Other]
|
A new |
pick
¶
Pick keys from the output dict of this Runnable.
Pick a single key:
import json
from langchain_core.runnables import RunnableLambda, RunnableMap
as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)
chain = RunnableMap(str=as_str, json=as_json)
chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3]}
json_only_chain = chain.pick("json")
json_only_chain.invoke("[1, 2, 3]")
# -> [1, 2, 3]
Pick a list of keys:
from typing import Any
import json
from langchain_core.runnables import RunnableLambda, RunnableMap
as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)
def as_bytes(x: Any) -> bytes:
return bytes(x, "utf-8")
chain = RunnableMap(str=as_str, json=as_json, bytes=RunnableLambda(as_bytes))
chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3], "bytes": b"[1, 2, 3]"}
json_and_bytes_chain = chain.pick(["json", "bytes"])
json_and_bytes_chain.invoke("[1, 2, 3]")
# -> {"json": [1, 2, 3], "bytes": b"[1, 2, 3]"}
| PARAMETER | DESCRIPTION |
|---|---|
keys
|
A key or list of keys to pick from the output dict. |
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Any, Any]
|
a new |
assign
¶
assign(
**kwargs: Runnable[dict[str, Any], Any]
| Callable[[dict[str, Any]], Any]
| Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]],
) -> RunnableSerializable[Any, Any]
Assigns new fields to the dict output of this Runnable.
from langchain_core.language_models.fake import FakeStreamingListLLM
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import SystemMessagePromptTemplate
from langchain_core.runnables import Runnable
from operator import itemgetter
prompt = (
SystemMessagePromptTemplate.from_template("You are a nice assistant.")
+ "{question}"
)
model = FakeStreamingListLLM(responses=["foo-lish"])
chain: Runnable = prompt | model | {"str": StrOutputParser()}
chain_with_assign = chain.assign(hello=itemgetter("str") | model)
print(chain_with_assign.input_schema.model_json_schema())
# {'title': 'PromptInput', 'type': 'object', 'properties':
{'question': {'title': 'Question', 'type': 'string'}}}
print(chain_with_assign.output_schema.model_json_schema())
# {'title': 'RunnableSequenceOutput', 'type': 'object', 'properties':
{'str': {'title': 'Str',
'type': 'string'}, 'hello': {'title': 'Hello', 'type': 'string'}}}
| PARAMETER | DESCRIPTION |
|---|---|
**kwargs
|
A mapping of keys to
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Any, Any]
|
A new |
invoke
¶
invoke(
input: str, config: RunnableConfig | None = None, **kwargs: Any
) -> list[Document]
Invoke the retriever to get relevant documents.
Main entry point for synchronous retriever invocations.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The query string.
TYPE:
|
config
|
Configuration for the retriever.
TYPE:
|
**kwargs
|
Additional arguments to pass to the retriever.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of relevant documents. |
Examples:
ainvoke
async
¶
ainvoke(
input: str, config: RunnableConfig | None = None, **kwargs: Any
) -> list[Document]
Asynchronously invoke the retriever to get relevant documents.
Main entry point for asynchronous retriever invocations.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The query string.
TYPE:
|
config
|
Configuration for the retriever.
TYPE:
|
**kwargs
|
Additional arguments to pass to the retriever.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of relevant documents. |
Examples:
batch
¶
batch(
inputs: list[Input],
config: RunnableConfig | list[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any | None,
) -> list[Output]
Default implementation runs invoke in parallel using a thread pool executor.
The default implementation of batch works well for IO bound runnables.
Subclasses must override this method if they can batch more efficiently;
e.g., if the underlying Runnable uses an API which supports a batch mode.
| PARAMETER | DESCRIPTION |
|---|---|
inputs
|
A list of inputs to the
TYPE:
|
config
|
A config to use when invoking the Please refer to
TYPE:
|
return_exceptions
|
Whether to return exceptions instead of raising them.
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Output]
|
A list of outputs from the |
batch_as_completed
¶
batch_as_completed(
inputs: Sequence[Input],
config: RunnableConfig | Sequence[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any | None,
) -> Iterator[tuple[int, Output | Exception]]
Run invoke in parallel on a list of inputs.
Yields results as they complete.
| PARAMETER | DESCRIPTION |
|---|---|
inputs
|
A list of inputs to the
TYPE:
|
config
|
A config to use when invoking the The config supports standard keys like Please refer to
TYPE:
|
return_exceptions
|
Whether to return exceptions instead of raising them.
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
tuple[int, Output | Exception]
|
Tuples of the index of the input and the output from the |
abatch
async
¶
abatch(
inputs: list[Input],
config: RunnableConfig | list[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any | None,
) -> list[Output]
Default implementation runs ainvoke in parallel using asyncio.gather.
The default implementation of batch works well for IO bound runnables.
Subclasses must override this method if they can batch more efficiently;
e.g., if the underlying Runnable uses an API which supports a batch mode.
| PARAMETER | DESCRIPTION |
|---|---|
inputs
|
A list of inputs to the
TYPE:
|
config
|
A config to use when invoking the The config supports standard keys like Please refer to
TYPE:
|
return_exceptions
|
Whether to return exceptions instead of raising them.
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Output]
|
A list of outputs from the |
abatch_as_completed
async
¶
abatch_as_completed(
inputs: Sequence[Input],
config: RunnableConfig | Sequence[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any | None,
) -> AsyncIterator[tuple[int, Output | Exception]]
Run ainvoke in parallel on a list of inputs.
Yields results as they complete.
| PARAMETER | DESCRIPTION |
|---|---|
inputs
|
A list of inputs to the
TYPE:
|
config
|
A config to use when invoking the The config supports standard keys like Please refer to
TYPE:
|
return_exceptions
|
Whether to return exceptions instead of raising them.
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[tuple[int, Output | Exception]]
|
A tuple of the index of the input and the output from the |
stream
¶
stream(
input: Input, config: RunnableConfig | None = None, **kwargs: Any | None
) -> Iterator[Output]
Default implementation of stream, which calls invoke.
Subclasses must override this method if they support streaming output.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
Output
|
The output of the |
astream
async
¶
astream(
input: Input, config: RunnableConfig | None = None, **kwargs: Any | None
) -> AsyncIterator[Output]
Default implementation of astream, which calls ainvoke.
Subclasses must override this method if they support streaming output.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[Output]
|
The output of the |
astream_log
async
¶
astream_log(
input: Any,
config: RunnableConfig | None = None,
*,
diff: bool = True,
with_streamed_output_list: bool = True,
include_names: Sequence[str] | None = None,
include_types: Sequence[str] | None = None,
include_tags: Sequence[str] | None = None,
exclude_names: Sequence[str] | None = None,
exclude_types: Sequence[str] | None = None,
exclude_tags: Sequence[str] | None = None,
**kwargs: Any,
) -> AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]
Stream all output from a Runnable, as reported to the callback system.
This includes all inner runs of LLMs, Retrievers, Tools, etc.
Output is streamed as Log objects, which include a list of Jsonpatch ops that describe how the state of the run has changed in each step, and the final state of the run.
The Jsonpatch ops can be applied in order to construct state.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
diff
|
Whether to yield diffs between each step or the current state.
TYPE:
|
with_streamed_output_list
|
Whether to yield the
TYPE:
|
include_names
|
Only include logs with these names. |
include_types
|
Only include logs with these types. |
include_tags
|
Only include logs with these tags. |
exclude_names
|
Exclude logs with these names. |
exclude_types
|
Exclude logs with these types. |
exclude_tags
|
Exclude logs with these tags. |
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]
|
A |
astream_events
async
¶
astream_events(
input: Any,
config: RunnableConfig | None = None,
*,
version: Literal["v1", "v2"] = "v2",
include_names: Sequence[str] | None = None,
include_types: Sequence[str] | None = None,
include_tags: Sequence[str] | None = None,
exclude_names: Sequence[str] | None = None,
exclude_types: Sequence[str] | None = None,
exclude_tags: Sequence[str] | None = None,
**kwargs: Any,
) -> AsyncIterator[StreamEvent]
Generate a stream of events.
Use to create an iterator over StreamEvent that provide real-time information
about the progress of the Runnable, including StreamEvent from intermediate
results.
A StreamEvent is a dictionary with the following schema:
event: Event names are of the format:on_[runnable_type]_(start|stream|end).name: The name of theRunnablethat generated the event.run_id: Randomly generated ID associated with the given execution of theRunnablethat emitted the event. A childRunnablethat gets invoked as part of the execution of a parentRunnableis assigned its own unique ID.parent_ids: The IDs of the parent runnables that generated the event. The rootRunnablewill have an empty list. The order of the parent IDs is from the root to the immediate parent. Only available for v2 version of the API. The v1 version of the API will return an empty list.tags: The tags of theRunnablethat generated the event.metadata: The metadata of theRunnablethat generated the event.data: The data associated with the event. The contents of this field depend on the type of event. See the table below for more details.
Below is a table that illustrates some events that might be emitted by various chains. Metadata fields have been omitted from the table for brevity. Chain definitions have been included after the table.
Note
This reference table is for the v2 version of the schema.
| event | name | chunk | input | output |
|---|---|---|---|---|
on_chat_model_start |
'[model name]' |
{"messages": [[SystemMessage, HumanMessage]]} |
||
on_chat_model_stream |
'[model name]' |
AIMessageChunk(content="hello") |
||
on_chat_model_end |
'[model name]' |
{"messages": [[SystemMessage, HumanMessage]]} |
AIMessageChunk(content="hello world") |
|
on_llm_start |
'[model name]' |
{'input': 'hello'} |
||
on_llm_stream |
'[model name]' |
'Hello' |
||
on_llm_end |
'[model name]' |
'Hello human!' |
||
on_chain_start |
'format_docs' |
|||
on_chain_stream |
'format_docs' |
'hello world!, goodbye world!' |
||
on_chain_end |
'format_docs' |
[Document(...)] |
'hello world!, goodbye world!' |
|
on_tool_start |
'some_tool' |
{"x": 1, "y": "2"} |
||
on_tool_end |
'some_tool' |
{"x": 1, "y": "2"} |
||
on_retriever_start |
'[retriever name]' |
{"query": "hello"} |
||
on_retriever_end |
'[retriever name]' |
{"query": "hello"} |
[Document(...), ..] |
|
on_prompt_start |
'[template_name]' |
{"question": "hello"} |
||
on_prompt_end |
'[template_name]' |
{"question": "hello"} |
ChatPromptValue(messages: [SystemMessage, ...]) |
In addition to the standard events, users can also dispatch custom events (see example below).
Custom events will be only be surfaced with in the v2 version of the API!
A custom event has following format:
| Attribute | Type | Description |
|---|---|---|
name |
str |
A user defined name for the event. |
data |
Any |
The data associated with the event. This can be anything, though we suggest making it JSON serializable. |
Here are declarations associated with the standard events shown above:
format_docs:
def format_docs(docs: list[Document]) -> str:
'''Format the docs.'''
return ", ".join([doc.page_content for doc in docs])
format_docs = RunnableLambda(format_docs)
some_tool:
prompt:
template = ChatPromptTemplate.from_messages(
[
("system", "You are Cat Agent 007"),
("human", "{question}"),
]
).with_config({"run_name": "my_template", "tags": ["my_template"]})
For instance:
from langchain_core.runnables import RunnableLambda
async def reverse(s: str) -> str:
return s[::-1]
chain = RunnableLambda(func=reverse)
events = [event async for event in chain.astream_events("hello", version="v2")]
# Will produce the following events
# (run_id, and parent_ids has been omitted for brevity):
[
{
"data": {"input": "hello"},
"event": "on_chain_start",
"metadata": {},
"name": "reverse",
"tags": [],
},
{
"data": {"chunk": "olleh"},
"event": "on_chain_stream",
"metadata": {},
"name": "reverse",
"tags": [],
},
{
"data": {"output": "olleh"},
"event": "on_chain_end",
"metadata": {},
"name": "reverse",
"tags": [],
},
]
from langchain_core.callbacks.manager import (
adispatch_custom_event,
)
from langchain_core.runnables import RunnableLambda, RunnableConfig
import asyncio
async def slow_thing(some_input: str, config: RunnableConfig) -> str:
"""Do something that takes a long time."""
await asyncio.sleep(1) # Placeholder for some slow operation
await adispatch_custom_event(
"progress_event",
{"message": "Finished step 1 of 3"},
config=config # Must be included for python < 3.10
)
await asyncio.sleep(1) # Placeholder for some slow operation
await adispatch_custom_event(
"progress_event",
{"message": "Finished step 2 of 3"},
config=config # Must be included for python < 3.10
)
await asyncio.sleep(1) # Placeholder for some slow operation
return "Done"
slow_thing = RunnableLambda(slow_thing)
async for event in slow_thing.astream_events("some_input", version="v2"):
print(event)
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
version
|
The version of the schema to use either
TYPE:
|
include_names
|
Only include events from |
include_types
|
Only include events from |
include_tags
|
Only include events from |
exclude_names
|
Exclude events from |
exclude_types
|
Exclude events from |
exclude_tags
|
Exclude events from |
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[StreamEvent]
|
An async stream of |
| RAISES | DESCRIPTION |
|---|---|
NotImplementedError
|
If the version is not |
transform
¶
transform(
input: Iterator[Input], config: RunnableConfig | None = None, **kwargs: Any | None
) -> Iterator[Output]
Transform inputs to outputs.
Default implementation of transform, which buffers input and calls astream.
Subclasses must override this method if they can start producing output while input is still being generated.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
An iterator of inputs to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
Output
|
The output of the |
atransform
async
¶
atransform(
input: AsyncIterator[Input],
config: RunnableConfig | None = None,
**kwargs: Any | None,
) -> AsyncIterator[Output]
Transform inputs to outputs.
Default implementation of atransform, which buffers input and calls astream.
Subclasses must override this method if they can start producing output while input is still being generated.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
An async iterator of inputs to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[Output]
|
The output of the |
bind
¶
Bind arguments to a Runnable, returning a new Runnable.
Useful when a Runnable in a chain requires an argument that is not
in the output of the previous Runnable or included in the user input.
| PARAMETER | DESCRIPTION |
|---|---|
**kwargs
|
The arguments to bind to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser
model = ChatOllama(model="llama3.1")
# Without bind
chain = model | StrOutputParser()
chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two three four five.'
# With bind
chain = model.bind(stop=["three"]) | StrOutputParser()
chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two'
with_config
¶
with_config(
config: RunnableConfig | None = None, **kwargs: Any
) -> Runnable[Input, Output]
Bind config to a Runnable, returning a new Runnable.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
The config to bind to the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
with_listeners
¶
with_listeners(
*,
on_start: Callable[[Run], None]
| Callable[[Run, RunnableConfig], None]
| None = None,
on_end: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None = None,
on_error: Callable[[Run], None]
| Callable[[Run, RunnableConfig], None]
| None = None,
) -> Runnable[Input, Output]
Bind lifecycle listeners to a Runnable, returning a new Runnable.
The Run object contains information about the run, including its id,
type, input, output, error, start_time, end_time, and
any tags or metadata added to the run.
| PARAMETER | DESCRIPTION |
|---|---|
on_start
|
Called before the
TYPE:
|
on_end
|
Called after the
TYPE:
|
on_error
|
Called if the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_core.runnables import RunnableLambda
from langchain_core.tracers.schemas import Run
import time
def test_runnable(time_to_sleep: int):
time.sleep(time_to_sleep)
def fn_start(run_obj: Run):
print("start_time:", run_obj.start_time)
def fn_end(run_obj: Run):
print("end_time:", run_obj.end_time)
chain = RunnableLambda(test_runnable).with_listeners(
on_start=fn_start, on_end=fn_end
)
chain.invoke(2)
with_alisteners
¶
with_alisteners(
*,
on_start: AsyncListener | None = None,
on_end: AsyncListener | None = None,
on_error: AsyncListener | None = None,
) -> Runnable[Input, Output]
Bind async lifecycle listeners to a Runnable.
Returns a new Runnable.
The Run object contains information about the run, including its id,
type, input, output, error, start_time, end_time, and
any tags or metadata added to the run.
| PARAMETER | DESCRIPTION |
|---|---|
on_start
|
Called asynchronously before the
TYPE:
|
on_end
|
Called asynchronously after the
TYPE:
|
on_error
|
Called asynchronously if the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_core.runnables import RunnableLambda, Runnable
from datetime import datetime, timezone
import time
import asyncio
def format_t(timestamp: float) -> str:
return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat()
async def test_runnable(time_to_sleep: int):
print(f"Runnable[{time_to_sleep}s]: starts at {format_t(time.time())}")
await asyncio.sleep(time_to_sleep)
print(f"Runnable[{time_to_sleep}s]: ends at {format_t(time.time())}")
async def fn_start(run_obj: Runnable):
print(f"on start callback starts at {format_t(time.time())}")
await asyncio.sleep(3)
print(f"on start callback ends at {format_t(time.time())}")
async def fn_end(run_obj: Runnable):
print(f"on end callback starts at {format_t(time.time())}")
await asyncio.sleep(2)
print(f"on end callback ends at {format_t(time.time())}")
runnable = RunnableLambda(test_runnable).with_alisteners(
on_start=fn_start, on_end=fn_end
)
async def concurrent_runs():
await asyncio.gather(runnable.ainvoke(2), runnable.ainvoke(3))
asyncio.run(concurrent_runs())
# Result:
# on start callback starts at 2025-03-01T07:05:22.875378+00:00
# on start callback starts at 2025-03-01T07:05:22.875495+00:00
# on start callback ends at 2025-03-01T07:05:25.878862+00:00
# on start callback ends at 2025-03-01T07:05:25.878947+00:00
# Runnable[2s]: starts at 2025-03-01T07:05:25.879392+00:00
# Runnable[3s]: starts at 2025-03-01T07:05:25.879804+00:00
# Runnable[2s]: ends at 2025-03-01T07:05:27.881998+00:00
# on end callback starts at 2025-03-01T07:05:27.882360+00:00
# Runnable[3s]: ends at 2025-03-01T07:05:28.881737+00:00
# on end callback starts at 2025-03-01T07:05:28.882428+00:00
# on end callback ends at 2025-03-01T07:05:29.883893+00:00
# on end callback ends at 2025-03-01T07:05:30.884831+00:00
with_types
¶
with_types(
*, input_type: type[Input] | None = None, output_type: type[Output] | None = None
) -> Runnable[Input, Output]
Bind input and output types to a Runnable, returning a new Runnable.
| PARAMETER | DESCRIPTION |
|---|---|
input_type
|
The input type to bind to the
TYPE:
|
output_type
|
The output type to bind to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
with_retry
¶
with_retry(
*,
retry_if_exception_type: tuple[type[BaseException], ...] = (Exception,),
wait_exponential_jitter: bool = True,
exponential_jitter_params: ExponentialJitterParams | None = None,
stop_after_attempt: int = 3,
) -> Runnable[Input, Output]
Create a new Runnable that retries the original Runnable on exceptions.
| PARAMETER | DESCRIPTION |
|---|---|
retry_if_exception_type
|
A tuple of exception types to retry on.
TYPE:
|
wait_exponential_jitter
|
Whether to add jitter to the wait time between retries.
TYPE:
|
stop_after_attempt
|
The maximum number of attempts to make before giving up.
TYPE:
|
exponential_jitter_params
|
Parameters for
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_core.runnables import RunnableLambda
count = 0
def _lambda(x: int) -> None:
global count
count = count + 1
if x == 1:
raise ValueError("x is 1")
else:
pass
runnable = RunnableLambda(_lambda)
try:
runnable.with_retry(
stop_after_attempt=2,
retry_if_exception_type=(ValueError,),
).invoke(1)
except ValueError:
pass
assert count == 2
map
¶
with_fallbacks
¶
with_fallbacks(
fallbacks: Sequence[Runnable[Input, Output]],
*,
exceptions_to_handle: tuple[type[BaseException], ...] = (Exception,),
exception_key: str | None = None,
) -> RunnableWithFallbacks[Input, Output]
Add fallbacks to a Runnable, returning a new Runnable.
The new Runnable will try the original Runnable, and then each fallback
in order, upon failures.
| PARAMETER | DESCRIPTION |
|---|---|
fallbacks
|
A sequence of runnables to try if the original |
exceptions_to_handle
|
A tuple of exception types to handle.
TYPE:
|
exception_key
|
If If If used, the base
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableWithFallbacks[Input, Output]
|
A new |
Example
from typing import Iterator
from langchain_core.runnables import RunnableGenerator
def _generate_immediate_error(input: Iterator) -> Iterator[str]:
raise ValueError()
yield ""
def _generate(input: Iterator) -> Iterator[str]:
yield from "foo bar"
runnable = RunnableGenerator(_generate_immediate_error).with_fallbacks(
[RunnableGenerator(_generate)]
)
print("".join(runnable.stream({}))) # foo bar
| PARAMETER | DESCRIPTION |
|---|---|
fallbacks
|
A sequence of runnables to try if the original |
exceptions_to_handle
|
A tuple of exception types to handle.
TYPE:
|
exception_key
|
If If If used, the base
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableWithFallbacks[Input, Output]
|
A new |
as_tool
¶
as_tool(
args_schema: type[BaseModel] | None = None,
*,
name: str | None = None,
description: str | None = None,
arg_types: dict[str, type] | None = None,
) -> BaseTool
Create a BaseTool from a Runnable.
as_tool will instantiate a BaseTool with a name, description, and
args_schema from a Runnable. Where possible, schemas are inferred
from runnable.get_input_schema.
Alternatively (e.g., if the Runnable takes a dict as input and the specific
dict keys are not typed), the schema can be specified directly with
args_schema.
You can also pass arg_types to just specify the required arguments and their
types.
| PARAMETER | DESCRIPTION |
|---|---|
args_schema
|
The schema for the tool. |
name
|
The name of the tool.
TYPE:
|
description
|
The description of the tool.
TYPE:
|
arg_types
|
A dictionary of argument names to types. |
| RETURNS | DESCRIPTION |
|---|---|
BaseTool
|
A |
Typed dict input:
from typing_extensions import TypedDict
from langchain_core.runnables import RunnableLambda
class Args(TypedDict):
a: int
b: list[int]
def f(x: Args) -> str:
return str(x["a"] * max(x["b"]))
runnable = RunnableLambda(f)
as_tool = runnable.as_tool()
as_tool.invoke({"a": 3, "b": [1, 2]})
dict input, specifying schema via args_schema:
from typing import Any
from pydantic import BaseModel, Field
from langchain_core.runnables import RunnableLambda
def f(x: dict[str, Any]) -> str:
return str(x["a"] * max(x["b"]))
class FSchema(BaseModel):
"""Apply a function to an integer and list of integers."""
a: int = Field(..., description="Integer")
b: list[int] = Field(..., description="List of ints")
runnable = RunnableLambda(f)
as_tool = runnable.as_tool(FSchema)
as_tool.invoke({"a": 3, "b": [1, 2]})
dict input, specifying schema via arg_types:
from typing import Any
from langchain_core.runnables import RunnableLambda
def f(x: dict[str, Any]) -> str:
return str(x["a"] * max(x["b"]))
runnable = RunnableLambda(f)
as_tool = runnable.as_tool(arg_types={"a": int, "b": list[int]})
as_tool.invoke({"a": 3, "b": [1, 2]})
str input:
is_lc_serializable
classmethod
¶
is_lc_serializable() -> bool
Is this class serializable?
By design, even if a class inherits from Serializable, it is not serializable
by default. This is to prevent accidental serialization of objects that should
not be serialized.
| RETURNS | DESCRIPTION |
|---|---|
bool
|
Whether the class is serializable. Default is |
get_lc_namespace
classmethod
¶
lc_id
classmethod
¶
Return a unique identifier for this class for serialization purposes.
The unique identifier is a list of strings that describes the path to the object.
For example, for the class langchain.llms.openai.OpenAI, the id is
["langchain", "llms", "openai", "OpenAI"].
to_json
¶
Serialize the Runnable to JSON.
| RETURNS | DESCRIPTION |
|---|---|
SerializedConstructor | SerializedNotImplemented
|
A JSON-serializable representation of the |
to_json_not_implemented
¶
Serialize a "not implemented" object.
| RETURNS | DESCRIPTION |
|---|---|
SerializedNotImplemented
|
|
configurable_fields
¶
configurable_fields(
**kwargs: AnyConfigurableField,
) -> RunnableSerializable[Input, Output]
Configure particular Runnable fields at runtime.
| PARAMETER | DESCRIPTION |
|---|---|
**kwargs
|
A dictionary of
TYPE:
|
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If a configuration key is not found in the |
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Input, Output]
|
A new |
from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI
model = ChatOpenAI(max_tokens=20).configurable_fields(
max_tokens=ConfigurableField(
id="output_token_number",
name="Max tokens in the output",
description="The maximum number of tokens in the output",
)
)
# max_tokens = 20
print("max_tokens_20: ", model.invoke("tell me something about chess").content)
# max_tokens = 200
print(
"max_tokens_200: ",
model.with_config(configurable={"output_token_number": 200})
.invoke("tell me something about chess")
.content,
)
configurable_alternatives
¶
configurable_alternatives(
which: ConfigurableField,
*,
default_key: str = "default",
prefix_keys: bool = False,
**kwargs: Runnable[Input, Output] | Callable[[], Runnable[Input, Output]],
) -> RunnableSerializable[Input, Output]
Configure alternatives for Runnable objects that can be set at runtime.
| PARAMETER | DESCRIPTION |
|---|---|
which
|
The
TYPE:
|
default_key
|
The default key to use if no alternative is selected.
TYPE:
|
prefix_keys
|
Whether to prefix the keys with the
TYPE:
|
**kwargs
|
A dictionary of keys to
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Input, Output]
|
A new |
from langchain_anthropic import ChatAnthropic
from langchain_core.runnables.utils import ConfigurableField
from langchain_openai import ChatOpenAI
model = ChatAnthropic(
model_name="claude-sonnet-4-5-20250929"
).configurable_alternatives(
ConfigurableField(id="llm"),
default_key="anthropic",
openai=ChatOpenAI(),
)
# uses the default model ChatAnthropic
print(model.invoke("which organization created you?").content)
# uses ChatOpenAI
print(
model.with_config(configurable={"llm": "openai"})
.invoke("which organization created you?")
.content
)
AzureCognitiveSearchRetriever
¶
Bases: AzureAISearchRetriever
Azure Cognitive Search service retriever.
This version of the retriever will soon be depreciated. Please switch to AzureAISearchRetriever.
| METHOD | DESCRIPTION |
|---|---|
get_name |
Get the name of the |
get_input_schema |
Get a Pydantic model that can be used to validate input to the |
get_input_jsonschema |
Get a JSON schema that represents the input to the |
get_output_schema |
Get a Pydantic model that can be used to validate output to the |
get_output_jsonschema |
Get a JSON schema that represents the output of the |
config_schema |
The type of config this |
get_config_jsonschema |
Get a JSON schema that represents the config of the |
get_graph |
Return a graph representation of this |
get_prompts |
Return a list of prompts used by this |
__or__ |
Runnable "or" operator. |
__ror__ |
Runnable "reverse-or" operator. |
pipe |
Pipe |
pick |
Pick keys from the output |
assign |
Assigns new fields to the |
invoke |
Invoke the retriever to get relevant documents. |
ainvoke |
Asynchronously invoke the retriever to get relevant documents. |
batch |
Default implementation runs invoke in parallel using a thread pool executor. |
batch_as_completed |
Run |
abatch |
Default implementation runs |
abatch_as_completed |
Run |
stream |
Default implementation of |
astream |
Default implementation of |
astream_log |
Stream all output from a |
astream_events |
Generate a stream of events. |
transform |
Transform inputs to outputs. |
atransform |
Transform inputs to outputs. |
bind |
Bind arguments to a |
with_config |
Bind config to a |
with_listeners |
Bind lifecycle listeners to a |
with_alisteners |
Bind async lifecycle listeners to a |
with_types |
Bind input and output types to a |
with_retry |
Create a new |
map |
Return a new |
with_fallbacks |
Add fallbacks to a |
as_tool |
Create a |
__init__ |
|
is_lc_serializable |
Is this class serializable? |
get_lc_namespace |
Get the namespace of the LangChain object. |
lc_id |
Return a unique identifier for this class for serialization purposes. |
to_json |
Serialize the |
to_json_not_implemented |
Serialize a "not implemented" object. |
configurable_fields |
Configure particular |
configurable_alternatives |
Configure alternatives for |
validate_environment |
Validate that service name, index name and api key exists in environment. |
name
class-attribute
instance-attribute
¶
name: str | None = None
The name of the Runnable. Used for debugging and tracing.
InputType
property
¶
InputType: type[Input]
Input type.
The type of input this Runnable accepts specified as a type annotation.
| RAISES | DESCRIPTION |
|---|---|
TypeError
|
If the input type cannot be inferred. |
OutputType
property
¶
OutputType: type[Output]
Output Type.
The type of output this Runnable produces specified as a type annotation.
| RAISES | DESCRIPTION |
|---|---|
TypeError
|
If the output type cannot be inferred. |
input_schema
property
¶
The type of input this Runnable accepts specified as a Pydantic model.
output_schema
property
¶
Output schema.
The type of output this Runnable produces specified as a Pydantic model.
config_specs
property
¶
config_specs: list[ConfigurableFieldSpec]
List configurable fields for this Runnable.
lc_secrets
property
¶
A map of constructor argument names to secret ids.
For example, {"openai_api_key": "OPENAI_API_KEY"}
lc_attributes
property
¶
lc_attributes: dict
List of attribute names that should be included in the serialized kwargs.
These attributes must be accepted by the constructor.
Default is an empty dictionary.
tags
class-attribute
instance-attribute
¶
Optional list of tags associated with the retriever.
These tags will be associated with each call to this retriever,
and passed as arguments to the handlers defined in callbacks.
You can use these to eg identify a specific instance of a retriever with its use case.
metadata
class-attribute
instance-attribute
¶
Optional metadata associated with the retriever.
This metadata will be associated with each call to this retriever,
and passed as arguments to the handlers defined in callbacks.
You can use these to eg identify a specific instance of a retriever with its use case.
service_name
class-attribute
instance-attribute
¶
service_name: str = ''
Name of Azure AI Search service
index_name
class-attribute
instance-attribute
¶
index_name: str = ''
Name of Index inside Azure AI Search service
api_key
class-attribute
instance-attribute
¶
api_key: str = ''
API Key. Both Admin and Query keys work, but for reading data it's recommended to use a Query key.
aiosession
class-attribute
instance-attribute
¶
ClientSession, in case we want to reuse connection for better performance.
azure_ad_token
class-attribute
instance-attribute
¶
azure_ad_token: str = ''
Your Azure Active Directory token.
Automatically inferred from env var AZURE_AI_SEARCH_AD_TOKEN if not provided.
For more: https://www.microsoft.com/en-us/security/business/identity-access/microsoft-entra-id.
content_key
class-attribute
instance-attribute
¶
content_key: str = 'content'
Key in a retrieved result to set as the Document page_content.
top_k
class-attribute
instance-attribute
¶
top_k: int | None = None
Number of results to retrieve. Set to None to retrieve all results.
filter
class-attribute
instance-attribute
¶
filter: str | None = None
OData $filter expression to apply to the search query.
get_name
¶
get_input_schema
¶
get_input_schema(config: RunnableConfig | None = None) -> type[BaseModel]
Get a Pydantic model that can be used to validate input to the Runnable.
Runnable objects that leverage the configurable_fields and
configurable_alternatives methods will have a dynamic input schema that
depends on which configuration the Runnable is invoked with.
This method allows to get an input schema for a specific configuration.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
A config to use when generating the schema.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
type[BaseModel]
|
A Pydantic model that can be used to validate input. |
get_input_jsonschema
¶
get_input_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]
Get a JSON schema that represents the input to the Runnable.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
A config to use when generating the schema.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
dict[str, Any]
|
A JSON schema that represents the input to the |
Example
Added in langchain-core 0.3.0
get_output_schema
¶
get_output_schema(config: RunnableConfig | None = None) -> type[BaseModel]
Get a Pydantic model that can be used to validate output to the Runnable.
Runnable objects that leverage the configurable_fields and
configurable_alternatives methods will have a dynamic output schema that
depends on which configuration the Runnable is invoked with.
This method allows to get an output schema for a specific configuration.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
A config to use when generating the schema.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
type[BaseModel]
|
A Pydantic model that can be used to validate output. |
get_output_jsonschema
¶
get_output_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]
Get a JSON schema that represents the output of the Runnable.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
A config to use when generating the schema.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
dict[str, Any]
|
A JSON schema that represents the output of the |
Example
Added in langchain-core 0.3.0
config_schema
¶
The type of config this Runnable accepts specified as a Pydantic model.
To mark a field as configurable, see the configurable_fields
and configurable_alternatives methods.
| PARAMETER | DESCRIPTION |
|---|---|
include
|
A list of fields to include in the config schema. |
| RETURNS | DESCRIPTION |
|---|---|
type[BaseModel]
|
A Pydantic model that can be used to validate config. |
get_config_jsonschema
¶
get_graph
¶
get_graph(config: RunnableConfig | None = None) -> Graph
Return a graph representation of this Runnable.
get_prompts
¶
get_prompts(config: RunnableConfig | None = None) -> list[BasePromptTemplate]
Return a list of prompts used by this Runnable.
__or__
¶
__or__(
other: Runnable[Any, Other]
| Callable[[Iterator[Any]], Iterator[Other]]
| Callable[[AsyncIterator[Any]], AsyncIterator[Other]]
| Callable[[Any], Other]
| Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any],
) -> RunnableSerializable[Input, Other]
Runnable "or" operator.
Compose this Runnable with another object to create a
RunnableSequence.
| PARAMETER | DESCRIPTION |
|---|---|
other
|
Another
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Input, Other]
|
A new |
__ror__
¶
__ror__(
other: Runnable[Other, Any]
| Callable[[Iterator[Other]], Iterator[Any]]
| Callable[[AsyncIterator[Other]], AsyncIterator[Any]]
| Callable[[Other], Any]
| Mapping[str, Runnable[Other, Any] | Callable[[Other], Any] | Any],
) -> RunnableSerializable[Other, Output]
Runnable "reverse-or" operator.
Compose this Runnable with another object to create a
RunnableSequence.
| PARAMETER | DESCRIPTION |
|---|---|
other
|
Another
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Other, Output]
|
A new |
pipe
¶
pipe(
*others: Runnable[Any, Other] | Callable[[Any], Other], name: str | None = None
) -> RunnableSerializable[Input, Other]
Pipe Runnable objects.
Compose this Runnable with Runnable-like objects to make a
RunnableSequence.
Equivalent to RunnableSequence(self, *others) or self | others[0] | ...
Example
from langchain_core.runnables import RunnableLambda
def add_one(x: int) -> int:
return x + 1
def mul_two(x: int) -> int:
return x * 2
runnable_1 = RunnableLambda(add_one)
runnable_2 = RunnableLambda(mul_two)
sequence = runnable_1.pipe(runnable_2)
# Or equivalently:
# sequence = runnable_1 | runnable_2
# sequence = RunnableSequence(first=runnable_1, last=runnable_2)
sequence.invoke(1)
await sequence.ainvoke(1)
# -> 4
sequence.batch([1, 2, 3])
await sequence.abatch([1, 2, 3])
# -> [4, 6, 8]
| PARAMETER | DESCRIPTION |
|---|---|
*others
|
Other
TYPE:
|
name
|
An optional name for the resulting
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Input, Other]
|
A new |
pick
¶
Pick keys from the output dict of this Runnable.
Pick a single key:
import json
from langchain_core.runnables import RunnableLambda, RunnableMap
as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)
chain = RunnableMap(str=as_str, json=as_json)
chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3]}
json_only_chain = chain.pick("json")
json_only_chain.invoke("[1, 2, 3]")
# -> [1, 2, 3]
Pick a list of keys:
from typing import Any
import json
from langchain_core.runnables import RunnableLambda, RunnableMap
as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)
def as_bytes(x: Any) -> bytes:
return bytes(x, "utf-8")
chain = RunnableMap(str=as_str, json=as_json, bytes=RunnableLambda(as_bytes))
chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3], "bytes": b"[1, 2, 3]"}
json_and_bytes_chain = chain.pick(["json", "bytes"])
json_and_bytes_chain.invoke("[1, 2, 3]")
# -> {"json": [1, 2, 3], "bytes": b"[1, 2, 3]"}
| PARAMETER | DESCRIPTION |
|---|---|
keys
|
A key or list of keys to pick from the output dict. |
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Any, Any]
|
a new |
assign
¶
assign(
**kwargs: Runnable[dict[str, Any], Any]
| Callable[[dict[str, Any]], Any]
| Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]],
) -> RunnableSerializable[Any, Any]
Assigns new fields to the dict output of this Runnable.
from langchain_core.language_models.fake import FakeStreamingListLLM
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import SystemMessagePromptTemplate
from langchain_core.runnables import Runnable
from operator import itemgetter
prompt = (
SystemMessagePromptTemplate.from_template("You are a nice assistant.")
+ "{question}"
)
model = FakeStreamingListLLM(responses=["foo-lish"])
chain: Runnable = prompt | model | {"str": StrOutputParser()}
chain_with_assign = chain.assign(hello=itemgetter("str") | model)
print(chain_with_assign.input_schema.model_json_schema())
# {'title': 'PromptInput', 'type': 'object', 'properties':
{'question': {'title': 'Question', 'type': 'string'}}}
print(chain_with_assign.output_schema.model_json_schema())
# {'title': 'RunnableSequenceOutput', 'type': 'object', 'properties':
{'str': {'title': 'Str',
'type': 'string'}, 'hello': {'title': 'Hello', 'type': 'string'}}}
| PARAMETER | DESCRIPTION |
|---|---|
**kwargs
|
A mapping of keys to
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Any, Any]
|
A new |
invoke
¶
invoke(
input: str, config: RunnableConfig | None = None, **kwargs: Any
) -> list[Document]
Invoke the retriever to get relevant documents.
Main entry point for synchronous retriever invocations.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The query string.
TYPE:
|
config
|
Configuration for the retriever.
TYPE:
|
**kwargs
|
Additional arguments to pass to the retriever.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of relevant documents. |
Examples:
ainvoke
async
¶
ainvoke(
input: str, config: RunnableConfig | None = None, **kwargs: Any
) -> list[Document]
Asynchronously invoke the retriever to get relevant documents.
Main entry point for asynchronous retriever invocations.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The query string.
TYPE:
|
config
|
Configuration for the retriever.
TYPE:
|
**kwargs
|
Additional arguments to pass to the retriever.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of relevant documents. |
Examples:
batch
¶
batch(
inputs: list[Input],
config: RunnableConfig | list[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any | None,
) -> list[Output]
Default implementation runs invoke in parallel using a thread pool executor.
The default implementation of batch works well for IO bound runnables.
Subclasses must override this method if they can batch more efficiently;
e.g., if the underlying Runnable uses an API which supports a batch mode.
| PARAMETER | DESCRIPTION |
|---|---|
inputs
|
A list of inputs to the
TYPE:
|
config
|
A config to use when invoking the Please refer to
TYPE:
|
return_exceptions
|
Whether to return exceptions instead of raising them.
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Output]
|
A list of outputs from the |
batch_as_completed
¶
batch_as_completed(
inputs: Sequence[Input],
config: RunnableConfig | Sequence[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any | None,
) -> Iterator[tuple[int, Output | Exception]]
Run invoke in parallel on a list of inputs.
Yields results as they complete.
| PARAMETER | DESCRIPTION |
|---|---|
inputs
|
A list of inputs to the
TYPE:
|
config
|
A config to use when invoking the The config supports standard keys like Please refer to
TYPE:
|
return_exceptions
|
Whether to return exceptions instead of raising them.
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
tuple[int, Output | Exception]
|
Tuples of the index of the input and the output from the |
abatch
async
¶
abatch(
inputs: list[Input],
config: RunnableConfig | list[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any | None,
) -> list[Output]
Default implementation runs ainvoke in parallel using asyncio.gather.
The default implementation of batch works well for IO bound runnables.
Subclasses must override this method if they can batch more efficiently;
e.g., if the underlying Runnable uses an API which supports a batch mode.
| PARAMETER | DESCRIPTION |
|---|---|
inputs
|
A list of inputs to the
TYPE:
|
config
|
A config to use when invoking the The config supports standard keys like Please refer to
TYPE:
|
return_exceptions
|
Whether to return exceptions instead of raising them.
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Output]
|
A list of outputs from the |
abatch_as_completed
async
¶
abatch_as_completed(
inputs: Sequence[Input],
config: RunnableConfig | Sequence[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any | None,
) -> AsyncIterator[tuple[int, Output | Exception]]
Run ainvoke in parallel on a list of inputs.
Yields results as they complete.
| PARAMETER | DESCRIPTION |
|---|---|
inputs
|
A list of inputs to the
TYPE:
|
config
|
A config to use when invoking the The config supports standard keys like Please refer to
TYPE:
|
return_exceptions
|
Whether to return exceptions instead of raising them.
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[tuple[int, Output | Exception]]
|
A tuple of the index of the input and the output from the |
stream
¶
stream(
input: Input, config: RunnableConfig | None = None, **kwargs: Any | None
) -> Iterator[Output]
Default implementation of stream, which calls invoke.
Subclasses must override this method if they support streaming output.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
Output
|
The output of the |
astream
async
¶
astream(
input: Input, config: RunnableConfig | None = None, **kwargs: Any | None
) -> AsyncIterator[Output]
Default implementation of astream, which calls ainvoke.
Subclasses must override this method if they support streaming output.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[Output]
|
The output of the |
astream_log
async
¶
astream_log(
input: Any,
config: RunnableConfig | None = None,
*,
diff: bool = True,
with_streamed_output_list: bool = True,
include_names: Sequence[str] | None = None,
include_types: Sequence[str] | None = None,
include_tags: Sequence[str] | None = None,
exclude_names: Sequence[str] | None = None,
exclude_types: Sequence[str] | None = None,
exclude_tags: Sequence[str] | None = None,
**kwargs: Any,
) -> AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]
Stream all output from a Runnable, as reported to the callback system.
This includes all inner runs of LLMs, Retrievers, Tools, etc.
Output is streamed as Log objects, which include a list of Jsonpatch ops that describe how the state of the run has changed in each step, and the final state of the run.
The Jsonpatch ops can be applied in order to construct state.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
diff
|
Whether to yield diffs between each step or the current state.
TYPE:
|
with_streamed_output_list
|
Whether to yield the
TYPE:
|
include_names
|
Only include logs with these names. |
include_types
|
Only include logs with these types. |
include_tags
|
Only include logs with these tags. |
exclude_names
|
Exclude logs with these names. |
exclude_types
|
Exclude logs with these types. |
exclude_tags
|
Exclude logs with these tags. |
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]
|
A |
astream_events
async
¶
astream_events(
input: Any,
config: RunnableConfig | None = None,
*,
version: Literal["v1", "v2"] = "v2",
include_names: Sequence[str] | None = None,
include_types: Sequence[str] | None = None,
include_tags: Sequence[str] | None = None,
exclude_names: Sequence[str] | None = None,
exclude_types: Sequence[str] | None = None,
exclude_tags: Sequence[str] | None = None,
**kwargs: Any,
) -> AsyncIterator[StreamEvent]
Generate a stream of events.
Use to create an iterator over StreamEvent that provide real-time information
about the progress of the Runnable, including StreamEvent from intermediate
results.
A StreamEvent is a dictionary with the following schema:
event: Event names are of the format:on_[runnable_type]_(start|stream|end).name: The name of theRunnablethat generated the event.run_id: Randomly generated ID associated with the given execution of theRunnablethat emitted the event. A childRunnablethat gets invoked as part of the execution of a parentRunnableis assigned its own unique ID.parent_ids: The IDs of the parent runnables that generated the event. The rootRunnablewill have an empty list. The order of the parent IDs is from the root to the immediate parent. Only available for v2 version of the API. The v1 version of the API will return an empty list.tags: The tags of theRunnablethat generated the event.metadata: The metadata of theRunnablethat generated the event.data: The data associated with the event. The contents of this field depend on the type of event. See the table below for more details.
Below is a table that illustrates some events that might be emitted by various chains. Metadata fields have been omitted from the table for brevity. Chain definitions have been included after the table.
Note
This reference table is for the v2 version of the schema.
| event | name | chunk | input | output |
|---|---|---|---|---|
on_chat_model_start |
'[model name]' |
{"messages": [[SystemMessage, HumanMessage]]} |
||
on_chat_model_stream |
'[model name]' |
AIMessageChunk(content="hello") |
||
on_chat_model_end |
'[model name]' |
{"messages": [[SystemMessage, HumanMessage]]} |
AIMessageChunk(content="hello world") |
|
on_llm_start |
'[model name]' |
{'input': 'hello'} |
||
on_llm_stream |
'[model name]' |
'Hello' |
||
on_llm_end |
'[model name]' |
'Hello human!' |
||
on_chain_start |
'format_docs' |
|||
on_chain_stream |
'format_docs' |
'hello world!, goodbye world!' |
||
on_chain_end |
'format_docs' |
[Document(...)] |
'hello world!, goodbye world!' |
|
on_tool_start |
'some_tool' |
{"x": 1, "y": "2"} |
||
on_tool_end |
'some_tool' |
{"x": 1, "y": "2"} |
||
on_retriever_start |
'[retriever name]' |
{"query": "hello"} |
||
on_retriever_end |
'[retriever name]' |
{"query": "hello"} |
[Document(...), ..] |
|
on_prompt_start |
'[template_name]' |
{"question": "hello"} |
||
on_prompt_end |
'[template_name]' |
{"question": "hello"} |
ChatPromptValue(messages: [SystemMessage, ...]) |
In addition to the standard events, users can also dispatch custom events (see example below).
Custom events will be only be surfaced with in the v2 version of the API!
A custom event has following format:
| Attribute | Type | Description |
|---|---|---|
name |
str |
A user defined name for the event. |
data |
Any |
The data associated with the event. This can be anything, though we suggest making it JSON serializable. |
Here are declarations associated with the standard events shown above:
format_docs:
def format_docs(docs: list[Document]) -> str:
'''Format the docs.'''
return ", ".join([doc.page_content for doc in docs])
format_docs = RunnableLambda(format_docs)
some_tool:
prompt:
template = ChatPromptTemplate.from_messages(
[
("system", "You are Cat Agent 007"),
("human", "{question}"),
]
).with_config({"run_name": "my_template", "tags": ["my_template"]})
For instance:
from langchain_core.runnables import RunnableLambda
async def reverse(s: str) -> str:
return s[::-1]
chain = RunnableLambda(func=reverse)
events = [event async for event in chain.astream_events("hello", version="v2")]
# Will produce the following events
# (run_id, and parent_ids has been omitted for brevity):
[
{
"data": {"input": "hello"},
"event": "on_chain_start",
"metadata": {},
"name": "reverse",
"tags": [],
},
{
"data": {"chunk": "olleh"},
"event": "on_chain_stream",
"metadata": {},
"name": "reverse",
"tags": [],
},
{
"data": {"output": "olleh"},
"event": "on_chain_end",
"metadata": {},
"name": "reverse",
"tags": [],
},
]
from langchain_core.callbacks.manager import (
adispatch_custom_event,
)
from langchain_core.runnables import RunnableLambda, RunnableConfig
import asyncio
async def slow_thing(some_input: str, config: RunnableConfig) -> str:
"""Do something that takes a long time."""
await asyncio.sleep(1) # Placeholder for some slow operation
await adispatch_custom_event(
"progress_event",
{"message": "Finished step 1 of 3"},
config=config # Must be included for python < 3.10
)
await asyncio.sleep(1) # Placeholder for some slow operation
await adispatch_custom_event(
"progress_event",
{"message": "Finished step 2 of 3"},
config=config # Must be included for python < 3.10
)
await asyncio.sleep(1) # Placeholder for some slow operation
return "Done"
slow_thing = RunnableLambda(slow_thing)
async for event in slow_thing.astream_events("some_input", version="v2"):
print(event)
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
version
|
The version of the schema to use either
TYPE:
|
include_names
|
Only include events from |
include_types
|
Only include events from |
include_tags
|
Only include events from |
exclude_names
|
Exclude events from |
exclude_types
|
Exclude events from |
exclude_tags
|
Exclude events from |
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[StreamEvent]
|
An async stream of |
| RAISES | DESCRIPTION |
|---|---|
NotImplementedError
|
If the version is not |
transform
¶
transform(
input: Iterator[Input], config: RunnableConfig | None = None, **kwargs: Any | None
) -> Iterator[Output]
Transform inputs to outputs.
Default implementation of transform, which buffers input and calls astream.
Subclasses must override this method if they can start producing output while input is still being generated.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
An iterator of inputs to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
Output
|
The output of the |
atransform
async
¶
atransform(
input: AsyncIterator[Input],
config: RunnableConfig | None = None,
**kwargs: Any | None,
) -> AsyncIterator[Output]
Transform inputs to outputs.
Default implementation of atransform, which buffers input and calls astream.
Subclasses must override this method if they can start producing output while input is still being generated.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
An async iterator of inputs to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[Output]
|
The output of the |
bind
¶
Bind arguments to a Runnable, returning a new Runnable.
Useful when a Runnable in a chain requires an argument that is not
in the output of the previous Runnable or included in the user input.
| PARAMETER | DESCRIPTION |
|---|---|
**kwargs
|
The arguments to bind to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser
model = ChatOllama(model="llama3.1")
# Without bind
chain = model | StrOutputParser()
chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two three four five.'
# With bind
chain = model.bind(stop=["three"]) | StrOutputParser()
chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two'
with_config
¶
with_config(
config: RunnableConfig | None = None, **kwargs: Any
) -> Runnable[Input, Output]
Bind config to a Runnable, returning a new Runnable.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
The config to bind to the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
with_listeners
¶
with_listeners(
*,
on_start: Callable[[Run], None]
| Callable[[Run, RunnableConfig], None]
| None = None,
on_end: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None = None,
on_error: Callable[[Run], None]
| Callable[[Run, RunnableConfig], None]
| None = None,
) -> Runnable[Input, Output]
Bind lifecycle listeners to a Runnable, returning a new Runnable.
The Run object contains information about the run, including its id,
type, input, output, error, start_time, end_time, and
any tags or metadata added to the run.
| PARAMETER | DESCRIPTION |
|---|---|
on_start
|
Called before the
TYPE:
|
on_end
|
Called after the
TYPE:
|
on_error
|
Called if the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_core.runnables import RunnableLambda
from langchain_core.tracers.schemas import Run
import time
def test_runnable(time_to_sleep: int):
time.sleep(time_to_sleep)
def fn_start(run_obj: Run):
print("start_time:", run_obj.start_time)
def fn_end(run_obj: Run):
print("end_time:", run_obj.end_time)
chain = RunnableLambda(test_runnable).with_listeners(
on_start=fn_start, on_end=fn_end
)
chain.invoke(2)
with_alisteners
¶
with_alisteners(
*,
on_start: AsyncListener | None = None,
on_end: AsyncListener | None = None,
on_error: AsyncListener | None = None,
) -> Runnable[Input, Output]
Bind async lifecycle listeners to a Runnable.
Returns a new Runnable.
The Run object contains information about the run, including its id,
type, input, output, error, start_time, end_time, and
any tags or metadata added to the run.
| PARAMETER | DESCRIPTION |
|---|---|
on_start
|
Called asynchronously before the
TYPE:
|
on_end
|
Called asynchronously after the
TYPE:
|
on_error
|
Called asynchronously if the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_core.runnables import RunnableLambda, Runnable
from datetime import datetime, timezone
import time
import asyncio
def format_t(timestamp: float) -> str:
return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat()
async def test_runnable(time_to_sleep: int):
print(f"Runnable[{time_to_sleep}s]: starts at {format_t(time.time())}")
await asyncio.sleep(time_to_sleep)
print(f"Runnable[{time_to_sleep}s]: ends at {format_t(time.time())}")
async def fn_start(run_obj: Runnable):
print(f"on start callback starts at {format_t(time.time())}")
await asyncio.sleep(3)
print(f"on start callback ends at {format_t(time.time())}")
async def fn_end(run_obj: Runnable):
print(f"on end callback starts at {format_t(time.time())}")
await asyncio.sleep(2)
print(f"on end callback ends at {format_t(time.time())}")
runnable = RunnableLambda(test_runnable).with_alisteners(
on_start=fn_start, on_end=fn_end
)
async def concurrent_runs():
await asyncio.gather(runnable.ainvoke(2), runnable.ainvoke(3))
asyncio.run(concurrent_runs())
# Result:
# on start callback starts at 2025-03-01T07:05:22.875378+00:00
# on start callback starts at 2025-03-01T07:05:22.875495+00:00
# on start callback ends at 2025-03-01T07:05:25.878862+00:00
# on start callback ends at 2025-03-01T07:05:25.878947+00:00
# Runnable[2s]: starts at 2025-03-01T07:05:25.879392+00:00
# Runnable[3s]: starts at 2025-03-01T07:05:25.879804+00:00
# Runnable[2s]: ends at 2025-03-01T07:05:27.881998+00:00
# on end callback starts at 2025-03-01T07:05:27.882360+00:00
# Runnable[3s]: ends at 2025-03-01T07:05:28.881737+00:00
# on end callback starts at 2025-03-01T07:05:28.882428+00:00
# on end callback ends at 2025-03-01T07:05:29.883893+00:00
# on end callback ends at 2025-03-01T07:05:30.884831+00:00
with_types
¶
with_types(
*, input_type: type[Input] | None = None, output_type: type[Output] | None = None
) -> Runnable[Input, Output]
Bind input and output types to a Runnable, returning a new Runnable.
| PARAMETER | DESCRIPTION |
|---|---|
input_type
|
The input type to bind to the
TYPE:
|
output_type
|
The output type to bind to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
with_retry
¶
with_retry(
*,
retry_if_exception_type: tuple[type[BaseException], ...] = (Exception,),
wait_exponential_jitter: bool = True,
exponential_jitter_params: ExponentialJitterParams | None = None,
stop_after_attempt: int = 3,
) -> Runnable[Input, Output]
Create a new Runnable that retries the original Runnable on exceptions.
| PARAMETER | DESCRIPTION |
|---|---|
retry_if_exception_type
|
A tuple of exception types to retry on.
TYPE:
|
wait_exponential_jitter
|
Whether to add jitter to the wait time between retries.
TYPE:
|
stop_after_attempt
|
The maximum number of attempts to make before giving up.
TYPE:
|
exponential_jitter_params
|
Parameters for
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_core.runnables import RunnableLambda
count = 0
def _lambda(x: int) -> None:
global count
count = count + 1
if x == 1:
raise ValueError("x is 1")
else:
pass
runnable = RunnableLambda(_lambda)
try:
runnable.with_retry(
stop_after_attempt=2,
retry_if_exception_type=(ValueError,),
).invoke(1)
except ValueError:
pass
assert count == 2
map
¶
with_fallbacks
¶
with_fallbacks(
fallbacks: Sequence[Runnable[Input, Output]],
*,
exceptions_to_handle: tuple[type[BaseException], ...] = (Exception,),
exception_key: str | None = None,
) -> RunnableWithFallbacks[Input, Output]
Add fallbacks to a Runnable, returning a new Runnable.
The new Runnable will try the original Runnable, and then each fallback
in order, upon failures.
| PARAMETER | DESCRIPTION |
|---|---|
fallbacks
|
A sequence of runnables to try if the original |
exceptions_to_handle
|
A tuple of exception types to handle.
TYPE:
|
exception_key
|
If If If used, the base
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableWithFallbacks[Input, Output]
|
A new |
Example
from typing import Iterator
from langchain_core.runnables import RunnableGenerator
def _generate_immediate_error(input: Iterator) -> Iterator[str]:
raise ValueError()
yield ""
def _generate(input: Iterator) -> Iterator[str]:
yield from "foo bar"
runnable = RunnableGenerator(_generate_immediate_error).with_fallbacks(
[RunnableGenerator(_generate)]
)
print("".join(runnable.stream({}))) # foo bar
| PARAMETER | DESCRIPTION |
|---|---|
fallbacks
|
A sequence of runnables to try if the original |
exceptions_to_handle
|
A tuple of exception types to handle.
TYPE:
|
exception_key
|
If If If used, the base
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableWithFallbacks[Input, Output]
|
A new |
as_tool
¶
as_tool(
args_schema: type[BaseModel] | None = None,
*,
name: str | None = None,
description: str | None = None,
arg_types: dict[str, type] | None = None,
) -> BaseTool
Create a BaseTool from a Runnable.
as_tool will instantiate a BaseTool with a name, description, and
args_schema from a Runnable. Where possible, schemas are inferred
from runnable.get_input_schema.
Alternatively (e.g., if the Runnable takes a dict as input and the specific
dict keys are not typed), the schema can be specified directly with
args_schema.
You can also pass arg_types to just specify the required arguments and their
types.
| PARAMETER | DESCRIPTION |
|---|---|
args_schema
|
The schema for the tool. |
name
|
The name of the tool.
TYPE:
|
description
|
The description of the tool.
TYPE:
|
arg_types
|
A dictionary of argument names to types. |
| RETURNS | DESCRIPTION |
|---|---|
BaseTool
|
A |
Typed dict input:
from typing_extensions import TypedDict
from langchain_core.runnables import RunnableLambda
class Args(TypedDict):
a: int
b: list[int]
def f(x: Args) -> str:
return str(x["a"] * max(x["b"]))
runnable = RunnableLambda(f)
as_tool = runnable.as_tool()
as_tool.invoke({"a": 3, "b": [1, 2]})
dict input, specifying schema via args_schema:
from typing import Any
from pydantic import BaseModel, Field
from langchain_core.runnables import RunnableLambda
def f(x: dict[str, Any]) -> str:
return str(x["a"] * max(x["b"]))
class FSchema(BaseModel):
"""Apply a function to an integer and list of integers."""
a: int = Field(..., description="Integer")
b: list[int] = Field(..., description="List of ints")
runnable = RunnableLambda(f)
as_tool = runnable.as_tool(FSchema)
as_tool.invoke({"a": 3, "b": [1, 2]})
dict input, specifying schema via arg_types:
from typing import Any
from langchain_core.runnables import RunnableLambda
def f(x: dict[str, Any]) -> str:
return str(x["a"] * max(x["b"]))
runnable = RunnableLambda(f)
as_tool = runnable.as_tool(arg_types={"a": int, "b": list[int]})
as_tool.invoke({"a": 3, "b": [1, 2]})
str input:
is_lc_serializable
classmethod
¶
is_lc_serializable() -> bool
Is this class serializable?
By design, even if a class inherits from Serializable, it is not serializable
by default. This is to prevent accidental serialization of objects that should
not be serialized.
| RETURNS | DESCRIPTION |
|---|---|
bool
|
Whether the class is serializable. Default is |
get_lc_namespace
classmethod
¶
lc_id
classmethod
¶
Return a unique identifier for this class for serialization purposes.
The unique identifier is a list of strings that describes the path to the object.
For example, for the class langchain.llms.openai.OpenAI, the id is
["langchain", "llms", "openai", "OpenAI"].
to_json
¶
Serialize the Runnable to JSON.
| RETURNS | DESCRIPTION |
|---|---|
SerializedConstructor | SerializedNotImplemented
|
A JSON-serializable representation of the |
to_json_not_implemented
¶
Serialize a "not implemented" object.
| RETURNS | DESCRIPTION |
|---|---|
SerializedNotImplemented
|
|
configurable_fields
¶
configurable_fields(
**kwargs: AnyConfigurableField,
) -> RunnableSerializable[Input, Output]
Configure particular Runnable fields at runtime.
| PARAMETER | DESCRIPTION |
|---|---|
**kwargs
|
A dictionary of
TYPE:
|
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If a configuration key is not found in the |
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Input, Output]
|
A new |
from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI
model = ChatOpenAI(max_tokens=20).configurable_fields(
max_tokens=ConfigurableField(
id="output_token_number",
name="Max tokens in the output",
description="The maximum number of tokens in the output",
)
)
# max_tokens = 20
print("max_tokens_20: ", model.invoke("tell me something about chess").content)
# max_tokens = 200
print(
"max_tokens_200: ",
model.with_config(configurable={"output_token_number": 200})
.invoke("tell me something about chess")
.content,
)
configurable_alternatives
¶
configurable_alternatives(
which: ConfigurableField,
*,
default_key: str = "default",
prefix_keys: bool = False,
**kwargs: Runnable[Input, Output] | Callable[[], Runnable[Input, Output]],
) -> RunnableSerializable[Input, Output]
Configure alternatives for Runnable objects that can be set at runtime.
| PARAMETER | DESCRIPTION |
|---|---|
which
|
The
TYPE:
|
default_key
|
The default key to use if no alternative is selected.
TYPE:
|
prefix_keys
|
Whether to prefix the keys with the
TYPE:
|
**kwargs
|
A dictionary of keys to
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Input, Output]
|
A new |
from langchain_anthropic import ChatAnthropic
from langchain_core.runnables.utils import ConfigurableField
from langchain_openai import ChatOpenAI
model = ChatAnthropic(
model_name="claude-sonnet-4-5-20250929"
).configurable_alternatives(
ConfigurableField(id="llm"),
default_key="anthropic",
openai=ChatOpenAI(),
)
# uses the default model ChatAnthropic
print(model.invoke("which organization created you?").content)
# uses ChatOpenAI
print(
model.with_config(configurable={"llm": "openai"})
.invoke("which organization created you?")
.content
)
langchain_azure_ai.tools
¶
Tools provided by Azure AI Foundry.
AzureLogicAppTool
¶
Bases: BaseTool
A tool that interacts with Azure Logic Apps.
| METHOD | DESCRIPTION |
|---|---|
get_name |
Get the name of the |
get_input_schema |
The tool's input schema. |
get_input_jsonschema |
Get a JSON schema that represents the input to the |
get_output_schema |
Get a Pydantic model that can be used to validate output to the |
get_output_jsonschema |
Get a JSON schema that represents the output of the |
config_schema |
The type of config this |
get_config_jsonschema |
Get a JSON schema that represents the config of the |
get_graph |
Return a graph representation of this |
get_prompts |
Return a list of prompts used by this |
__or__ |
Runnable "or" operator. |
__ror__ |
Runnable "reverse-or" operator. |
pipe |
Pipe |
pick |
Pick keys from the output |
assign |
Assigns new fields to the |
invoke |
Transform a single input into an output. |
ainvoke |
Transform a single input into an output. |
batch |
Default implementation runs invoke in parallel using a thread pool executor. |
batch_as_completed |
Run |
abatch |
Default implementation runs |
abatch_as_completed |
Run |
stream |
Default implementation of |
astream |
Default implementation of |
astream_log |
Stream all output from a |
astream_events |
Generate a stream of events. |
transform |
Transform inputs to outputs. |
atransform |
Transform inputs to outputs. |
bind |
Bind arguments to a |
with_config |
Bind config to a |
with_listeners |
Bind lifecycle listeners to a |
with_alisteners |
Bind async lifecycle listeners to a |
with_types |
Bind input and output types to a |
with_retry |
Create a new |
map |
Return a new |
with_fallbacks |
Add fallbacks to a |
as_tool |
Create a |
__init__ |
Initialize the tool. |
is_lc_serializable |
Is this class serializable? |
get_lc_namespace |
Get the namespace of the LangChain object. |
lc_id |
Return a unique identifier for this class for serialization purposes. |
to_json |
Serialize the |
to_json_not_implemented |
Serialize a "not implemented" object. |
configurable_fields |
Configure particular |
configurable_alternatives |
Configure alternatives for |
__init_subclass__ |
Validate the tool class definition during subclass creation. |
run |
Run the tool. |
arun |
Run the tool asynchronously. |
initialize_client |
Initialize the Azure Logic Apps client. |
register_logic_app |
Retrieves and stores a callback URL for a specific Logic App + trigger. |
invoke_logic_app |
Invokes the registered Logic App (by name) with the given JSON payload. |
InputType
property
¶
InputType: type[Input]
Input type.
The type of input this Runnable accepts specified as a type annotation.
| RAISES | DESCRIPTION |
|---|---|
TypeError
|
If the input type cannot be inferred. |
OutputType
property
¶
OutputType: type[Output]
Output Type.
The type of output this Runnable produces specified as a type annotation.
| RAISES | DESCRIPTION |
|---|---|
TypeError
|
If the output type cannot be inferred. |
input_schema
property
¶
The type of input this Runnable accepts specified as a Pydantic model.
output_schema
property
¶
Output schema.
The type of output this Runnable produces specified as a Pydantic model.
config_specs
property
¶
config_specs: list[ConfigurableFieldSpec]
List configurable fields for this Runnable.
lc_secrets
property
¶
A map of constructor argument names to secret ids.
For example, {"openai_api_key": "OPENAI_API_KEY"}
lc_attributes
property
¶
lc_attributes: dict
List of attribute names that should be included in the serialized kwargs.
These attributes must be accepted by the constructor.
Default is an empty dictionary.
args_schema
class-attribute
instance-attribute
¶
args_schema: Annotated[ArgsSchema | None, SkipValidation()] = Field(
default=None, description="The tool schema."
)
Pydantic model class to validate and parse the tool's input arguments.
Args schema should be either:
- A subclass of
pydantic.BaseModel. - A subclass of
pydantic.v1.BaseModelif accessing v1 namespace in pydantic 2 - A JSON schema dict
return_direct
class-attribute
instance-attribute
¶
return_direct: bool = False
Whether to return the tool's output directly.
Setting this to True means that after the tool is called, the AgentExecutor will
stop looping.
verbose
class-attribute
instance-attribute
¶
verbose: bool = False
Whether to log the tool's progress.
callbacks
class-attribute
instance-attribute
¶
callbacks: Callbacks = Field(default=None, exclude=True)
Callbacks to be called during tool execution.
tags
class-attribute
instance-attribute
¶
Optional list of tags associated with the tool.
These tags will be associated with each call to this tool,
and passed as arguments to the handlers defined in callbacks.
You can use these to, e.g., identify a specific instance of a tool with its use case.
metadata
class-attribute
instance-attribute
¶
Optional metadata associated with the tool.
This metadata will be associated with each call to this tool,
and passed as arguments to the handlers defined in callbacks.
You can use these to, e.g., identify a specific instance of a tool with its use case.
handle_tool_error
class-attribute
instance-attribute
¶
Handle the content of the ToolException thrown.
handle_validation_error
class-attribute
instance-attribute
¶
handle_validation_error: (
bool | str | Callable[[ValidationError | ValidationError], str] | None
) = False
Handle the content of the ValidationError thrown.
response_format
class-attribute
instance-attribute
¶
response_format: Literal['content', 'content_and_artifact'] = 'content'
The tool response format.
If 'content' then the output of the tool is interpreted as the contents of a
ToolMessage. If 'content_and_artifact' then the output is expected to be a
two-tuple corresponding to the (content, artifact) of a ToolMessage.
is_single_input
property
¶
is_single_input: bool
Check if the tool accepts only a single input argument.
| RETURNS | DESCRIPTION |
|---|---|
bool
|
|
args
property
¶
args: dict
Get the tool's input arguments schema.
| RETURNS | DESCRIPTION |
|---|---|
dict
|
|
tool_call_schema
property
¶
Get the schema for tool calls, excluding injected arguments.
| RETURNS | DESCRIPTION |
|---|---|
ArgsSchema
|
The schema that should be used for tool calls from language models. |
name
class-attribute
instance-attribute
¶
name: str = 'azure_logic_app_tool'
The name of the tool. Use a descriptive name that indicates its purpose.
description
class-attribute
instance-attribute
¶
description: str = "Invokes Azure Logic Apps workflows to trigger automated business processes and integrations. Use this to execute pre-configured workflows such as sending emails, processing data, calling APIs, or integrating with other Azure and third-party services. Input is JSON payload for the workflow trigger. Ideal for automation tasks, notifications, data synchronization, and orchestrating multi-step processes."
A description of the tool that explains its functionality and usage. Use this description to help users understand when to use this tool.
subscription_id
instance-attribute
¶
subscription_id: str
Azure Subscription ID where the Logic Apps are hosted.
resource_group
instance-attribute
¶
resource_group: str
Azure Resource Group where the Logic Apps are hosted.
credential
class-attribute
instance-attribute
¶
The API key or credential to use to connect to the service. I f None, DefaultAzureCredential is used.
trigger_name
instance-attribute
¶
trigger_name: str
The name of the trigger in the Logic App to invoke.
get_name
¶
get_input_schema
¶
get_input_schema(config: RunnableConfig | None = None) -> type[BaseModel]
The tool's input schema.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
The configuration for the tool.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
type[BaseModel]
|
The input schema for the tool. |
get_input_jsonschema
¶
get_input_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]
Get a JSON schema that represents the input to the Runnable.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
A config to use when generating the schema.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
dict[str, Any]
|
A JSON schema that represents the input to the |
Example
Added in langchain-core 0.3.0
get_output_schema
¶
get_output_schema(config: RunnableConfig | None = None) -> type[BaseModel]
Get a Pydantic model that can be used to validate output to the Runnable.
Runnable objects that leverage the configurable_fields and
configurable_alternatives methods will have a dynamic output schema that
depends on which configuration the Runnable is invoked with.
This method allows to get an output schema for a specific configuration.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
A config to use when generating the schema.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
type[BaseModel]
|
A Pydantic model that can be used to validate output. |
get_output_jsonschema
¶
get_output_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]
Get a JSON schema that represents the output of the Runnable.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
A config to use when generating the schema.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
dict[str, Any]
|
A JSON schema that represents the output of the |
Example
Added in langchain-core 0.3.0
config_schema
¶
The type of config this Runnable accepts specified as a Pydantic model.
To mark a field as configurable, see the configurable_fields
and configurable_alternatives methods.
| PARAMETER | DESCRIPTION |
|---|---|
include
|
A list of fields to include in the config schema. |
| RETURNS | DESCRIPTION |
|---|---|
type[BaseModel]
|
A Pydantic model that can be used to validate config. |
get_config_jsonschema
¶
get_graph
¶
get_graph(config: RunnableConfig | None = None) -> Graph
Return a graph representation of this Runnable.
get_prompts
¶
get_prompts(config: RunnableConfig | None = None) -> list[BasePromptTemplate]
Return a list of prompts used by this Runnable.
__or__
¶
__or__(
other: Runnable[Any, Other]
| Callable[[Iterator[Any]], Iterator[Other]]
| Callable[[AsyncIterator[Any]], AsyncIterator[Other]]
| Callable[[Any], Other]
| Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any],
) -> RunnableSerializable[Input, Other]
Runnable "or" operator.
Compose this Runnable with another object to create a
RunnableSequence.
| PARAMETER | DESCRIPTION |
|---|---|
other
|
Another
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Input, Other]
|
A new |
__ror__
¶
__ror__(
other: Runnable[Other, Any]
| Callable[[Iterator[Other]], Iterator[Any]]
| Callable[[AsyncIterator[Other]], AsyncIterator[Any]]
| Callable[[Other], Any]
| Mapping[str, Runnable[Other, Any] | Callable[[Other], Any] | Any],
) -> RunnableSerializable[Other, Output]
Runnable "reverse-or" operator.
Compose this Runnable with another object to create a
RunnableSequence.
| PARAMETER | DESCRIPTION |
|---|---|
other
|
Another
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Other, Output]
|
A new |
pipe
¶
pipe(
*others: Runnable[Any, Other] | Callable[[Any], Other], name: str | None = None
) -> RunnableSerializable[Input, Other]
Pipe Runnable objects.
Compose this Runnable with Runnable-like objects to make a
RunnableSequence.
Equivalent to RunnableSequence(self, *others) or self | others[0] | ...
Example
from langchain_core.runnables import RunnableLambda
def add_one(x: int) -> int:
return x + 1
def mul_two(x: int) -> int:
return x * 2
runnable_1 = RunnableLambda(add_one)
runnable_2 = RunnableLambda(mul_two)
sequence = runnable_1.pipe(runnable_2)
# Or equivalently:
# sequence = runnable_1 | runnable_2
# sequence = RunnableSequence(first=runnable_1, last=runnable_2)
sequence.invoke(1)
await sequence.ainvoke(1)
# -> 4
sequence.batch([1, 2, 3])
await sequence.abatch([1, 2, 3])
# -> [4, 6, 8]
| PARAMETER | DESCRIPTION |
|---|---|
*others
|
Other
TYPE:
|
name
|
An optional name for the resulting
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Input, Other]
|
A new |
pick
¶
Pick keys from the output dict of this Runnable.
Pick a single key:
import json
from langchain_core.runnables import RunnableLambda, RunnableMap
as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)
chain = RunnableMap(str=as_str, json=as_json)
chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3]}
json_only_chain = chain.pick("json")
json_only_chain.invoke("[1, 2, 3]")
# -> [1, 2, 3]
Pick a list of keys:
from typing import Any
import json
from langchain_core.runnables import RunnableLambda, RunnableMap
as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)
def as_bytes(x: Any) -> bytes:
return bytes(x, "utf-8")
chain = RunnableMap(str=as_str, json=as_json, bytes=RunnableLambda(as_bytes))
chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3], "bytes": b"[1, 2, 3]"}
json_and_bytes_chain = chain.pick(["json", "bytes"])
json_and_bytes_chain.invoke("[1, 2, 3]")
# -> {"json": [1, 2, 3], "bytes": b"[1, 2, 3]"}
| PARAMETER | DESCRIPTION |
|---|---|
keys
|
A key or list of keys to pick from the output dict. |
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Any, Any]
|
a new |
assign
¶
assign(
**kwargs: Runnable[dict[str, Any], Any]
| Callable[[dict[str, Any]], Any]
| Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]],
) -> RunnableSerializable[Any, Any]
Assigns new fields to the dict output of this Runnable.
from langchain_core.language_models.fake import FakeStreamingListLLM
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import SystemMessagePromptTemplate
from langchain_core.runnables import Runnable
from operator import itemgetter
prompt = (
SystemMessagePromptTemplate.from_template("You are a nice assistant.")
+ "{question}"
)
model = FakeStreamingListLLM(responses=["foo-lish"])
chain: Runnable = prompt | model | {"str": StrOutputParser()}
chain_with_assign = chain.assign(hello=itemgetter("str") | model)
print(chain_with_assign.input_schema.model_json_schema())
# {'title': 'PromptInput', 'type': 'object', 'properties':
{'question': {'title': 'Question', 'type': 'string'}}}
print(chain_with_assign.output_schema.model_json_schema())
# {'title': 'RunnableSequenceOutput', 'type': 'object', 'properties':
{'str': {'title': 'Str',
'type': 'string'}, 'hello': {'title': 'Hello', 'type': 'string'}}}
| PARAMETER | DESCRIPTION |
|---|---|
**kwargs
|
A mapping of keys to
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Any, Any]
|
A new |
invoke
¶
Transform a single input into an output.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
A config to use when invoking the The config supports standard keys like Please refer to
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Output
|
The output of the |
ainvoke
async
¶
Transform a single input into an output.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
A config to use when invoking the The config supports standard keys like Please refer to
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Output
|
The output of the |
batch
¶
batch(
inputs: list[Input],
config: RunnableConfig | list[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any | None,
) -> list[Output]
Default implementation runs invoke in parallel using a thread pool executor.
The default implementation of batch works well for IO bound runnables.
Subclasses must override this method if they can batch more efficiently;
e.g., if the underlying Runnable uses an API which supports a batch mode.
| PARAMETER | DESCRIPTION |
|---|---|
inputs
|
A list of inputs to the
TYPE:
|
config
|
A config to use when invoking the Please refer to
TYPE:
|
return_exceptions
|
Whether to return exceptions instead of raising them.
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Output]
|
A list of outputs from the |
batch_as_completed
¶
batch_as_completed(
inputs: Sequence[Input],
config: RunnableConfig | Sequence[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any | None,
) -> Iterator[tuple[int, Output | Exception]]
Run invoke in parallel on a list of inputs.
Yields results as they complete.
| PARAMETER | DESCRIPTION |
|---|---|
inputs
|
A list of inputs to the
TYPE:
|
config
|
A config to use when invoking the The config supports standard keys like Please refer to
TYPE:
|
return_exceptions
|
Whether to return exceptions instead of raising them.
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
tuple[int, Output | Exception]
|
Tuples of the index of the input and the output from the |
abatch
async
¶
abatch(
inputs: list[Input],
config: RunnableConfig | list[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any | None,
) -> list[Output]
Default implementation runs ainvoke in parallel using asyncio.gather.
The default implementation of batch works well for IO bound runnables.
Subclasses must override this method if they can batch more efficiently;
e.g., if the underlying Runnable uses an API which supports a batch mode.
| PARAMETER | DESCRIPTION |
|---|---|
inputs
|
A list of inputs to the
TYPE:
|
config
|
A config to use when invoking the The config supports standard keys like Please refer to
TYPE:
|
return_exceptions
|
Whether to return exceptions instead of raising them.
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Output]
|
A list of outputs from the |
abatch_as_completed
async
¶
abatch_as_completed(
inputs: Sequence[Input],
config: RunnableConfig | Sequence[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any | None,
) -> AsyncIterator[tuple[int, Output | Exception]]
Run ainvoke in parallel on a list of inputs.
Yields results as they complete.
| PARAMETER | DESCRIPTION |
|---|---|
inputs
|
A list of inputs to the
TYPE:
|
config
|
A config to use when invoking the The config supports standard keys like Please refer to
TYPE:
|
return_exceptions
|
Whether to return exceptions instead of raising them.
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[tuple[int, Output | Exception]]
|
A tuple of the index of the input and the output from the |
stream
¶
stream(
input: Input, config: RunnableConfig | None = None, **kwargs: Any | None
) -> Iterator[Output]
Default implementation of stream, which calls invoke.
Subclasses must override this method if they support streaming output.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
Output
|
The output of the |
astream
async
¶
astream(
input: Input, config: RunnableConfig | None = None, **kwargs: Any | None
) -> AsyncIterator[Output]
Default implementation of astream, which calls ainvoke.
Subclasses must override this method if they support streaming output.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[Output]
|
The output of the |
astream_log
async
¶
astream_log(
input: Any,
config: RunnableConfig | None = None,
*,
diff: bool = True,
with_streamed_output_list: bool = True,
include_names: Sequence[str] | None = None,
include_types: Sequence[str] | None = None,
include_tags: Sequence[str] | None = None,
exclude_names: Sequence[str] | None = None,
exclude_types: Sequence[str] | None = None,
exclude_tags: Sequence[str] | None = None,
**kwargs: Any,
) -> AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]
Stream all output from a Runnable, as reported to the callback system.
This includes all inner runs of LLMs, Retrievers, Tools, etc.
Output is streamed as Log objects, which include a list of Jsonpatch ops that describe how the state of the run has changed in each step, and the final state of the run.
The Jsonpatch ops can be applied in order to construct state.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
diff
|
Whether to yield diffs between each step or the current state.
TYPE:
|
with_streamed_output_list
|
Whether to yield the
TYPE:
|
include_names
|
Only include logs with these names. |
include_types
|
Only include logs with these types. |
include_tags
|
Only include logs with these tags. |
exclude_names
|
Exclude logs with these names. |
exclude_types
|
Exclude logs with these types. |
exclude_tags
|
Exclude logs with these tags. |
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]
|
A |
astream_events
async
¶
astream_events(
input: Any,
config: RunnableConfig | None = None,
*,
version: Literal["v1", "v2"] = "v2",
include_names: Sequence[str] | None = None,
include_types: Sequence[str] | None = None,
include_tags: Sequence[str] | None = None,
exclude_names: Sequence[str] | None = None,
exclude_types: Sequence[str] | None = None,
exclude_tags: Sequence[str] | None = None,
**kwargs: Any,
) -> AsyncIterator[StreamEvent]
Generate a stream of events.
Use to create an iterator over StreamEvent that provide real-time information
about the progress of the Runnable, including StreamEvent from intermediate
results.
A StreamEvent is a dictionary with the following schema:
event: Event names are of the format:on_[runnable_type]_(start|stream|end).name: The name of theRunnablethat generated the event.run_id: Randomly generated ID associated with the given execution of theRunnablethat emitted the event. A childRunnablethat gets invoked as part of the execution of a parentRunnableis assigned its own unique ID.parent_ids: The IDs of the parent runnables that generated the event. The rootRunnablewill have an empty list. The order of the parent IDs is from the root to the immediate parent. Only available for v2 version of the API. The v1 version of the API will return an empty list.tags: The tags of theRunnablethat generated the event.metadata: The metadata of theRunnablethat generated the event.data: The data associated with the event. The contents of this field depend on the type of event. See the table below for more details.
Below is a table that illustrates some events that might be emitted by various chains. Metadata fields have been omitted from the table for brevity. Chain definitions have been included after the table.
Note
This reference table is for the v2 version of the schema.
| event | name | chunk | input | output |
|---|---|---|---|---|
on_chat_model_start |
'[model name]' |
{"messages": [[SystemMessage, HumanMessage]]} |
||
on_chat_model_stream |
'[model name]' |
AIMessageChunk(content="hello") |
||
on_chat_model_end |
'[model name]' |
{"messages": [[SystemMessage, HumanMessage]]} |
AIMessageChunk(content="hello world") |
|
on_llm_start |
'[model name]' |
{'input': 'hello'} |
||
on_llm_stream |
'[model name]' |
'Hello' |
||
on_llm_end |
'[model name]' |
'Hello human!' |
||
on_chain_start |
'format_docs' |
|||
on_chain_stream |
'format_docs' |
'hello world!, goodbye world!' |
||
on_chain_end |
'format_docs' |
[Document(...)] |
'hello world!, goodbye world!' |
|
on_tool_start |
'some_tool' |
{"x": 1, "y": "2"} |
||
on_tool_end |
'some_tool' |
{"x": 1, "y": "2"} |
||
on_retriever_start |
'[retriever name]' |
{"query": "hello"} |
||
on_retriever_end |
'[retriever name]' |
{"query": "hello"} |
[Document(...), ..] |
|
on_prompt_start |
'[template_name]' |
{"question": "hello"} |
||
on_prompt_end |
'[template_name]' |
{"question": "hello"} |
ChatPromptValue(messages: [SystemMessage, ...]) |
In addition to the standard events, users can also dispatch custom events (see example below).
Custom events will be only be surfaced with in the v2 version of the API!
A custom event has following format:
| Attribute | Type | Description |
|---|---|---|
name |
str |
A user defined name for the event. |
data |
Any |
The data associated with the event. This can be anything, though we suggest making it JSON serializable. |
Here are declarations associated with the standard events shown above:
format_docs:
def format_docs(docs: list[Document]) -> str:
'''Format the docs.'''
return ", ".join([doc.page_content for doc in docs])
format_docs = RunnableLambda(format_docs)
some_tool:
prompt:
template = ChatPromptTemplate.from_messages(
[
("system", "You are Cat Agent 007"),
("human", "{question}"),
]
).with_config({"run_name": "my_template", "tags": ["my_template"]})
For instance:
from langchain_core.runnables import RunnableLambda
async def reverse(s: str) -> str:
return s[::-1]
chain = RunnableLambda(func=reverse)
events = [event async for event in chain.astream_events("hello", version="v2")]
# Will produce the following events
# (run_id, and parent_ids has been omitted for brevity):
[
{
"data": {"input": "hello"},
"event": "on_chain_start",
"metadata": {},
"name": "reverse",
"tags": [],
},
{
"data": {"chunk": "olleh"},
"event": "on_chain_stream",
"metadata": {},
"name": "reverse",
"tags": [],
},
{
"data": {"output": "olleh"},
"event": "on_chain_end",
"metadata": {},
"name": "reverse",
"tags": [],
},
]
from langchain_core.callbacks.manager import (
adispatch_custom_event,
)
from langchain_core.runnables import RunnableLambda, RunnableConfig
import asyncio
async def slow_thing(some_input: str, config: RunnableConfig) -> str:
"""Do something that takes a long time."""
await asyncio.sleep(1) # Placeholder for some slow operation
await adispatch_custom_event(
"progress_event",
{"message": "Finished step 1 of 3"},
config=config # Must be included for python < 3.10
)
await asyncio.sleep(1) # Placeholder for some slow operation
await adispatch_custom_event(
"progress_event",
{"message": "Finished step 2 of 3"},
config=config # Must be included for python < 3.10
)
await asyncio.sleep(1) # Placeholder for some slow operation
return "Done"
slow_thing = RunnableLambda(slow_thing)
async for event in slow_thing.astream_events("some_input", version="v2"):
print(event)
| PARAMETER | DESCRIPTION |
|---|---|
input
|
The input to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
version
|
The version of the schema to use either
TYPE:
|
include_names
|
Only include events from |
include_types
|
Only include events from |
include_tags
|
Only include events from |
exclude_names
|
Exclude events from |
exclude_types
|
Exclude events from |
exclude_tags
|
Exclude events from |
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[StreamEvent]
|
An async stream of |
| RAISES | DESCRIPTION |
|---|---|
NotImplementedError
|
If the version is not |
transform
¶
transform(
input: Iterator[Input], config: RunnableConfig | None = None, **kwargs: Any | None
) -> Iterator[Output]
Transform inputs to outputs.
Default implementation of transform, which buffers input and calls astream.
Subclasses must override this method if they can start producing output while input is still being generated.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
An iterator of inputs to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
Output
|
The output of the |
atransform
async
¶
atransform(
input: AsyncIterator[Input],
config: RunnableConfig | None = None,
**kwargs: Any | None,
) -> AsyncIterator[Output]
Transform inputs to outputs.
Default implementation of atransform, which buffers input and calls astream.
Subclasses must override this method if they can start producing output while input is still being generated.
| PARAMETER | DESCRIPTION |
|---|---|
input
|
An async iterator of inputs to the
TYPE:
|
config
|
The config to use for the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| YIELDS | DESCRIPTION |
|---|---|
AsyncIterator[Output]
|
The output of the |
bind
¶
Bind arguments to a Runnable, returning a new Runnable.
Useful when a Runnable in a chain requires an argument that is not
in the output of the previous Runnable or included in the user input.
| PARAMETER | DESCRIPTION |
|---|---|
**kwargs
|
The arguments to bind to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser
model = ChatOllama(model="llama3.1")
# Without bind
chain = model | StrOutputParser()
chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two three four five.'
# With bind
chain = model.bind(stop=["three"]) | StrOutputParser()
chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two'
with_config
¶
with_config(
config: RunnableConfig | None = None, **kwargs: Any
) -> Runnable[Input, Output]
Bind config to a Runnable, returning a new Runnable.
| PARAMETER | DESCRIPTION |
|---|---|
config
|
The config to bind to the
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
with_listeners
¶
with_listeners(
*,
on_start: Callable[[Run], None]
| Callable[[Run, RunnableConfig], None]
| None = None,
on_end: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None = None,
on_error: Callable[[Run], None]
| Callable[[Run, RunnableConfig], None]
| None = None,
) -> Runnable[Input, Output]
Bind lifecycle listeners to a Runnable, returning a new Runnable.
The Run object contains information about the run, including its id,
type, input, output, error, start_time, end_time, and
any tags or metadata added to the run.
| PARAMETER | DESCRIPTION |
|---|---|
on_start
|
Called before the
TYPE:
|
on_end
|
Called after the
TYPE:
|
on_error
|
Called if the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_core.runnables import RunnableLambda
from langchain_core.tracers.schemas import Run
import time
def test_runnable(time_to_sleep: int):
time.sleep(time_to_sleep)
def fn_start(run_obj: Run):
print("start_time:", run_obj.start_time)
def fn_end(run_obj: Run):
print("end_time:", run_obj.end_time)
chain = RunnableLambda(test_runnable).with_listeners(
on_start=fn_start, on_end=fn_end
)
chain.invoke(2)
with_alisteners
¶
with_alisteners(
*,
on_start: AsyncListener | None = None,
on_end: AsyncListener | None = None,
on_error: AsyncListener | None = None,
) -> Runnable[Input, Output]
Bind async lifecycle listeners to a Runnable.
Returns a new Runnable.
The Run object contains information about the run, including its id,
type, input, output, error, start_time, end_time, and
any tags or metadata added to the run.
| PARAMETER | DESCRIPTION |
|---|---|
on_start
|
Called asynchronously before the
TYPE:
|
on_end
|
Called asynchronously after the
TYPE:
|
on_error
|
Called asynchronously if the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_core.runnables import RunnableLambda, Runnable
from datetime import datetime, timezone
import time
import asyncio
def format_t(timestamp: float) -> str:
return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat()
async def test_runnable(time_to_sleep: int):
print(f"Runnable[{time_to_sleep}s]: starts at {format_t(time.time())}")
await asyncio.sleep(time_to_sleep)
print(f"Runnable[{time_to_sleep}s]: ends at {format_t(time.time())}")
async def fn_start(run_obj: Runnable):
print(f"on start callback starts at {format_t(time.time())}")
await asyncio.sleep(3)
print(f"on start callback ends at {format_t(time.time())}")
async def fn_end(run_obj: Runnable):
print(f"on end callback starts at {format_t(time.time())}")
await asyncio.sleep(2)
print(f"on end callback ends at {format_t(time.time())}")
runnable = RunnableLambda(test_runnable).with_alisteners(
on_start=fn_start, on_end=fn_end
)
async def concurrent_runs():
await asyncio.gather(runnable.ainvoke(2), runnable.ainvoke(3))
asyncio.run(concurrent_runs())
# Result:
# on start callback starts at 2025-03-01T07:05:22.875378+00:00
# on start callback starts at 2025-03-01T07:05:22.875495+00:00
# on start callback ends at 2025-03-01T07:05:25.878862+00:00
# on start callback ends at 2025-03-01T07:05:25.878947+00:00
# Runnable[2s]: starts at 2025-03-01T07:05:25.879392+00:00
# Runnable[3s]: starts at 2025-03-01T07:05:25.879804+00:00
# Runnable[2s]: ends at 2025-03-01T07:05:27.881998+00:00
# on end callback starts at 2025-03-01T07:05:27.882360+00:00
# Runnable[3s]: ends at 2025-03-01T07:05:28.881737+00:00
# on end callback starts at 2025-03-01T07:05:28.882428+00:00
# on end callback ends at 2025-03-01T07:05:29.883893+00:00
# on end callback ends at 2025-03-01T07:05:30.884831+00:00
with_types
¶
with_types(
*, input_type: type[Input] | None = None, output_type: type[Output] | None = None
) -> Runnable[Input, Output]
Bind input and output types to a Runnable, returning a new Runnable.
| PARAMETER | DESCRIPTION |
|---|---|
input_type
|
The input type to bind to the
TYPE:
|
output_type
|
The output type to bind to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
with_retry
¶
with_retry(
*,
retry_if_exception_type: tuple[type[BaseException], ...] = (Exception,),
wait_exponential_jitter: bool = True,
exponential_jitter_params: ExponentialJitterParams | None = None,
stop_after_attempt: int = 3,
) -> Runnable[Input, Output]
Create a new Runnable that retries the original Runnable on exceptions.
| PARAMETER | DESCRIPTION |
|---|---|
retry_if_exception_type
|
A tuple of exception types to retry on.
TYPE:
|
wait_exponential_jitter
|
Whether to add jitter to the wait time between retries.
TYPE:
|
stop_after_attempt
|
The maximum number of attempts to make before giving up.
TYPE:
|
exponential_jitter_params
|
Parameters for
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_core.runnables import RunnableLambda
count = 0
def _lambda(x: int) -> None:
global count
count = count + 1
if x == 1:
raise ValueError("x is 1")
else:
pass
runnable = RunnableLambda(_lambda)
try:
runnable.with_retry(
stop_after_attempt=2,
retry_if_exception_type=(ValueError,),
).invoke(1)
except ValueError:
pass
assert count == 2
map
¶
with_fallbacks
¶
with_fallbacks(
fallbacks: Sequence[Runnable[Input, Output]],
*,
exceptions_to_handle: tuple[type[BaseException], ...] = (Exception,),
exception_key: str | None = None,
) -> RunnableWithFallbacks[Input, Output]
Add fallbacks to a Runnable, returning a new Runnable.
The new Runnable will try the original Runnable, and then each fallback
in order, upon failures.
| PARAMETER | DESCRIPTION |
|---|---|
fallbacks
|
A sequence of runnables to try if the original |
exceptions_to_handle
|
A tuple of exception types to handle.
TYPE:
|
exception_key
|
If If If used, the base
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableWithFallbacks[Input, Output]
|
A new |
Example
from typing import Iterator
from langchain_core.runnables import RunnableGenerator
def _generate_immediate_error(input: Iterator) -> Iterator[str]:
raise ValueError()
yield ""
def _generate(input: Iterator) -> Iterator[str]:
yield from "foo bar"
runnable = RunnableGenerator(_generate_immediate_error).with_fallbacks(
[RunnableGenerator(_generate)]
)
print("".join(runnable.stream({}))) # foo bar
| PARAMETER | DESCRIPTION |
|---|---|
fallbacks
|
A sequence of runnables to try if the original |
exceptions_to_handle
|
A tuple of exception types to handle.
TYPE:
|
exception_key
|
If If If used, the base
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableWithFallbacks[Input, Output]
|
A new |
as_tool
¶
as_tool(
args_schema: type[BaseModel] | None = None,
*,
name: str | None = None,
description: str | None = None,
arg_types: dict[str, type] | None = None,
) -> BaseTool
Create a BaseTool from a Runnable.
as_tool will instantiate a BaseTool with a name, description, and
args_schema from a Runnable. Where possible, schemas are inferred
from runnable.get_input_schema.
Alternatively (e.g., if the Runnable takes a dict as input and the specific
dict keys are not typed), the schema can be specified directly with
args_schema.
You can also pass arg_types to just specify the required arguments and their
types.
| PARAMETER | DESCRIPTION |
|---|---|
args_schema
|
The schema for the tool. |
name
|
The name of the tool.
TYPE:
|
description
|
The description of the tool.
TYPE:
|
arg_types
|
A dictionary of argument names to types. |
| RETURNS | DESCRIPTION |
|---|---|
BaseTool
|
A |
Typed dict input:
from typing_extensions import TypedDict
from langchain_core.runnables import RunnableLambda
class Args(TypedDict):
a: int
b: list[int]
def f(x: Args) -> str:
return str(x["a"] * max(x["b"]))
runnable = RunnableLambda(f)
as_tool = runnable.as_tool()
as_tool.invoke({"a": 3, "b": [1, 2]})
dict input, specifying schema via args_schema:
from typing import Any
from pydantic import BaseModel, Field
from langchain_core.runnables import RunnableLambda
def f(x: dict[str, Any]) -> str:
return str(x["a"] * max(x["b"]))
class FSchema(BaseModel):
"""Apply a function to an integer and list of integers."""
a: int = Field(..., description="Integer")
b: list[int] = Field(..., description="List of ints")
runnable = RunnableLambda(f)
as_tool = runnable.as_tool(FSchema)
as_tool.invoke({"a": 3, "b": [1, 2]})
dict input, specifying schema via arg_types:
from typing import Any
from langchain_core.runnables import RunnableLambda
def f(x: dict[str, Any]) -> str:
return str(x["a"] * max(x["b"]))
runnable = RunnableLambda(f)
as_tool = runnable.as_tool(arg_types={"a": int, "b": list[int]})
as_tool.invoke({"a": 3, "b": [1, 2]})
str input:
__init__
¶
__init__(**kwargs: Any) -> None
Initialize the tool.
| RAISES | DESCRIPTION |
|---|---|
TypeError
|
If |
is_lc_serializable
classmethod
¶
is_lc_serializable() -> bool
Is this class serializable?
By design, even if a class inherits from Serializable, it is not serializable
by default. This is to prevent accidental serialization of objects that should
not be serialized.
| RETURNS | DESCRIPTION |
|---|---|
bool
|
Whether the class is serializable. Default is |
get_lc_namespace
classmethod
¶
lc_id
classmethod
¶
Return a unique identifier for this class for serialization purposes.
The unique identifier is a list of strings that describes the path to the object.
For example, for the class langchain.llms.openai.OpenAI, the id is
["langchain", "llms", "openai", "OpenAI"].
to_json
¶
Serialize the Runnable to JSON.
| RETURNS | DESCRIPTION |
|---|---|
SerializedConstructor | SerializedNotImplemented
|
A JSON-serializable representation of the |
to_json_not_implemented
¶
Serialize a "not implemented" object.
| RETURNS | DESCRIPTION |
|---|---|
SerializedNotImplemented
|
|
configurable_fields
¶
configurable_fields(
**kwargs: AnyConfigurableField,
) -> RunnableSerializable[Input, Output]
Configure particular Runnable fields at runtime.
| PARAMETER | DESCRIPTION |
|---|---|
**kwargs
|
A dictionary of
TYPE:
|
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If a configuration key is not found in the |
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Input, Output]
|
A new |
from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI
model = ChatOpenAI(max_tokens=20).configurable_fields(
max_tokens=ConfigurableField(
id="output_token_number",
name="Max tokens in the output",
description="The maximum number of tokens in the output",
)
)
# max_tokens = 20
print("max_tokens_20: ", model.invoke("tell me something about chess").content)
# max_tokens = 200
print(
"max_tokens_200: ",
model.with_config(configurable={"output_token_number": 200})
.invoke("tell me something about chess")
.content,
)
configurable_alternatives
¶
configurable_alternatives(
which: ConfigurableField,
*,
default_key: str = "default",
prefix_keys: bool = False,
**kwargs: Runnable[Input, Output] | Callable[[], Runnable[Input, Output]],
) -> RunnableSerializable[Input, Output]
Configure alternatives for Runnable objects that can be set at runtime.
| PARAMETER | DESCRIPTION |
|---|---|
which
|
The
TYPE:
|
default_key
|
The default key to use if no alternative is selected.
TYPE:
|
prefix_keys
|
Whether to prefix the keys with the
TYPE:
|
**kwargs
|
A dictionary of keys to
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
RunnableSerializable[Input, Output]
|
A new |
from langchain_anthropic import ChatAnthropic
from langchain_core.runnables.utils import ConfigurableField
from langchain_openai import ChatOpenAI
model = ChatAnthropic(
model_name="claude-sonnet-4-5-20250929"
).configurable_alternatives(
ConfigurableField(id="llm"),
default_key="anthropic",
openai=ChatOpenAI(),
)
# uses the default model ChatAnthropic
print(model.invoke("which organization created you?").content)
# uses ChatOpenAI
print(
model.with_config(configurable={"llm": "openai"})
.invoke("which organization created you?")
.content
)
__init_subclass__
¶
__init_subclass__(**kwargs: Any) -> None
Validate the tool class definition during subclass creation.
| PARAMETER | DESCRIPTION |
|---|---|
**kwargs
|
Additional keyword arguments passed to the parent class.
TYPE:
|
| RAISES | DESCRIPTION |
|---|---|
SchemaAnnotationError
|
If |
run
¶
run(
tool_input: str | dict[str, Any],
verbose: bool | None = None,
start_color: str | None = "green",
color: str | None = "green",
callbacks: Callbacks = None,
*,
tags: list[str] | None = None,
metadata: dict[str, Any] | None = None,
run_name: str | None = None,
run_id: UUID | None = None,
config: RunnableConfig | None = None,
tool_call_id: str | None = None,
**kwargs: Any,
) -> Any
Run the tool.
| PARAMETER | DESCRIPTION |
|---|---|
tool_input
|
The input to the tool. |
verbose
|
Whether to log the tool's progress.
TYPE:
|
start_color
|
The color to use when starting the tool.
TYPE:
|
color
|
The color to use when ending the tool.
TYPE:
|
callbacks
|
Callbacks to be called during tool execution.
TYPE:
|
tags
|
Optional list of tags associated with the tool. |
metadata
|
Optional metadata associated with the tool. |
run_name
|
The name of the run.
TYPE:
|
run_id
|
The id of the run.
TYPE:
|
config
|
The configuration for the tool.
TYPE:
|
tool_call_id
|
The id of the tool call.
TYPE:
|
**kwargs
|
Keyword arguments to be passed to tool callbacks (event handler)
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Any
|
The output of the tool. |
| RAISES | DESCRIPTION |
|---|---|
ToolException
|
If an error occurs during tool execution. |
arun
async
¶
arun(
tool_input: str | dict,
verbose: bool | None = None,
start_color: str | None = "green",
color: str | None = "green",
callbacks: Callbacks = None,
*,
tags: list[str] | None = None,
metadata: dict[str, Any] | None = None,
run_name: str | None = None,
run_id: UUID | None = None,
config: RunnableConfig | None = None,
tool_call_id: str | None = None,
**kwargs: Any,
) -> Any
Run the tool asynchronously.
| PARAMETER | DESCRIPTION |
|---|---|
tool_input
|
The input to the tool. |
verbose
|
Whether to log the tool's progress.
TYPE:
|
start_color
|
The color to use when starting the tool.
TYPE:
|
color
|
The color to use when ending the tool.
TYPE:
|
callbacks
|
Callbacks to be called during tool execution.
TYPE:
|
tags
|
Optional list of tags associated with the tool. |
metadata
|
Optional metadata associated with the tool. |
run_name
|
The name of the run.
TYPE:
|
run_id
|
The id of the run.
TYPE:
|
config
|
The configuration for the tool.
TYPE:
|
tool_call_id
|
The id of the tool call.
TYPE:
|
**kwargs
|
Keyword arguments to be passed to tool callbacks
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Any
|
The output of the tool. |
| RAISES | DESCRIPTION |
|---|---|
ToolException
|
If an error occurs during tool execution. |
initialize_client
¶
initialize_client() -> AzureLogicAppTool
Initialize the Azure Logic Apps client.
register_logic_app
¶
Retrieves and stores a callback URL for a specific Logic App + trigger.
Raises a ValueError if the callback URL is missing.
AIServicesToolkit
¶
Bases: BaseToolkit, AIServicesService
Toolkit for Azure AI Services.
| METHOD | DESCRIPTION |
|---|---|
get_tools |
Get the tools in the toolkit. |
validate_environment |
Validate that required values are present in the environment. |
project_endpoint
class-attribute
instance-attribute
¶
project_endpoint: str | None = None
The project endpoint associated with the AI project. If this is specified,
then the endpoint parameter becomes optional and credential has to be of type
TokenCredential.
endpoint
class-attribute
instance-attribute
¶
endpoint: str | None = None
The endpoint of the specific service to connect to. If you are connecting to a model, use the URL of the model deployment.
credential
class-attribute
instance-attribute
¶
credential: str | AzureKeyCredential | TokenCredential | None = None
The API key or credential to use to connect to the service. If using a project
endpoint, this must be of type TokenCredential since only Microsoft EntraID is
supported.
api_version
class-attribute
instance-attribute
¶
api_version: str | None = None
The API version to use with Azure. If None, the default version is used.
client_kwargs
class-attribute
instance-attribute
¶
Additional keyword arguments to pass to the client.
langchain_azure_ai.vectorstores
¶
Vector store stores embedded data and performs vector search.
One of the most common ways to store and search over unstructured data is to embed it and store the resulting embedding vectors, and then query the store and retrieve the data that are 'most similar' to the embedded query.
Class hierarchy:
VectorStore --> <name> # Examples: AzureSearch, FAISS, Milvus
BaseRetriever --> VectorStoreRetriever --> <name>Retriever # Example: AzureAISearchRetriever
Main helpers:
AzureCosmosDBMongoVCoreVectorSearch
¶
Bases: VectorStore
Azure Cosmos DB for MongoDB vCore vector store.
To use, you should have both:
- the pymongo python package installed
- a connection string associated with a MongoDB VCore Cluster
Example
. code-block:: python
from langchain_azure_ai.vectorstores.azure_cosmos_db import
AzureCosmosDBMongoVCoreVectorSearch
from langchain.embeddings.openai import OpenAIEmbeddings
from pymongo import MongoClient
mongo_client = MongoClient("<YOUR-CONNECTION-STRING>")
collection = mongo_client["<db_name>"]["<collection_name>"]
embeddings = OpenAIEmbeddings()
vectorstore = AzureCosmosDBMongoVCoreVectorSearch(collection, embeddings)
| METHOD | DESCRIPTION |
|---|---|
get_by_ids |
Get documents by their IDs. |
aget_by_ids |
Async get documents by their IDs. |
adelete |
Async delete by vector ID or other criteria. |
aadd_texts |
Async run more texts through the embeddings and add to the |
add_documents |
Add or update documents in the |
aadd_documents |
Async run more documents through the embeddings and add to the |
search |
Return docs most similar to query using a specified search type. |
asearch |
Async return docs most similar to query using a specified search type. |
asimilarity_search_with_score |
Async run similarity search with distance. |
similarity_search_with_relevance_scores |
Return docs and relevance scores in the range |
asimilarity_search_with_relevance_scores |
Async return docs and relevance scores in the range |
asimilarity_search |
Async return docs most similar to query. |
similarity_search_by_vector |
Return docs most similar to embedding vector. |
asimilarity_search_by_vector |
Async return docs most similar to embedding vector. |
amax_marginal_relevance_search |
Async return docs selected using the maximal marginal relevance. |
amax_marginal_relevance_search_by_vector |
Async return docs selected using the maximal marginal relevance. |
from_documents |
Return |
afrom_documents |
Async return |
afrom_texts |
Async return |
as_retriever |
Return |
__init__ |
Constructor for AzureCosmosDBMongoVCoreVectorSearch. |
get_index_name |
Returns the index name. |
from_connection_string |
Creates an Instance of AzureCosmosDBMongoVCoreVectorSearch from a Connection String. |
index_exists |
Verifies if the specified index name during instance construction exists on the collection. |
delete_index |
Deletes the index specified during instance construction if it exists. |
create_index |
Creates an index using the index name specified at instance construction. |
create_filter_index |
Creates a filter index. |
add_texts |
Used to Load Documents into the collection. |
from_texts |
Creates Azure CosmosDB MongoVCore Vector Store using the texts provided. |
delete |
Removes the documents with the list of documentIds provided from the collection. |
delete_document_by_id |
Removes a Specific Document by Id. |
similarity_search_with_score |
Returns a list of similar documents with their scores. |
similarity_search |
Returns a list of similar documents. |
max_marginal_relevance_search_by_vector |
Retrieves the docs with similarity scores. |
max_marginal_relevance_search |
Retrieves the similar docs. |
get_collection |
Returns the collection. |
get_by_ids
¶
Get documents by their IDs.
The returned documents are expected to have the ID field set to the ID of the document in the vector store.
Fewer documents may be returned than requested if some IDs are not found or if there are duplicated IDs.
Users should not assume that the order of the returned documents matches the order of the input IDs. Instead, users should rely on the ID field of the returned documents.
This method should NOT raise exceptions if no documents are found for some IDs.
| PARAMETER | DESCRIPTION |
|---|---|
ids
|
List of IDs to retrieve. |
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
aget_by_ids
async
¶
Async get documents by their IDs.
The returned documents are expected to have the ID field set to the ID of the document in the vector store.
Fewer documents may be returned than requested if some IDs are not found or if there are duplicated IDs.
Users should not assume that the order of the returned documents matches the order of the input IDs. Instead, users should rely on the ID field of the returned documents.
This method should NOT raise exceptions if no documents are found for some IDs.
| PARAMETER | DESCRIPTION |
|---|---|
ids
|
List of IDs to retrieve. |
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
adelete
async
¶
Async delete by vector ID or other criteria.
| PARAMETER | DESCRIPTION |
|---|---|
ids
|
List of IDs to delete. If |
**kwargs
|
Other keyword arguments that subclasses might use.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
bool | None
|
|
aadd_texts
async
¶
aadd_texts(
texts: Iterable[str],
metadatas: list[dict] | None = None,
*,
ids: list[str] | None = None,
**kwargs: Any,
) -> list[str]
Async run more texts through the embeddings and add to the VectorStore.
| PARAMETER | DESCRIPTION |
|---|---|
texts
|
Iterable of strings to add to the |
metadatas
|
Optional list of metadatas associated with the texts. |
ids
|
Optional list |
**kwargs
|
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[str]
|
List of IDs from adding the texts into the |
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If the number of metadatas does not match the number of texts. |
ValueError
|
If the number of IDs does not match the number of texts. |
add_documents
¶
Add or update documents in the VectorStore.
| PARAMETER | DESCRIPTION |
|---|---|
documents
|
Documents to add to the |
**kwargs
|
Additional keyword arguments. If kwargs contains IDs and documents contain ids, the IDs in the kwargs will receive precedence.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[str]
|
List of IDs of the added texts. |
aadd_documents
async
¶
search
¶
Return docs most similar to query using a specified search type.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Input text.
TYPE:
|
search_type
|
Type of search to perform. Can be
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If |
asearch
async
¶
Async return docs most similar to query using a specified search type.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Input text.
TYPE:
|
search_type
|
Type of search to perform. Can be
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If |
asimilarity_search_with_score
async
¶
similarity_search_with_relevance_scores
¶
similarity_search_with_relevance_scores(
query: str, k: int = 4, **kwargs: Any
) -> list[tuple[Document, float]]
Return docs and relevance scores in the range [0, 1].
0 is dissimilar, 1 is most similar.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Input text.
TYPE:
|
k
|
Number of
TYPE:
|
**kwargs
|
kwargs to be passed to similarity search. Should include
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List of tuples of |
asimilarity_search_with_relevance_scores
async
¶
asimilarity_search_with_relevance_scores(
query: str, k: int = 4, **kwargs: Any
) -> list[tuple[Document, float]]
Async return docs and relevance scores in the range [0, 1].
0 is dissimilar, 1 is most similar.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Input text.
TYPE:
|
k
|
Number of
TYPE:
|
**kwargs
|
kwargs to be passed to similarity search. Should include
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List of tuples of |
asimilarity_search
async
¶
Async return docs most similar to query.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Input text.
TYPE:
|
k
|
Number of
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
similarity_search_by_vector
¶
Return docs most similar to embedding vector.
| PARAMETER | DESCRIPTION |
|---|---|
embedding
|
Embedding to look up documents similar to. |
k
|
Number of
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
asimilarity_search_by_vector
async
¶
Async return docs most similar to embedding vector.
| PARAMETER | DESCRIPTION |
|---|---|
embedding
|
Embedding to look up documents similar to. |
k
|
Number of
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
amax_marginal_relevance_search
async
¶
amax_marginal_relevance_search(
query: str, k: int = 4, fetch_k: int = 20, lambda_mult: float = 0.5, **kwargs: Any
) -> list[Document]
Async return docs selected using the maximal marginal relevance.
Maximal marginal relevance optimizes for similarity to query AND diversity among selected documents.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Text to look up documents similar to.
TYPE:
|
k
|
Number of
TYPE:
|
fetch_k
|
Number of
TYPE:
|
lambda_mult
|
Number between
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
amax_marginal_relevance_search_by_vector
async
¶
amax_marginal_relevance_search_by_vector(
embedding: list[float],
k: int = 4,
fetch_k: int = 20,
lambda_mult: float = 0.5,
**kwargs: Any,
) -> list[Document]
Async return docs selected using the maximal marginal relevance.
Maximal marginal relevance optimizes for similarity to query AND diversity among selected documents.
| PARAMETER | DESCRIPTION |
|---|---|
embedding
|
Embedding to look up documents similar to. |
k
|
Number of
TYPE:
|
fetch_k
|
Number of
TYPE:
|
lambda_mult
|
Number between
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
from_documents
classmethod
¶
from_documents(documents: list[Document], embedding: Embeddings, **kwargs: Any) -> Self
Return VectorStore initialized from documents and embeddings.
| PARAMETER | DESCRIPTION |
|---|---|
documents
|
List of |
embedding
|
Embedding function to use.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Self
|
|
afrom_documents
async
classmethod
¶
afrom_documents(
documents: list[Document], embedding: Embeddings, **kwargs: Any
) -> Self
Async return VectorStore initialized from documents and embeddings.
| PARAMETER | DESCRIPTION |
|---|---|
documents
|
List of |
embedding
|
Embedding function to use.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Self
|
|
afrom_texts
async
classmethod
¶
afrom_texts(
texts: list[str],
embedding: Embeddings,
metadatas: list[dict] | None = None,
*,
ids: list[str] | None = None,
**kwargs: Any,
) -> Self
Async return VectorStore initialized from texts and embeddings.
| PARAMETER | DESCRIPTION |
|---|---|
texts
|
Texts to add to the |
embedding
|
Embedding function to use.
TYPE:
|
metadatas
|
Optional list of metadatas associated with the texts. |
ids
|
Optional list of IDs associated with the texts. |
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Self
|
|
as_retriever
¶
as_retriever(**kwargs: Any) -> VectorStoreRetriever
Return VectorStoreRetriever initialized from this VectorStore.
| PARAMETER | DESCRIPTION |
|---|---|
**kwargs
|
Keyword arguments to pass to the search function. Can include:
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
VectorStoreRetriever
|
Retriever class for |
Examples:
# Retrieve more documents with higher diversity
# Useful if your dataset has many similar documents
docsearch.as_retriever(
search_type="mmr", search_kwargs={"k": 6, "lambda_mult": 0.25}
)
# Fetch more documents for the MMR algorithm to consider
# But only return the top 5
docsearch.as_retriever(search_type="mmr", search_kwargs={"k": 5, "fetch_k": 50})
# Only retrieve documents that have a relevance score
# Above a certain threshold
docsearch.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={"score_threshold": 0.8},
)
# Only get the single most similar document from the dataset
docsearch.as_retriever(search_kwargs={"k": 1})
# Use a filter to only retrieve documents from a specific paper
docsearch.as_retriever(
search_kwargs={"filter": {"paper_title": "GPT-4 Technical Report"}}
)
__init__
¶
__init__(
collection: Collection,
embedding: Embeddings,
*,
index_name: str = "vectorSearchIndex",
text_key: str = "textContent",
embedding_key: str = "vectorContent",
application_name: str = "langchainpy",
)
Constructor for AzureCosmosDBMongoVCoreVectorSearch.
| PARAMETER | DESCRIPTION |
|---|---|
collection
|
MongoDB collection to add the texts to.
TYPE:
|
embedding
|
Text embedding model to use.
TYPE:
|
index_name
|
Name of the Atlas Search index.
TYPE:
|
text_key
|
MongoDB field that will contain the text for each document.
TYPE:
|
embedding_key
|
MongoDB field that will contain the embedding for each document.
TYPE:
|
application_name
|
The user agent for telemetry
TYPE:
|
get_index_name
¶
get_index_name() -> str
Returns the index name.
| RETURNS | DESCRIPTION |
|---|---|
str
|
Returns the index name |
from_connection_string
classmethod
¶
from_connection_string(
connection_string: str,
namespace: str,
embedding: Embeddings,
application_name: str = "langchainpy",
**kwargs: Any,
) -> AzureCosmosDBMongoVCoreVectorSearch
Creates an Instance of AzureCosmosDBMongoVCoreVectorSearch from a Connection String.
| PARAMETER | DESCRIPTION |
|---|---|
connection_string
|
The MongoDB vCore instance connection string
TYPE:
|
namespace
|
The namespace (database.collection)
TYPE:
|
embedding
|
The embedding utility
TYPE:
|
application_name
|
The user agent for telemetry
TYPE:
|
**kwargs
|
Dynamic keyword arguments
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
AzureCosmosDBMongoVCoreVectorSearch
|
an instance of the vector store |
index_exists
¶
index_exists() -> bool
Verifies if the specified index name during instance construction exists on the collection.
| RETURNS | DESCRIPTION |
|---|---|
bool
|
Returns True on success and False if no such index exists on the collection |
delete_index
¶
Deletes the index specified during instance construction if it exists.
create_index
¶
create_index(
num_lists: int = 100,
dimensions: int = 1536,
similarity: CosmosDBSimilarityType = COS,
kind: str = "vector-ivf",
m: int = 16,
ef_construction: int = 64,
max_degree: int = 32,
l_build: int = 50,
compression: CosmosDBVectorSearchCompression | None = None,
pq_compressed_dims: int | None = None,
pq_sample_size: int | None = None,
) -> dict[str, Any]
Creates an index using the index name specified at instance construction.
Setting the numLists parameter correctly is important for achieving good accuracy and performance. Since the vector store uses IVF as the indexing strategy, you should create the index only after you have loaded a large enough sample documents to ensure that the centroids for the respective buckets are faily distributed.
We recommend that numLists is set to documentCount/1000 for up to 1 million documents and to sqrt(documentCount) for more than 1 million documents. As the number of items in your database grows, you should tune numLists to be larger in order to achieve good latency performance for vector search.
If you're experimenting with a new scenario or creating a
small demo, you can start with numLists
set to 1 to perform a brute-force search across all vectors.
This should provide you with the most
accurate results from the vector search, however be aware that
the search speed and latency will be slow.
After your initial setup, you should go ahead and tune
the numLists parameter using the above guidance.
| PARAMETER | DESCRIPTION |
|---|---|
kind
|
Type of vector index to create. Possible options are: - vector-ivf - vector-hnsw - vector-diskann
TYPE:
|
num_lists
|
This integer is the number of clusters that the inverted file (IVF) index uses to group the vector data. We recommend that numLists is set to documentCount/1000 for up to 1 million documents and to sqrt(documentCount) for more than 1 million documents. Using a numLists value of 1 is akin to performing brute-force search, which has limited performance
TYPE:
|
dimensions
|
Number of dimensions for vector similarity. The maximum number of supported dimensions is 2000
TYPE:
|
similarity
|
Similarity metric to use with the IVF index. Possible options are: - CosmosDBSimilarityType.COS (cosine distance), - CosmosDBSimilarityType.L2 (Euclidean distance), and - CosmosDBSimilarityType.IP (inner product).
TYPE:
|
m
|
The max number of connections per layer (16 by default, minimum value is 2, maximum value is 100). Higher m is suitable for datasets with high dimensionality and/or high accuracy requirements.
TYPE:
|
ef_construction
|
the size of the dynamic candidate list for constructing the graph (64 by default, minimum value is 4, maximum value is 1000). Higher ef_construction will result in better index quality and higher accuracy, but it will also increase the time required to build the index. ef_construction has to be at least 2 * m
TYPE:
|
max_degree
|
Max number of neighbors. Default value is 32, range from 20 to 2048. Only vector-diskann search supports this for now.
TYPE:
|
l_build
|
l value for index building. Default value is 50, range from 10 to 500. Only vector-diskann search supports this for now.
TYPE:
|
compression
|
compression type for vector indexes. Product quantization compression is only supported for DISKANN and half precision compression is only supported for IVF and HNSW for now.
TYPE:
|
pq_compressed_dims
|
Number of dimensions after compression for product quantization. Must be less than original dimensions. Automatically calculated if omitted. Range: 1-8000.
TYPE:
|
pq_sample_size
|
Number of samples for PQ centroid training. Higher value means better quality but longer build time. Default: 1000. Range: 1000-100000.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
dict[str, Any]
|
An object describing the created index |
create_filter_index
¶
Creates a filter index.
add_texts
¶
add_texts(
texts: Iterable[str], metadatas: list[dict[str, Any]] | None = None, **kwargs: Any
) -> list
Used to Load Documents into the collection.
from_texts
classmethod
¶
from_texts(
texts: list[str],
embedding: Embeddings,
metadatas: list[dict] | None = None,
collection: Collection | None = None,
**kwargs: Any,
) -> AzureCosmosDBMongoVCoreVectorSearch
Creates Azure CosmosDB MongoVCore Vector Store using the texts provided.
delete
¶
Removes the documents with the list of documentIds provided from the collection.
delete_document_by_id
¶
delete_document_by_id(document_id: str | None = None) -> None
Removes a Specific Document by Id.
| PARAMETER | DESCRIPTION |
|---|---|
document_id
|
The document identifier
TYPE:
|
similarity_search_with_score
¶
similarity_search_with_score(
query: str,
k: int = 4,
kind: CosmosDBVectorSearchType = VECTOR_IVF,
pre_filter: dict | None = None,
ef_search: int = 40,
score_threshold: float = 0.0,
l_search: int = 40,
with_embedding: bool = False,
oversampling: float | None = 1.0,
) -> list[tuple[Document, float]]
Returns a list of similar documents with their scores.
similarity_search
¶
similarity_search(
query: str,
k: int = 4,
kind: CosmosDBVectorSearchType = VECTOR_IVF,
pre_filter: dict | None = None,
ef_search: int = 40,
score_threshold: float = 0.0,
l_search: int = 40,
with_embedding: bool = False,
oversampling: float | None = 1.0,
**kwargs: Any,
) -> list[Document]
Returns a list of similar documents.
max_marginal_relevance_search_by_vector
¶
max_marginal_relevance_search_by_vector(
embedding: list[float],
k: int = 4,
fetch_k: int = 20,
lambda_mult: float = 0.5,
kind: CosmosDBVectorSearchType = VECTOR_IVF,
pre_filter: dict | None = None,
ef_search: int = 40,
score_threshold: float = 0.0,
l_search: int = 40,
with_embedding: bool = False,
oversampling: float | None = 1.0,
**kwargs: Any,
) -> list[Document]
Retrieves the docs with similarity scores.
max_marginal_relevance_search
¶
max_marginal_relevance_search(
query: str,
k: int = 4,
fetch_k: int = 20,
lambda_mult: float = 0.5,
kind: CosmosDBVectorSearchType = VECTOR_IVF,
pre_filter: dict | None = None,
ef_search: int = 40,
score_threshold: float = 0.0,
l_search: int = 40,
with_embedding: bool = False,
oversampling: float | None = 1.0,
**kwargs: Any,
) -> list[Document]
Retrieves the similar docs.
AzureCosmosDBNoSqlVectorSearch
¶
Bases: VectorStore
Azure Cosmos DB for NoSQL vector store.
To use, you should have both:
- the azure-cosmos python package installed
You can read more about vector search, full text search and hybrid search using AzureCosmosDBNoSQL here: https://learn.microsoft.com/en-us/azure/cosmos-db/nosql/vector-search https://learn.microsoft.com/en-us/azure/cosmos-db/gen-ai/full-text-search https://learn.microsoft.com/en-us/azure/cosmos-db/gen-ai/hybrid-search
| METHOD | DESCRIPTION |
|---|---|
get_by_ids |
Get documents by their IDs. |
aget_by_ids |
Async get documents by their IDs. |
adelete |
Async delete by vector ID or other criteria. |
aadd_texts |
Async run more texts through the embeddings and add to the |
add_documents |
Add or update documents in the |
aadd_documents |
Async run more documents through the embeddings and add to the |
search |
Return docs most similar to query using a specified search type. |
asearch |
Async return docs most similar to query using a specified search type. |
asimilarity_search_with_score |
Async run similarity search with distance. |
similarity_search_with_relevance_scores |
Return docs and relevance scores in the range |
asimilarity_search_with_relevance_scores |
Async return docs and relevance scores in the range |
asimilarity_search |
Async return docs most similar to query. |
similarity_search_by_vector |
Return docs most similar to embedding vector. |
asimilarity_search_by_vector |
Async return docs most similar to embedding vector. |
amax_marginal_relevance_search |
Async return docs selected using the maximal marginal relevance. |
amax_marginal_relevance_search_by_vector |
Async return docs selected using the maximal marginal relevance. |
from_documents |
Return |
afrom_documents |
Async return |
afrom_texts |
Async return |
__init__ |
Constructor for AzureCosmosDBNoSqlVectorSearch. |
add_texts |
Run more texts through the embeddings and add to the vectorstore. |
from_texts |
Create an AzureCosmosDBNoSqlVectorSearch vectorstore from raw texts. |
from_connection_string_and_aad |
Initialize an AzureCosmosDBNoSqlVectorSearch vectorstore. |
from_connection_string_and_key |
Initialize an AzureCosmosDBNoSqlVectorSearch vectorstore. |
delete |
Removes the documents based on ids. |
delete_document_by_id |
Removes a Specific Document by id. |
similarity_search |
Return docs most similar to query. |
similarity_search_with_score |
Run similarity search with distance. |
max_marginal_relevance_search_by_vector |
Return docs selected using the maximal marginal relevance. |
max_marginal_relevance_search |
Return docs selected using the maximal marginal relevance. |
vector_search_with_score |
Returns the most similar indexed documents to the embeddings. |
vector_search_with_threshold |
Returns the most similar indexed documents to the embeddings. |
full_text_search |
Returns the documents based on the search text provided in the filters. |
full_text_ranking |
Returns the documents based on the search text provided full text rank filters. |
hybrid_search_with_score |
Returns the documents based on the embeddings and text provided full text rank filters. |
hybrid_search_with_threshold |
Returns the documents based on the embeddings and text provided full text rank filters. |
get_container |
Gets the container for the vector store. |
as_retriever |
Return AzureCosmosDBNoSqlVectorStoreRetriever initialized from this VectorStore. |
get_by_ids
¶
Get documents by their IDs.
The returned documents are expected to have the ID field set to the ID of the document in the vector store.
Fewer documents may be returned than requested if some IDs are not found or if there are duplicated IDs.
Users should not assume that the order of the returned documents matches the order of the input IDs. Instead, users should rely on the ID field of the returned documents.
This method should NOT raise exceptions if no documents are found for some IDs.
| PARAMETER | DESCRIPTION |
|---|---|
ids
|
List of IDs to retrieve. |
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
aget_by_ids
async
¶
Async get documents by their IDs.
The returned documents are expected to have the ID field set to the ID of the document in the vector store.
Fewer documents may be returned than requested if some IDs are not found or if there are duplicated IDs.
Users should not assume that the order of the returned documents matches the order of the input IDs. Instead, users should rely on the ID field of the returned documents.
This method should NOT raise exceptions if no documents are found for some IDs.
| PARAMETER | DESCRIPTION |
|---|---|
ids
|
List of IDs to retrieve. |
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
adelete
async
¶
Async delete by vector ID or other criteria.
| PARAMETER | DESCRIPTION |
|---|---|
ids
|
List of IDs to delete. If |
**kwargs
|
Other keyword arguments that subclasses might use.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
bool | None
|
|
aadd_texts
async
¶
aadd_texts(
texts: Iterable[str],
metadatas: list[dict] | None = None,
*,
ids: list[str] | None = None,
**kwargs: Any,
) -> list[str]
Async run more texts through the embeddings and add to the VectorStore.
| PARAMETER | DESCRIPTION |
|---|---|
texts
|
Iterable of strings to add to the |
metadatas
|
Optional list of metadatas associated with the texts. |
ids
|
Optional list |
**kwargs
|
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[str]
|
List of IDs from adding the texts into the |
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If the number of metadatas does not match the number of texts. |
ValueError
|
If the number of IDs does not match the number of texts. |
add_documents
¶
Add or update documents in the VectorStore.
| PARAMETER | DESCRIPTION |
|---|---|
documents
|
Documents to add to the |
**kwargs
|
Additional keyword arguments. If kwargs contains IDs and documents contain ids, the IDs in the kwargs will receive precedence.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[str]
|
List of IDs of the added texts. |
aadd_documents
async
¶
search
¶
Return docs most similar to query using a specified search type.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Input text.
TYPE:
|
search_type
|
Type of search to perform. Can be
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If |
asearch
async
¶
Async return docs most similar to query using a specified search type.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Input text.
TYPE:
|
search_type
|
Type of search to perform. Can be
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If |
asimilarity_search_with_score
async
¶
similarity_search_with_relevance_scores
¶
similarity_search_with_relevance_scores(
query: str, k: int = 4, **kwargs: Any
) -> list[tuple[Document, float]]
Return docs and relevance scores in the range [0, 1].
0 is dissimilar, 1 is most similar.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Input text.
TYPE:
|
k
|
Number of
TYPE:
|
**kwargs
|
kwargs to be passed to similarity search. Should include
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List of tuples of |
asimilarity_search_with_relevance_scores
async
¶
asimilarity_search_with_relevance_scores(
query: str, k: int = 4, **kwargs: Any
) -> list[tuple[Document, float]]
Async return docs and relevance scores in the range [0, 1].
0 is dissimilar, 1 is most similar.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Input text.
TYPE:
|
k
|
Number of
TYPE:
|
**kwargs
|
kwargs to be passed to similarity search. Should include
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List of tuples of |
asimilarity_search
async
¶
Async return docs most similar to query.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Input text.
TYPE:
|
k
|
Number of
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
similarity_search_by_vector
¶
Return docs most similar to embedding vector.
| PARAMETER | DESCRIPTION |
|---|---|
embedding
|
Embedding to look up documents similar to. |
k
|
Number of
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
asimilarity_search_by_vector
async
¶
Async return docs most similar to embedding vector.
| PARAMETER | DESCRIPTION |
|---|---|
embedding
|
Embedding to look up documents similar to. |
k
|
Number of
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
amax_marginal_relevance_search
async
¶
amax_marginal_relevance_search(
query: str, k: int = 4, fetch_k: int = 20, lambda_mult: float = 0.5, **kwargs: Any
) -> list[Document]
Async return docs selected using the maximal marginal relevance.
Maximal marginal relevance optimizes for similarity to query AND diversity among selected documents.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Text to look up documents similar to.
TYPE:
|
k
|
Number of
TYPE:
|
fetch_k
|
Number of
TYPE:
|
lambda_mult
|
Number between
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
amax_marginal_relevance_search_by_vector
async
¶
amax_marginal_relevance_search_by_vector(
embedding: list[float],
k: int = 4,
fetch_k: int = 20,
lambda_mult: float = 0.5,
**kwargs: Any,
) -> list[Document]
Async return docs selected using the maximal marginal relevance.
Maximal marginal relevance optimizes for similarity to query AND diversity among selected documents.
| PARAMETER | DESCRIPTION |
|---|---|
embedding
|
Embedding to look up documents similar to. |
k
|
Number of
TYPE:
|
fetch_k
|
Number of
TYPE:
|
lambda_mult
|
Number between
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
from_documents
classmethod
¶
from_documents(documents: list[Document], embedding: Embeddings, **kwargs: Any) -> Self
Return VectorStore initialized from documents and embeddings.
| PARAMETER | DESCRIPTION |
|---|---|
documents
|
List of |
embedding
|
Embedding function to use.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Self
|
|
afrom_documents
async
classmethod
¶
afrom_documents(
documents: list[Document], embedding: Embeddings, **kwargs: Any
) -> Self
Async return VectorStore initialized from documents and embeddings.
| PARAMETER | DESCRIPTION |
|---|---|
documents
|
List of |
embedding
|
Embedding function to use.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Self
|
|
afrom_texts
async
classmethod
¶
afrom_texts(
texts: list[str],
embedding: Embeddings,
metadatas: list[dict] | None = None,
*,
ids: list[str] | None = None,
**kwargs: Any,
) -> Self
Async return VectorStore initialized from texts and embeddings.
| PARAMETER | DESCRIPTION |
|---|---|
texts
|
Texts to add to the |
embedding
|
Embedding function to use.
TYPE:
|
metadatas
|
Optional list of metadatas associated with the texts. |
ids
|
Optional list of IDs associated with the texts. |
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Self
|
|
__init__
¶
__init__(
*,
cosmos_client: CosmosClient,
embedding: Embeddings,
vector_embedding_policy: dict[str, Any],
indexing_policy: dict[str, Any],
cosmos_container_properties: dict[str, Any],
cosmos_database_properties: dict[str, Any],
full_text_policy: dict[str, Any] | None = None,
vector_search_fields: dict[str, Any],
database_name: str = "vectorSearchDB",
container_name: str = "vectorSearchContainer",
search_type: str = "vector",
metadata_key: str = "metadata",
create_container: bool = True,
full_text_search_enabled: bool = False,
table_alias: str = "c",
)
Constructor for AzureCosmosDBNoSqlVectorSearch.
| PARAMETER | DESCRIPTION |
|---|---|
cosmos_client
|
Client used to connect to azure cosmosdb no sql account.
TYPE:
|
database_name
|
Name of the database to be created.
TYPE:
|
container_name
|
Name of the container to be created.
TYPE:
|
embedding
|
Text embedding model to use.
TYPE:
|
vector_embedding_policy
|
Vector Embedding Policy for the container. |
full_text_policy
|
Full Text Policy for the container. |
indexing_policy
|
Indexing Policy for the container. |
cosmos_container_properties
|
Container Properties for the container. |
cosmos_database_properties
|
Database Properties for the container. |
vector_search_fields
|
Vector Search and Text Search Fields for the container. |
search_type
|
CosmosDB Search Type to be performed.
TYPE:
|
metadata_key
|
Metadata key to use for data schema.
TYPE:
|
create_container
|
Set to true if the container does not exist.
TYPE:
|
full_text_search_enabled
|
Set to true if the full text search is enabled.
TYPE:
|
table_alias
|
Alias for the table to use in the WHERE clause.
TYPE:
|
add_texts
¶
add_texts(
texts: Iterable[str],
metadatas: list[dict] | None = None,
ids: list[str] | None = None,
**kwargs: Any,
) -> list[str]
Run more texts through the embeddings and add to the vectorstore.
| PARAMETER | DESCRIPTION |
|---|---|
texts
|
Iterable of strings to add to the vectorstore. |
metadatas
|
Optional list of metadatas associated with the texts. |
ids
|
Optional list of ids associated with the texts. |
**kwargs
|
Additional keyword arguments to pass to the embedding method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[str]
|
List of ids from adding the texts into the vectorstore. |
from_texts
classmethod
¶
from_texts(
texts: list[str],
embedding: Embeddings,
metadatas: list[dict] | None = None,
ids: list[str] | None = None,
**kwargs: Any,
) -> AzureCosmosDBNoSqlVectorSearch
Create an AzureCosmosDBNoSqlVectorSearch vectorstore from raw texts.
| PARAMETER | DESCRIPTION |
|---|---|
texts
|
the texts to insert. |
embedding
|
the embedding function to use in the store.
TYPE:
|
metadatas
|
metadata dicts for the texts. |
ids
|
id dicts for the texts. |
**kwargs
|
you can pass any argument that you would
to :meth:
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
AzureCosmosDBNoSqlVectorSearch
|
an |
from_connection_string_and_aad
classmethod
¶
from_connection_string_and_aad(
connection_string: str,
defaultAzureCredential: DefaultAzureCredential,
texts: list[str],
embedding: Embeddings,
metadatas: list[dict] | None = None,
ids: list[str] | None = None,
**kwargs: Any,
) -> AzureCosmosDBNoSqlVectorSearch
Initialize an AzureCosmosDBNoSqlVectorSearch vectorstore.
from_connection_string_and_key
classmethod
¶
from_connection_string_and_key(
connection_string: str,
key: str,
texts: list[str],
embedding: Embeddings,
metadatas: list[dict] | None = None,
ids: list[str] | None = None,
**kwargs: Any,
) -> AzureCosmosDBNoSqlVectorSearch
Initialize an AzureCosmosDBNoSqlVectorSearch vectorstore.
delete
¶
Removes the documents based on ids.
delete_document_by_id
¶
delete_document_by_id(document_id: str | None = None) -> None
Removes a Specific Document by id.
| PARAMETER | DESCRIPTION |
|---|---|
document_id
|
The document identifier
TYPE:
|
similarity_search
¶
similarity_search(
query: str,
k: int = 4,
with_embedding: bool = False,
search_type: str | None = "vector",
offset_limit: str | None = None,
projection_mapping: dict[str, Any] | None = None,
full_text_rank_filter: list[dict[str, str]] | None = None,
where: str | None = None,
weights: list[float] | None = None,
threshold: float | None = 0.5,
**kwargs: Any,
) -> list[Document]
Return docs most similar to query.
similarity_search_with_score
¶
similarity_search_with_score(
query: str,
k: int = 4,
with_embedding: bool = False,
search_type: str | None = "vector",
offset_limit: str | None = None,
full_text_rank_filter: list[dict[str, str]] | None = None,
projection_mapping: dict[str, Any] | None = None,
where: str | None = None,
weights: list[float] | None = None,
threshold: float | None = 0.5,
**kwargs: Any,
) -> list[tuple[Document, float]]
Run similarity search with distance.
max_marginal_relevance_search_by_vector
¶
max_marginal_relevance_search_by_vector(
embedding: list[float],
k: int = 4,
fetch_k: int = 20,
lambda_mult: float = 0.5,
search_type: str = "vector",
with_embedding: bool = False,
offset_limit: str | None = None,
full_text_rank_filter: list[dict[str, str]] | None = None,
projection_mapping: dict[str, Any] | None = None,
where: str | None = None,
weights: list[float] | None = None,
**kwargs: Any,
) -> list[Document]
Return docs selected using the maximal marginal relevance.
max_marginal_relevance_search
¶
max_marginal_relevance_search(
query: str,
k: int = 4,
fetch_k: int = 20,
lambda_mult: float = 0.5,
search_type: str = "vector",
with_embedding: bool = False,
offset_limit: str | None = None,
full_text_rank_filter: list[dict[str, str]] | None = None,
projection_mapping: dict[str, Any] | None = None,
where: str | None = None,
weights: list[float] | None = None,
**kwargs: Any,
) -> list[Document]
Return docs selected using the maximal marginal relevance.
vector_search_with_score
¶
vector_search_with_score(
search_type: str,
embeddings: list[float],
k: int = 4,
with_embedding: bool = False,
offset_limit: str | None = None,
*,
projection_mapping: dict[str, Any] | None = None,
where: str | None = None,
**kwargs: Any,
) -> list[tuple[Document, float]]
Returns the most similar indexed documents to the embeddings.
vector_search_with_threshold
¶
vector_search_with_threshold(
search_type: str,
embeddings: list[float],
threshold: float = 0.5,
k: int = 4,
with_embedding: bool = False,
offset_limit: str | None = None,
*,
projection_mapping: dict[str, Any] | None = None,
where: str | None = None,
**kwargs: Any,
) -> list[tuple[Document, float]]
Returns the most similar indexed documents to the embeddings.
full_text_search
¶
full_text_search(
search_type: str,
k: int = 4,
offset_limit: str | None = None,
*,
projection_mapping: dict[str, Any] | None = None,
where: str | None = None,
) -> list[tuple[Document, float]]
Returns the documents based on the search text provided in the filters.
full_text_ranking
¶
full_text_ranking(
search_type: str,
k: int = 4,
offset_limit: str | None = None,
*,
projection_mapping: dict[str, Any] | None = None,
full_text_rank_filter: list[dict[str, str]] | None = None,
where: str | None = None,
) -> list[tuple[Document, float]]
Returns the documents based on the search text provided full text rank filters.
hybrid_search_with_score
¶
hybrid_search_with_score(
search_type: str,
embeddings: list[float],
k: int = 4,
with_embedding: bool = False,
offset_limit: str | None = None,
*,
projection_mapping: dict[str, Any] | None = None,
full_text_rank_filter: list[dict[str, str]] | None = None,
where: str | None = None,
weights: list[float] | None = None,
) -> list[tuple[Document, float]]
Returns the documents based on the embeddings and text provided full text rank filters.
hybrid_search_with_threshold
¶
hybrid_search_with_threshold(
search_type: str,
embeddings: list[float],
threshold: float = 0.5,
k: int = 4,
with_embedding: bool = False,
offset_limit: str | None = None,
*,
projection_mapping: dict[str, Any] | None = None,
full_text_rank_filter: list[dict[str, str]] | None = None,
where: str | None = None,
weights: list[float] | None = None,
) -> list[tuple[Document, float]]
Returns the documents based on the embeddings and text provided full text rank filters.
as_retriever
¶
as_retriever(**kwargs: Any) -> AzureCosmosDBNoSqlVectorStoreRetriever
Return AzureCosmosDBNoSqlVectorStoreRetriever initialized from this VectorStore.
| PARAMETER | DESCRIPTION |
|---|---|
search_type
|
Overrides the type of search that
the Retriever should perform. Defaults to
TYPE:
|
search_kwargs
|
Keyword arguments to pass to the search function. Can include things like: score_threshold: Minimum relevance threshold for similarity_score_threshold fetch_k: Amount of documents to pass to MMR algorithm (Default: 20) lambda_mult: Diversity of results returned by MMR; 1 for minimum diversity and 0 for maximum. (Default: 0.5) filter: Filter by document metadata
TYPE:
|
**kwargs
|
Additional keyword arguments to pass to the
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
AzureCosmosDBNoSqlVectorStoreRetriever
|
Retriever class for VectorStore.
TYPE:
|
AzureSearch
¶
Bases: VectorStore
Azure Cognitive Search vector store.
| METHOD | DESCRIPTION |
|---|---|
get_by_ids |
Get documents by their IDs. |
aget_by_ids |
Async get documents by their IDs. |
add_documents |
Add or update documents in the |
aadd_documents |
Async run more documents through the embeddings and add to the |
search |
Return docs most similar to query using a specified search type. |
asearch |
Async return docs most similar to query using a specified search type. |
similarity_search_by_vector |
Return docs most similar to embedding vector. |
asimilarity_search_by_vector |
Async return docs most similar to embedding vector. |
max_marginal_relevance_search |
Return docs selected using the maximal marginal relevance. |
amax_marginal_relevance_search |
Async return docs selected using the maximal marginal relevance. |
max_marginal_relevance_search_by_vector |
Return docs selected using the maximal marginal relevance. |
amax_marginal_relevance_search_by_vector |
Async return docs selected using the maximal marginal relevance. |
from_documents |
Return |
afrom_documents |
Async return |
__init__ |
Initialize the AzureSearch vector store. |
__del__ |
Clean up resources by closing sync and async clients. |
add_texts |
Add texts data to an existing index. |
aadd_texts |
Asynchronously add texts data to an existing index. |
add_embeddings |
Add embeddings to an existing index. |
aadd_embeddings |
Add embeddings to an existing index. |
delete |
Delete by vector ID. |
adelete |
Asynchronously delete by vector ID. |
similarity_search |
Return documents most similar to the query using the specified search type. |
similarity_search_with_score |
Run similarity search with distance. |
asimilarity_search |
Asynchronously return documents most similar to the query. |
asimilarity_search_with_score |
Asynchronously run similarity search with distance. |
similarity_search_with_relevance_scores |
Return documents and scores above a threshold using similarity search. |
asimilarity_search_with_relevance_scores |
Asynchronously return documents and scores above a threshold. |
vector_search |
Returns the most similar indexed documents to the query text. |
avector_search |
Returns the most similar indexed documents to the query text. |
vector_search_with_score |
Return docs most similar to query. |
avector_search_with_score |
Return docs most similar to query. |
max_marginal_relevance_search_with_score |
Perform a search and return results that are reordered by MMR. |
amax_marginal_relevance_search_with_score |
Perform a search and return results that are reordered by MMR. |
hybrid_search |
Returns the most similar indexed documents to the query text. |
ahybrid_search |
Returns the most similar indexed documents to the query text. |
hybrid_search_with_score |
Return docs most similar to query with a hybrid query. |
ahybrid_search_with_score |
Return docs most similar to query with a hybrid query. |
hybrid_search_with_relevance_scores |
Return documents and scores above a threshold using hybrid search. |
ahybrid_search_with_relevance_scores |
Asynchronously return documents and scores above a threshold. |
hybrid_max_marginal_relevance_search_with_score |
Return docs most similar to query with hybrid query and MMR reordering. |
ahybrid_max_marginal_relevance_search_with_score |
Asynchronously return docs with hybrid query and MMR reordering. |
semantic_hybrid_search |
Returns the most similar indexed documents to the query text. |
asemantic_hybrid_search |
Returns the most similar indexed documents to the query text. |
semantic_hybrid_search_with_score |
Returns the most similar indexed documents to the query text. |
asemantic_hybrid_search_with_score |
Returns the most similar indexed documents to the query text. |
semantic_hybrid_search_with_score_and_rerank |
Return docs most similar to query with a hybrid query. |
asemantic_hybrid_search_with_score_and_rerank |
Return docs most similar to query with a hybrid query. |
from_texts |
Create Azure Search vector store from a list of texts. |
afrom_texts |
Asynchronously create Azure Search vector store from a list of texts. |
afrom_embeddings |
Asynchronously create Azure Search vector store from text embeddings. |
from_embeddings |
Create Azure Search vector store from text embeddings. |
as_retriever |
Return AzureSearchVectorStoreRetriever initialized from this VectorStore. |
get_by_ids
¶
Get documents by their IDs.
The returned documents are expected to have the ID field set to the ID of the document in the vector store.
Fewer documents may be returned than requested if some IDs are not found or if there are duplicated IDs.
Users should not assume that the order of the returned documents matches the order of the input IDs. Instead, users should rely on the ID field of the returned documents.
This method should NOT raise exceptions if no documents are found for some IDs.
| PARAMETER | DESCRIPTION |
|---|---|
ids
|
List of IDs to retrieve. |
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
aget_by_ids
async
¶
Async get documents by their IDs.
The returned documents are expected to have the ID field set to the ID of the document in the vector store.
Fewer documents may be returned than requested if some IDs are not found or if there are duplicated IDs.
Users should not assume that the order of the returned documents matches the order of the input IDs. Instead, users should rely on the ID field of the returned documents.
This method should NOT raise exceptions if no documents are found for some IDs.
| PARAMETER | DESCRIPTION |
|---|---|
ids
|
List of IDs to retrieve. |
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
add_documents
¶
Add or update documents in the VectorStore.
| PARAMETER | DESCRIPTION |
|---|---|
documents
|
Documents to add to the |
**kwargs
|
Additional keyword arguments. If kwargs contains IDs and documents contain ids, the IDs in the kwargs will receive precedence.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[str]
|
List of IDs of the added texts. |
aadd_documents
async
¶
search
¶
Return docs most similar to query using a specified search type.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Input text.
TYPE:
|
search_type
|
Type of search to perform. Can be
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If |
asearch
async
¶
Async return docs most similar to query using a specified search type.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Input text.
TYPE:
|
search_type
|
Type of search to perform. Can be
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If |
similarity_search_by_vector
¶
Return docs most similar to embedding vector.
| PARAMETER | DESCRIPTION |
|---|---|
embedding
|
Embedding to look up documents similar to. |
k
|
Number of
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
asimilarity_search_by_vector
async
¶
Async return docs most similar to embedding vector.
| PARAMETER | DESCRIPTION |
|---|---|
embedding
|
Embedding to look up documents similar to. |
k
|
Number of
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
max_marginal_relevance_search
¶
max_marginal_relevance_search(
query: str, k: int = 4, fetch_k: int = 20, lambda_mult: float = 0.5, **kwargs: Any
) -> list[Document]
Return docs selected using the maximal marginal relevance.
Maximal marginal relevance optimizes for similarity to query AND diversity among selected documents.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Text to look up documents similar to.
TYPE:
|
k
|
Number of
TYPE:
|
fetch_k
|
Number of
TYPE:
|
lambda_mult
|
Number between
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
amax_marginal_relevance_search
async
¶
amax_marginal_relevance_search(
query: str, k: int = 4, fetch_k: int = 20, lambda_mult: float = 0.5, **kwargs: Any
) -> list[Document]
Async return docs selected using the maximal marginal relevance.
Maximal marginal relevance optimizes for similarity to query AND diversity among selected documents.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Text to look up documents similar to.
TYPE:
|
k
|
Number of
TYPE:
|
fetch_k
|
Number of
TYPE:
|
lambda_mult
|
Number between
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
max_marginal_relevance_search_by_vector
¶
max_marginal_relevance_search_by_vector(
embedding: list[float],
k: int = 4,
fetch_k: int = 20,
lambda_mult: float = 0.5,
**kwargs: Any,
) -> list[Document]
Return docs selected using the maximal marginal relevance.
Maximal marginal relevance optimizes for similarity to query AND diversity among selected documents.
| PARAMETER | DESCRIPTION |
|---|---|
embedding
|
Embedding to look up documents similar to. |
k
|
Number of
TYPE:
|
fetch_k
|
Number of
TYPE:
|
lambda_mult
|
Number between
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
amax_marginal_relevance_search_by_vector
async
¶
amax_marginal_relevance_search_by_vector(
embedding: list[float],
k: int = 4,
fetch_k: int = 20,
lambda_mult: float = 0.5,
**kwargs: Any,
) -> list[Document]
Async return docs selected using the maximal marginal relevance.
Maximal marginal relevance optimizes for similarity to query AND diversity among selected documents.
| PARAMETER | DESCRIPTION |
|---|---|
embedding
|
Embedding to look up documents similar to. |
k
|
Number of
TYPE:
|
fetch_k
|
Number of
TYPE:
|
lambda_mult
|
Number between
TYPE:
|
**kwargs
|
Arguments to pass to the search method.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of |
from_documents
classmethod
¶
from_documents(documents: list[Document], embedding: Embeddings, **kwargs: Any) -> Self
Return VectorStore initialized from documents and embeddings.
| PARAMETER | DESCRIPTION |
|---|---|
documents
|
List of |
embedding
|
Embedding function to use.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Self
|
|
afrom_documents
async
classmethod
¶
afrom_documents(
documents: list[Document], embedding: Embeddings, **kwargs: Any
) -> Self
Async return VectorStore initialized from documents and embeddings.
| PARAMETER | DESCRIPTION |
|---|---|
documents
|
List of |
embedding
|
Embedding function to use.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Self
|
|
__init__
¶
__init__(
azure_search_endpoint: str,
azure_search_key: str | None,
index_name: str,
embedding_function: Callable | Embeddings,
search_type: str = "hybrid",
semantic_configuration_name: str | None = None,
fields: list[SearchField] | None = None,
vector_search: VectorSearch | None = None,
semantic_configurations: SemanticConfiguration
| list[SemanticConfiguration]
| None = None,
scoring_profiles: list[ScoringProfile] | None = None,
default_scoring_profile: str | None = None,
cors_options: CorsOptions | None = None,
*,
vector_search_dimensions: int | None = None,
additional_search_client_options: dict[str, Any] | None = None,
azure_ad_access_token: str | None = None,
azure_credential: TokenCredential | None = None,
azure_async_credential: AsyncTokenCredential | None = None,
**kwargs: Any,
)
Initialize the AzureSearch vector store.
| PARAMETER | DESCRIPTION |
|---|---|
azure_search_endpoint
|
The endpoint URL for Azure Cognitive Search.
TYPE:
|
azure_search_key
|
The API key for Azure Cognitive Search.
TYPE:
|
index_name
|
The name of the index to use.
TYPE:
|
embedding_function
|
The embedding function or object.
TYPE:
|
search_type
|
The type of search to perform (default: "hybrid").
TYPE:
|
semantic_configuration_name
|
Optional semantic configuration name.
TYPE:
|
fields
|
Optional list of search fields.
TYPE:
|
vector_search
|
Optional vector search configuration.
TYPE:
|
semantic_configurations
|
Optional semantic configurations.
TYPE:
|
scoring_profiles
|
Optional scoring profiles.
TYPE:
|
default_scoring_profile
|
Optional default scoring profile.
TYPE:
|
cors_options
|
Optional CORS options.
TYPE:
|
vector_search_dimensions
|
Optional vector search dimensions.
TYPE:
|
additional_search_client_options
|
Additional options for the search client. |
azure_ad_access_token
|
Optional Azure AD access token.
TYPE:
|
azure_credential
|
Optional Azure credential.
TYPE:
|
azure_async_credential
|
Optional async Azure credential.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
add_texts
¶
add_texts(
texts: Iterable[str],
metadatas: list[dict] | None = None,
*,
keys: list[str] | None = None,
**kwargs: Any,
) -> list[str]
Add texts data to an existing index.
aadd_texts
async
¶
aadd_texts(
texts: Iterable[str],
metadatas: list[dict] | None = None,
*,
keys: list[str] | None = None,
**kwargs: Any,
) -> list[str]
Asynchronously add texts data to an existing index.
| PARAMETER | DESCRIPTION |
|---|---|
texts
|
Iterable of text strings to add. |
metadatas
|
Optional list of metadata dicts. |
keys
|
Optional list of keys. |
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[str]
|
List of IDs for the added texts. |
add_embeddings
¶
add_embeddings(
text_embeddings: Iterable[tuple[str, list[float]]],
metadatas: list[dict] | None = None,
*,
keys: list[str] | None = None,
) -> list[str]
Add embeddings to an existing index.
aadd_embeddings
async
¶
aadd_embeddings(
text_embeddings: Iterable[tuple[str, list[float]]],
metadatas: list[dict] | None = None,
*,
keys: list[str] | None = None,
) -> list[str]
Add embeddings to an existing index.
delete
¶
adelete
async
¶
similarity_search
¶
similarity_search(
query: str, k: int = 4, *, search_type: str | None = None, **kwargs: Any
) -> list[Document]
Return documents most similar to the query using the specified search type.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
The query string.
TYPE:
|
k
|
Number of documents to return.
TYPE:
|
search_type
|
Optional search type override.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of similar documents. |
similarity_search_with_score
¶
similarity_search_with_score(
query: str, *, k: int = 4, **kwargs: Any
) -> list[tuple[Document, float]]
Run similarity search with distance.
asimilarity_search
async
¶
asimilarity_search(
query: str, k: int = 4, *, search_type: str | None = None, **kwargs: Any
) -> list[Document]
Asynchronously return documents most similar to the query.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
The query string.
TYPE:
|
k
|
Number of documents to return.
TYPE:
|
search_type
|
Optional search type override.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List of similar documents. |
asimilarity_search_with_score
async
¶
asimilarity_search_with_score(
query: str, *, k: int = 4, **kwargs: Any
) -> list[tuple[Document, float]]
Asynchronously run similarity search with distance.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
The query string.
TYPE:
|
k
|
Number of documents to return.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List of (Document, score) tuples. |
similarity_search_with_relevance_scores
¶
similarity_search_with_relevance_scores(
query: str, k: int = 4, *, score_threshold: float | None = None, **kwargs: Any
) -> list[tuple[Document, float]]
Return documents and scores above a threshold using similarity search.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
The query string.
TYPE:
|
k
|
Number of documents to return.
TYPE:
|
score_threshold
|
Optional minimum score threshold.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List of (Document, score) tuples. |
asimilarity_search_with_relevance_scores
async
¶
asimilarity_search_with_relevance_scores(
query: str, k: int = 4, *, score_threshold: float | None = None, **kwargs: Any
) -> list[tuple[Document, float]]
Asynchronously return documents and scores above a threshold.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
The query string.
TYPE:
|
k
|
Number of documents to return.
TYPE:
|
score_threshold
|
Optional minimum score threshold.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List of (Document, score) tuples. |
vector_search
¶
vector_search(
query: str, k: int = 4, *, filters: str | None = None, **kwargs: Any
) -> list[Document]
Returns the most similar indexed documents to the query text.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
The query text for which to find similar documents.
TYPE:
|
k
|
The number of documents to return. Default is 4.
TYPE:
|
filters
|
Filtering expression. Defaults to None.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List[Document]: A list of documents that are most similar to the query text. |
avector_search
async
¶
avector_search(
query: str, k: int = 4, *, filters: str | None = None, **kwargs: Any
) -> list[Document]
Returns the most similar indexed documents to the query text.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
The query text for which to find similar documents.
TYPE:
|
k
|
The number of documents to return. Default is 4.
TYPE:
|
filters
|
Filtering expression. Defaults to None.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List[Document]: A list of documents that are most similar to the query text. |
vector_search_with_score
¶
vector_search_with_score(
query: str, k: int = 4, filters: str | None = None, **kwargs: Any
) -> list[tuple[Document, float]]
Return docs most similar to query.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Text to look up documents similar to.
TYPE:
|
k
|
Number of Documents to return. Defaults to 4.
TYPE:
|
filters
|
Filtering expression. Defaults to None.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List[Tuple[Document, float]]: List of Documents most similar to the query and score for each |
avector_search_with_score
async
¶
avector_search_with_score(
query: str, k: int = 4, filters: str | None = None, **kwargs: Any
) -> list[tuple[Document, float]]
Return docs most similar to query.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Text to look up documents similar to.
TYPE:
|
k
|
Number of Documents to return. Defaults to 4.
TYPE:
|
filters
|
Filtering expression. Defaults to None.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List[Tuple[Document, float]]: List of Documents most similar to the query and score for each |
max_marginal_relevance_search_with_score
¶
max_marginal_relevance_search_with_score(
query: str,
k: int = 4,
fetch_k: int = 20,
lambda_mult: float = 0.5,
*,
filters: str | None = None,
**kwargs: Any,
) -> list[tuple[Document, float]]
Perform a search and return results that are reordered by MMR.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Text to look up documents similar to.
TYPE:
|
k
|
How many results to give. Defaults to 4.
TYPE:
|
fetch_k
|
Total results to select k from. Defaults to 20.
TYPE:
|
lambda_mult
|
Number between 0 and 1 that determines the degree of diversity among the results with 0 corresponding to maximum diversity and 1 to minimum diversity. Defaults to 0.5
TYPE:
|
filters
|
Filtering expression. Defaults to None.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List[Tuple[Document, float]]: List of Documents most similar to the query and score for each |
amax_marginal_relevance_search_with_score
async
¶
amax_marginal_relevance_search_with_score(
query: str,
k: int = 4,
fetch_k: int = 20,
lambda_mult: float = 0.5,
*,
filters: str | None = None,
**kwargs: Any,
) -> list[tuple[Document, float]]
Perform a search and return results that are reordered by MMR.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Text to look up documents similar to.
TYPE:
|
k
|
How many results to give. Defaults to 4.
TYPE:
|
fetch_k
|
Total results to select k from. Defaults to 20.
TYPE:
|
lambda_mult
|
Number between 0 and 1 that determines the degree of diversity among the results with 0 corresponding to maximum diversity and 1 to minimum diversity. Defaults to 0.5
TYPE:
|
filters
|
Filtering expression. Defaults to None.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List[Tuple[Document, float]]: List of Documents most similar to the query and score for each |
hybrid_search
¶
Returns the most similar indexed documents to the query text.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
The query text for which to find similar documents.
TYPE:
|
k
|
The number of documents to return. Default is 4.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List[Document]: A list of documents that are most similar to the query text. |
ahybrid_search
async
¶
Returns the most similar indexed documents to the query text.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
The query text for which to find similar documents.
TYPE:
|
k
|
The number of documents to return. Default is 4.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List[Document]: A list of documents that are most similar to the query text. |
hybrid_search_with_score
¶
hybrid_search_with_score(
query: str, k: int = 4, filters: str | None = None, **kwargs: Any
) -> list[tuple[Document, float]]
Return docs most similar to query with a hybrid query.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Text to look up documents similar to.
TYPE:
|
k
|
Number of Documents to return. Defaults to 4.
TYPE:
|
filters
|
Filtering expression. Defaults to None.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List of Documents most similar to the query and score for each |
ahybrid_search_with_score
async
¶
ahybrid_search_with_score(
query: str, k: int = 4, filters: str | None = None, **kwargs: Any
) -> list[tuple[Document, float]]
Return docs most similar to query with a hybrid query.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Text to look up documents similar to.
TYPE:
|
k
|
Number of Documents to return. Defaults to 4.
TYPE:
|
filters
|
Filtering expression. Defaults to None.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List of Documents most similar to the query and score for each |
hybrid_search_with_relevance_scores
¶
hybrid_search_with_relevance_scores(
query: str, k: int = 4, *, score_threshold: float | None = None, **kwargs: Any
) -> list[tuple[Document, float]]
Return documents and scores above a threshold using hybrid search.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
The query string.
TYPE:
|
k
|
Number of documents to return.
TYPE:
|
score_threshold
|
Optional minimum score threshold.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List of (Document, score) tuples. |
ahybrid_search_with_relevance_scores
async
¶
ahybrid_search_with_relevance_scores(
query: str, k: int = 4, *, score_threshold: float | None = None, **kwargs: Any
) -> list[tuple[Document, float]]
Asynchronously return documents and scores above a threshold.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
The query string.
TYPE:
|
k
|
Number of documents to return.
TYPE:
|
score_threshold
|
Optional minimum score threshold.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List of (Document, score) tuples. |
hybrid_max_marginal_relevance_search_with_score
¶
hybrid_max_marginal_relevance_search_with_score(
query: str,
k: int = 4,
fetch_k: int = 20,
lambda_mult: float = 0.5,
*,
filters: str | None = None,
**kwargs: Any,
) -> list[tuple[Document, float]]
Return docs most similar to query with hybrid query and MMR reordering.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Text to look up documents similar to.
TYPE:
|
k
|
Number of Documents to return. Defaults to 4.
TYPE:
|
fetch_k
|
Total results to select k from. Defaults to 20.
TYPE:
|
lambda_mult
|
Diversity of results returned by MMR; 1 for minimum diversity and 0 for maximum. Defaults to 0.5
TYPE:
|
filters
|
Filtering expression. Defaults to None.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List of Documents most similar to the query and score for each |
ahybrid_max_marginal_relevance_search_with_score
async
¶
ahybrid_max_marginal_relevance_search_with_score(
query: str,
k: int = 4,
fetch_k: int = 20,
lambda_mult: float = 0.5,
*,
filters: str | None = None,
**kwargs: Any,
) -> list[tuple[Document, float]]
Asynchronously return docs with hybrid query and MMR reordering.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Text to look up documents similar to.
TYPE:
|
k
|
Number of Documents to return. Defaults to 4.
TYPE:
|
fetch_k
|
Total results to select k from. Defaults to 20.
TYPE:
|
lambda_mult
|
Diversity of results returned by MMR; 1 for minimum diversity and 0 for maximum. Defaults to 0.5
TYPE:
|
filters
|
Filtering expression. Defaults to None.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List of Documents most similar to the query and score for each |
semantic_hybrid_search
¶
Returns the most similar indexed documents to the query text.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
The query text for which to find similar documents.
TYPE:
|
k
|
The number of documents to return. Default is 4.
TYPE:
|
filters
|
Filtering expression.
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List[Document]: A list of documents that are most similar to the query text. |
asemantic_hybrid_search
async
¶
Returns the most similar indexed documents to the query text.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
The query text for which to find similar documents.
TYPE:
|
k
|
The number of documents to return. Default is 4.
TYPE:
|
filters
|
Filtering expression.
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[Document]
|
List[Document]: A list of documents that are most similar to the query text. |
semantic_hybrid_search_with_score
¶
semantic_hybrid_search_with_score(
query: str,
k: int = 4,
score_type: Literal["score", "reranker_score"] = "score",
*,
score_threshold: float | None = None,
**kwargs: Any,
) -> list[tuple[Document, float]]
Returns the most similar indexed documents to the query text.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
The query text for which to find similar documents.
TYPE:
|
k
|
The number of documents to return. Default is 4.
TYPE:
|
score_type
|
Must either be "score" or "reranker_score". Defaulted to "score".
TYPE:
|
score_threshold
|
Minimum score threshold for results. Defaults to None.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List[Tuple[Document, float]]: A list of documents and their corresponding scores. |
asemantic_hybrid_search_with_score
async
¶
asemantic_hybrid_search_with_score(
query: str,
k: int = 4,
score_type: Literal["score", "reranker_score"] = "score",
*,
score_threshold: float | None = None,
**kwargs: Any,
) -> list[tuple[Document, float]]
Returns the most similar indexed documents to the query text.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
The query text for which to find similar documents.
TYPE:
|
k
|
The number of documents to return. Default is 4.
TYPE:
|
score_type
|
Must either be "score" or "reranker_score". Defaulted to "score".
TYPE:
|
score_threshold
|
Minimum score threshold for results. Defaults to None.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float]]
|
List[Tuple[Document, float]]: A list of documents and their corresponding scores. |
semantic_hybrid_search_with_score_and_rerank
¶
semantic_hybrid_search_with_score_and_rerank(
query: str, k: int = 4, *, filters: str | None = None, **kwargs: Any
) -> list[tuple[Document, float, float]]
Return docs most similar to query with a hybrid query.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Text to look up documents similar to.
TYPE:
|
k
|
Number of Documents to return. Defaults to 4.
TYPE:
|
filters
|
Filtering expression.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float, float]]
|
List of Documents most similar to the query and score for each |
asemantic_hybrid_search_with_score_and_rerank
async
¶
asemantic_hybrid_search_with_score_and_rerank(
query: str, k: int = 4, *, filters: str | None = None, **kwargs: Any
) -> list[tuple[Document, float, float]]
Return docs most similar to query with a hybrid query.
| PARAMETER | DESCRIPTION |
|---|---|
query
|
Text to look up documents similar to.
TYPE:
|
k
|
Number of Documents to return. Defaults to 4.
TYPE:
|
filters
|
Filtering expression.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
list[tuple[Document, float, float]]
|
List of Documents most similar to the query and score for each |
from_texts
classmethod
¶
from_texts(
texts: list[str],
embedding: Embeddings,
metadatas: list[dict] | None = None,
azure_search_endpoint: str = "",
azure_search_key: str = "",
azure_ad_access_token: str | None = None,
index_name: str = "langchain-index",
fields: list[SearchField] | None = None,
**kwargs: Any,
) -> AzureSearch
Create Azure Search vector store from a list of texts.
| PARAMETER | DESCRIPTION |
|---|---|
texts
|
List of texts to add to the vector store. |
embedding
|
Embeddings instance to use for encoding texts.
TYPE:
|
metadatas
|
Optional list of metadata dicts for each text. |
azure_search_endpoint
|
Azure Search service endpoint.
TYPE:
|
azure_search_key
|
Azure Search service API key.
TYPE:
|
azure_ad_access_token
|
Azure AD access token for authentication.
TYPE:
|
index_name
|
Name of the search index. Defaults to "langchain-index".
TYPE:
|
fields
|
List of search fields to use for the index.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
AzureSearch
|
The created vector store instance.
TYPE:
|
afrom_texts
async
classmethod
¶
afrom_texts(
texts: list[str],
embedding: Embeddings,
metadatas: list[dict] | None = None,
azure_search_endpoint: str = "",
azure_search_key: str = "",
azure_ad_access_token: str | None = None,
index_name: str = "langchain-index",
fields: list[SearchField] | None = None,
**kwargs: Any,
) -> "AzureSearch"
Asynchronously create Azure Search vector store from a list of texts.
| PARAMETER | DESCRIPTION |
|---|---|
texts
|
List of texts to add to the vector store. |
embedding
|
Embeddings instance to use for encoding texts.
TYPE:
|
metadatas
|
Optional list of metadata dicts for each text. |
azure_search_endpoint
|
Azure Search service endpoint.
TYPE:
|
azure_search_key
|
Azure Search service API key.
TYPE:
|
azure_ad_access_token
|
Azure AD access token for authentication.
TYPE:
|
index_name
|
Name of the search index. Defaults to "langchain-index".
TYPE:
|
fields
|
List of search fields to use for the index.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
AzureSearch
|
The created vector store instance.
TYPE:
|
afrom_embeddings
async
classmethod
¶
afrom_embeddings(
text_embeddings: Iterable[tuple[str, list[float]]],
embedding: Embeddings,
metadatas: list[dict] | None = None,
*,
azure_search_endpoint: str = "",
azure_search_key: str = "",
index_name: str = "langchain-index",
fields: list[SearchField] | None = None,
**kwargs: Any,
) -> AzureSearch
Asynchronously create Azure Search vector store from text embeddings.
| PARAMETER | DESCRIPTION |
|---|---|
text_embeddings
|
Iterable of (text, embedding) tuples. |
embedding
|
Embeddings instance to use for future queries.
TYPE:
|
metadatas
|
Optional list of metadata dicts for each text. |
azure_search_endpoint
|
Azure Search service endpoint.
TYPE:
|
azure_search_key
|
Azure Search service API key.
TYPE:
|
index_name
|
Name of the search index. Defaults to "langchain-index".
TYPE:
|
fields
|
List of search fields to use for the index.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
AzureSearch
|
The created vector store instance.
TYPE:
|
from_embeddings
classmethod
¶
from_embeddings(
text_embeddings: Iterable[tuple[str, list[float]]],
embedding: Embeddings,
metadatas: list[dict] | None = None,
*,
azure_search_endpoint: str = "",
azure_search_key: str = "",
index_name: str = "langchain-index",
fields: list[SearchField] | None = None,
**kwargs: Any,
) -> AzureSearch
Create Azure Search vector store from text embeddings.
| PARAMETER | DESCRIPTION |
|---|---|
text_embeddings
|
Iterable of (text, embedding) tuples. |
embedding
|
Embeddings instance to use for future queries.
TYPE:
|
metadatas
|
Optional list of metadata dicts for each text. |
azure_search_endpoint
|
Azure Search service endpoint.
TYPE:
|
azure_search_key
|
Azure Search service API key.
TYPE:
|
index_name
|
Name of the search index. Defaults to "langchain-index".
TYPE:
|
fields
|
List of search fields to use for the index.
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
AzureSearch
|
The created vector store instance.
TYPE:
|
as_retriever
¶
as_retriever(**kwargs: Any) -> AzureSearchVectorStoreRetriever
Return AzureSearchVectorStoreRetriever initialized from this VectorStore.
| PARAMETER | DESCRIPTION |
|---|---|
search_type
|
Overrides the type of search that
the Retriever should perform. Defaults to
TYPE:
|
search_kwargs
|
Keyword arguments to pass to the search function. Can include things like: score_threshold: Minimum relevance threshold for similarity_score_threshold fetch_k: Amount of documents to pass to MMR algorithm (Default: 20) lambda_mult: Diversity of results returned by MMR; 1 for minimum diversity and 0 for maximum. (Default: 0.5) filter: Filter by document metadata
TYPE:
|
**kwargs
|
Additional keyword arguments.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
AzureSearchVectorStoreRetriever
|
Retriever class for VectorStore.
TYPE:
|