### Example usage of LocalFileStore

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_150.md

Demonstrates creating a LocalFileStore, setting, getting, deleting, and iterating over keys. Shows typical workflow with byte values on a local filesystem.

```python
from langchain_classic.storage import LocalFileStore

# Instantiate the LocalFileStore with the root path
file_store = LocalFileStore("/path/to/rootn
# Set values for keys
file_store.mset([("key1", b"value1"), ("key2", b"value2")])

# Get values for keys
values = file_store.mget(["key1", "key2"])  # Returns [b"value1", b"value2"]

# keys
file_store.mdelete(["key1"])

# Iterate over keys
for key in file_store.yield_keys():
    print(key)  # noqa: T201

```

--------------------------------

### Create an Example Asynchronously

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_101.md

Creates a new example within a specified dataset. Accepts inputs and outputs, optionally associated with a dataset or dataset name. Returns an Example object.

```python
def create_example(inputs: dict[str, Any], outputs: dict[str, Any] | None = None, dataset_id: ID_TYPE | None = None, dataset_name: str | None = None, **kwargs: Any) -> Example:
  """Create an example."""
  pass
```

--------------------------------

### Complete aevaluate_existing Usage Example

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_161.md

Full workflow example showing dataset creation, initial evaluation with aevaluate, then re-evaluation using aevaluate_existing. Includes async result consumption and client cleanup. Demonstrates experiment evaluation lifecycle from data setup through result processing.

```python
import asyncio
import uuid
from langsmith import Client, aevaluate, aevaluate_existing

client = Client()
dataset_name = "__doctest_aevaluate_existing_" + uuid.uuid4().hex[:8]
dataset = client.create_dataset(dataset_name)
example = client.create_example(
    inputs={"question": "What is 2+2?"},
    outputs={"answer": "4"},
    dataset_id=dataset.id,
)

async def apredict(inputs: dict) -> dict:
    await asyncio.sleep(0.001)
    return {"output": "4"}

results = asyncio.run(
    aevaluate(
        apredict, data=dataset_name, experiment_prefix="doctest_experiment"
    )
)

# Consume all results to ensure evaluation is complete
async def consume_results():
    result_list = [r async for r in results]
    return len(result_list) > 0

asyncio.run(consume_results())

import time
time.sleep(3)

results = asyncio.run(
    aevaluate_existing(
        experiment_id,
        evaluators=[accuracy],
        summary_evaluators=[precision],
    )
)

client.delete_dataset(dataset_id=dataset.id)
```

--------------------------------

### Setup and Instantiate InMemoryVectorStore

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_119.md

Demonstrates how to set up and instantiate an in-memory vector store using `InMemoryVectorStore` from `langchain_core.vectorstores`. Requires `langchain-core` to be installed. The constructor takes an embedding function, such as `OpenAIEmbeddings`.

```python
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings

vector_store = InMemoryVectorStore(OpenAIEmbeddings())
```

--------------------------------

### Initialize Fully Configurable Chat Model with Default in Python

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_121.md

Example of setting up a fully configurable chat model with defaults and runtime overrides. Requires LangChain installation and provider packages. Allows configuring fields like model and temperature at invocation time.

```python
# pip install langchain langchain-openai langchain-anthropic

from langchain_classic.chat_models import init_chat_model

configurable_model_with_default = init_chat_model(
    "openai:gpt-4o",
    configurable_fields="any",  # This allows us to configure other params like temperature, max_tokens, etc at runtime.
    config_prefix="foo",
    temperature=0,
)

configurable_model_with_default.invoke("what's your name")
# GPT-4o response with temperature 0 (as set in default)

configurable_model_with_default.invoke(
    "what's your name",
    config={
        "configurable": {
            "foo_model": "anthropic:claude-sonnet-4-5-20250929",
            "foo_temperature": 0.6,
        }
    },
)
# Override default to use Sonnet 4.5 with temperature 0.6 to generate response
```

--------------------------------

### create_example Async

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_101.md

Create an example.

```APIDOC
## POST /create_example

### Description
Create an example.

### Method
POST

### Endpoint
/create_example

### Parameters
#### Request Body
- **inputs** (dict[str, Any]) - Required - The input data for the example.
- **outputs** (dict[str, Any] | None) - Optional - The output data for the example.
- **dataset_id** (ID_TYPE | None) - Optional - The ID of the dataset to associate this example with.
- **dataset_name** (str | None) - Optional - The name of the dataset to associate this example with.

```

--------------------------------

### Graph Compilation with InjectedStore

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_147.md

This example demonstrates how to compile a graph with a ToolNode and inject the persistent store. It shows the setup required for tools to access the injected store during execution.

```python
from langgraph.graph import StateGraph
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
tool_node = ToolNode([save_preference, get_preference])

graph = StateGraph(State)
graph.add_node("tools", tool_node)
compiled_graph = graph.compile(store=store)  # Store is injected automatically
```

--------------------------------

### List Examples Asynchronously

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_101.md

Lists examples within a specified dataset. Accepts optional keyword arguments for filtering and sorting.

```python
def list_examples(*, dataset_id: ID_TYPE | None = None, dataset_name: str | None = None, **kwargs: Any) -> AsyncIterator[Example]:
  """List examples."""
  pass
```

--------------------------------

### read_example Async

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_101.md

Read an example.

```APIDOC
## GET /read_example

### Description
Read an example.

### Method
GET

### Endpoint
/read_example

### Parameters
#### Path Parameters
- **example_id** (ID_TYPE) - Required - The ID of the example to read.

```

--------------------------------

### list_examples Async

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_101.md

List examples.

```APIDOC
## GET /list_examples

### Description
List examples.

### Method
GET

### Endpoint
/list_examples

### Parameters
#### Query Parameters
- **dataset_id** (ID_TYPE | None) - Optional - The ID of the dataset to filter examples by.
- **dataset_name** (str | None) - Optional - The name of the dataset to filter examples by.

```

--------------------------------

### String input example for as_tool

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_112.md

Demonstrates using `as_tool` with a string input. The example defines a simple function that processes a string and chains it with another function, then creates a tool from the resulting `Runnable`.

```python
from langchain_core.runnables import RunnableLambda

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

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

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

--------------------------------

### Read an Example Asynchronously

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_101.md

Retrieves an existing example from LangSmith, specified by its ID.

```python
def read_example(example_id: ID_TYPE) -> Example:
  """Read an example."""
  pass
```

--------------------------------

### Prepare Dataset Example

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_161.md

Example code demonstrating how to prepare a dataset for evaluation using the LangSmith client. It clones a public dataset and assigns it a name.

```python
from typing import Sequence
from langsmith import Client
from langsmith.evaluation import evaluate
from langsmith.schemas import Example, Run
client = Client()
dataset = client.clone_public_dataset(
    "https://smith.langchain.com/public/419dcab2-1d66-4b94-8901-0357ead390df/d"
)
dataset_name = "Evaluate Examples"
```

--------------------------------

### Basic Postgres Store Setup and Connection

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_49.md

Demonstrates the fundamental setup and connection process for LangGraph's PostgreSQL store. Shows how to import the necessary classes and establish a connection string for PostgreSQL. The connection string format follows standard PostgreSQL URI structure with authentication credentials, host, port, and database name. This is the minimal setup required before using any store operations.

```python
from langgraph.store.postgres import PostgresStore
from psycopg import Connection

conn_string = "postgresql://user:pass@localhost:5432/dbname"
```

--------------------------------

### AsyncPostgresStore Basic Setup and Usage (Python)

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_49.md

Demonstrates the fundamental setup and data manipulation operations (store, retrieve) using AsyncPostgresStore. Requires a PostgreSQL connection string.

```Python
from langgraph.store.postgres import AsyncPostgresStore

conn_string = "postgresql://user:pass@localhost:5432/dbname"

async with AsyncPostgresStore.from_conn_string(conn_string) as store:
    await store.setup()  # Run migrations. Done once

    # Store and retrieve data
    await store.aput(("users", "123"), "prefs", {"theme": "dark"})
    item = await store.aget(("users", "123"), "prefs")
```

--------------------------------

### Retrieve Similar Examples from Dataset - Python

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_101.md

Retrieves dataset examples that best match provided inputs. Requires few-shot indexing to be enabled on the dataset. Accepts inputs as a dictionary, a limit for results, a dataset ID, and an optional filter string. Returns a list of ExampleSearch objects.

```python
from langsmith import Client

client = Client()
await client.similar_examples(
    {"question": "When would i use the runnable generator"},
    limit=3,
    dataset_id="...",
)
```

--------------------------------

### Start MCP Session Explicitly (Python)

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_162.md

Illustrates how to explicitly start an MCP server session using the MultiServerMCPClient and then load MCP tools within that session. This approach is useful for managing sessions more directly. Dependencies include langchain_mcp_adapters.client and langchain_mcp_adapters.tools.

```python
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools

client = MultiServerMCPClient({...})
async with client.session("math") as session:
    tools = await load_mcp_tools(session)
```

--------------------------------

### Python Asynchronous Retriever Usage Example

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_119.md

Example demonstrating async retriever invocation using ainvoke method. Shows how to asynchronously query the retriever with a string input to get relevant documents. The method returns a list of Document objects and is the main entry point for async retriever operations.

```python
await retriever.ainvoke("query")
```

--------------------------------

### Initialize Non-Configurable Chat Models in Python

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_121.md

This example demonstrates initializing fixed chat models for specific providers without runtime configurability. It requires installation of relevant LangChain packages and uses the model name and provider strings to directly instantiate models. Outputs are model responses to invoked prompts.

```python
# pip install langchain langchain-openai langchain-anthropic langchain-google-vertexai

from langchain_classic.chat_models import init_chat_model

o3_mini = init_chat_model("openai:o3-mini", temperature=0)
claude_sonnet = init_chat_model("anthropic:claude-sonnet-4-5-20250929", temperature=0)
gemini_2-5_flash = init_chat_model(
    "google_vertexai:gemini-2.5-flash", temperature=0
)

o3_mini.invoke("what's your name")
claude_sonnet.invoke("what's your name")
gemini_2-5_flash.invoke("what's your name")
```

--------------------------------

### ToolCallLimitMiddleware Configuration Examples

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_114.md

Demonstrates how to configure and use the ToolCallLimitMiddleware to manage tool call limits within Langchain agents. It shows examples for continuing execution with blocked tools, stopping immediately, and raising exceptions when limits are exceeded.

```python
from langchain.agents.middleware.tool_call_limit import ToolCallLimitMiddleware
from langchain.agents import create_agent

# Block exceeded tools but let other tools and model continue
limiter = ToolCallLimitMiddleware(
    thread_limit=20,
    run_limit=10,
    exit_behavior="continue",  # default
)

agent = create_agent("openai:gpt-4o", middleware=[limiter])
```

```python
# End execution immediately with an AI message
limiter = ToolCallLimitMiddleware(run_limit=5, exit_behavior="end")

agent = create_agent("openai:gpt-4o", middleware=[limiter])
```

```python
# Strict limit with exception handling
limiter = ToolCallLimitMiddleware(tool_name="search", thread_limit=5, exit_behavior="error")

agent = create_agent("openai:gpt-4o", middleware=[limiter])

try:
    result = await agent.invoke({"messages": [HumanMessage("Task")]})
except ToolCallLimitExceededError as e:
    print(f"Search limit exceeded: {e}")
```

--------------------------------

### ExampleSearch Schema

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_135.md

Schema for retrieving examples via search, inheriting from ExampleBase. It includes a configuration class for schema settings.

```python
class ExampleSearch(ExampleBase):
    # Configuration class for the schema.
```

--------------------------------

### Typed dict input example for as_tool

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_112.md

Demonstrates using `as_tool` with a typed dictionary input. The example defines a typed dictionary `Args` and a function `f` that processes this input, then creates a tool from a `RunnableLambda` wrapping `f`.

```python
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]})
```

--------------------------------

### Bind async lifecycle listeners to Runnable using with_alisteners (Python)

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_154.md

Demonstrates attaching asynchronous start and end callbacks to a LangChain Runnable via `with_alisteners`. The example defines async listener functions, creates a RunnableLambda, binds the listeners, and runs two concurrent invocations. Requires `langchain-core` and Python 3.8+.

```text
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())
```

--------------------------------

### PutOp - Value Example

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_49.md

Example of a value dictionary that can be stored in the store.  Includes string, integer, and nested structure.

```python
{
"field1": "string value",
"field2": 123,
"nested": {"can": "contain", "any": "serializable data"}
}
```

--------------------------------

### Pick Dictionary Keys Example - Python

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_137.md

Illustrates using the 'pick' method to extract specific keys from a dictionary. The example creates a RunnableMap to process string and JSON transformations, then uses 'pick' to create a new runnable that only outputs the JSON part.

```python
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]")
```

--------------------------------

### Python: Tool Start Event Handler

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_179.md

Handles the event when a tool starts running. It accepts serialized tool data, input string, run ID, and optional parent run ID, tags, metadata, and inputs.

```python
def on_tool_start(
    serialized: dict[str, Any],
    input_str: str,
    *,
    run_id: UUID,
    parent_run_id: UUID | None = None,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    inputs: dict[str, Any] | None = None,
    **kwargs: Any,
) -> Any:
    """Run when the tool starts running."""
    pass
```

--------------------------------

### Handle tool start async in Python

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_179.md

Runs when a tool starts. Processes serialized tool data, input string, and optional run/parent IDs, returning a callback manager for the tool run.

```python
on_tool_start(
    serialized: dict[str, Any] | None,
    input_str: str,
    run_id: UUID | None = None,
    parent_run_id: UUID | None = None,
    **kwargs: Any,
) -> AsyncCallbackManagerForToolRun
```

--------------------------------

### Example Schema with Attachments

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_135.md

Represents a complete Example model, inheriting from ExampleBase. It includes fields for attachments and a URL, along with initialization and representation methods.

```python
class Example(BaseModel):
    attachments: dict[str, AttachmentInfo] | None = Field(default=None)
    url: str | None

    def __init__(self, _host_url: str | None = None, _tenant_id: UUID | None = None, **kwargs: Any) -> None:
        ...

    def __repr__(self) -> str:
        ...
```

--------------------------------

### Event Stream Example Data

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_119.md

Example of event stream data structure showing chain start, stream, and end events. Each event contains data, event type, metadata, name, and tags fields.

```json
[
    {
        "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": []
    }
]
```

--------------------------------

### abefore_agent Hook

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_114.md

Asynchronous lifecycle hook executed before the agent execution starts. Useful for async setup operations.

```APIDOC
## abefore_agent

### Description
Async logic to run before the agent execution starts.

### Signature
```python
abefore_agent(state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None
```

### Parameters
- **state** (`StateT`) - Required - Current agent state
- **runtime** (`Runtime[ContextT]`) - Required - Runtime context

### Returns
- `dict[str, Any] | None` - Optional modifications to the state
```

--------------------------------

### AsyncPostgresSaver: Setup Database

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_174.md

Asynchronously sets up the checkpoint database by creating necessary tables and running migrations. This method must be called by the user before the checkpointer is used for the first time.

```python
async def setup() -> None:
    # Implementation details...
```

--------------------------------

### PostgresSaver: Setup Database

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_174.md

Asynchronously sets up the checkpoint database. This method creates necessary tables and runs migrations, and must be called by the user before the checkpointer is used for the first time.

```python
from langgraph.checkpoint.postgres import PostgresSaver

# Assuming DB_URI is defined and PostgresSaver is instantiated
# Example: saver = PostgresSaver.from_conn_string(DB_URI)
# await saver.setup()
```

--------------------------------

### before_agent Hook

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_114.md

Lifecycle hook executed before the agent execution starts. Can modify state or perform setup tasks.

```APIDOC
## before_agent

### Description
Logic to run before the agent execution starts.

### Signature
```python
before_agent(state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None
```

### Parameters
- **state** (`StateT`) - Required - Current agent state
- **runtime** (`Runtime[ContextT]`) - Required - Runtime context

### Returns
- `dict[str, Any] | None` - Optional modifications to the state
```

--------------------------------

### Example: Get Runnable Output JSON Schema

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_154.md

Demonstrates how to obtain the JSON schema for the output of a RunnableLambda using the get_output_jsonschema method.

```python
from langchain_core.runnables import RunnableLambda

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

runnable = RunnableLambda(add_one)

print(runnable.get_output_jsonschema())

```

--------------------------------

### Get LangChain Namespace - Python

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_123.md

Returns the namespace of the LangChain object as a list of strings. For example, ['langchain', 'llms', 'openai'].

```python
get_lc_namespace() -> list[str]
```

--------------------------------

### get_lc_namespace

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_47.md

Get the namespace of the LangChain object. For example, if the class is langchain.llms.openai.OpenAI, then the namespace is ['langchain', 'llms', 'openai'].

```APIDOC
## get_lc_namespace

### Description
Get the namespace of the LangChain object. For example, if the class is `langchain.llms.openai.OpenAI`, then the namespace is `["langchain", "llms", "openai"]`.

### Method
CLASSMETHOD

### Returns
- **list[str]** - The namespace.
```

--------------------------------

### Set Up Checkpoint Database

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_174.md

Sets up the checkpoint database asynchronously, creating necessary tables if they don't exist. Should not be called directly by the user.

```Python
def setup() -> None:
    """Set up the checkpoint database asynchronously.

    This method creates the necessary tables in the SQLite database if they don't
    already exist. It is called automatically when needed and should not be called
    directly by the user.
    """
```

--------------------------------

### Initialize MultiServerMCPClient with Connections (Python)

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_162.md

Demonstrates how to initialize the MultiServerMCPClient with a dictionary of server connections. It shows basic usage where a new session is started for each tool call and how to retrieve all tools from the connected servers. Dependencies include langchain_mcp_adapters.client.

```python
from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient(
    {
        "math": {
            "command": "python",
            # Make sure to update to the full absolute path to your
            # math_server.py file
            "args": ["/path/to/math_server.py"],
            "transport": "stdio",
        },
        "weather": {
            # Make sure you start your weather server on port 8000
            "url": "http://localhost:8000/mcp",
            "transport": "streamable_http",
        }
    }
)
all_tools = await client.get_tools()
```

--------------------------------

### ExampleUpdate Schema

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_135.md

Schema for updating an existing example, potentially including attachments. It inherits from BaseModel and provides an __init__ method.

```python
class ExampleUpdate(BaseModel):
    def __init__(self, **data):
        ...
```

--------------------------------

### Handle Retriever Start Event in LangChain Python

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_179.md

This async callback executes at the start of a retriever operation. It takes serialized retriever data, query string, run context, metadata, and kwargs. Relies on LangChain's dict and UUID types; returns None. Useful for pre-retrieval setup, with no output.

```python
on_retriever_start(
    serialized: dict[str, Any],
    query: str,
    *,
    run_id: UUID,
    parent_run_id: UUID | None = None,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    **kwargs: Any,
) -> None
```

--------------------------------

### Setup dataset and run experiments for comparison

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_161.md

Demonstrates how to prepare a dataset and run different prompts for comparison. Uses LangSmith client to clone a public dataset and evaluates different system prompts in parallel.

```python
from typing import Sequence
from langsmith import Client
from langsmith.evaluation import evaluate
from langsmith.schemas import Example, Run
client = Client()
dataset = client.clone_public_dataset(
    "https://smith.langchain.com/public/419dcab2-1d66-4b94-8901-0357ead390df/d"
)
dataset_name = "Evaluate Examples"
```

```python
import functools
import openai
from langsmith.evaluation import evaluate
from langsmith.wrappers import wrap_openai
oai_client = openai.Client()
wrapped_client = wrap_openai(oai_client)
prompt_1 = "You are a helpful assistant."
prompt_2 = "You are an exceedingly helpful assistant."
def predict(inputs: dict, prompt: str) -> dict:
    completion = wrapped_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[\
            {"role": "system", "content": prompt},\
            {\
                "role": "user",\
                "content": f"Context: {inputs['context']}"\
                f"\n\ninputs['question']",\
            },\
        ],
    )
    return {"output": completion.choices[0].message.content}
results_1 = evaluate(
    functools.partial(predict, prompt=prompt_1),
    data=dataset_name,
    description="Evaluating our basic system prompt.",
    blocking=False,  # Run these experiments in parallel
)
View the evaluation results for experiment:...
results_2 = evaluate(
    functools.partial(predict, prompt=prompt_2),
    data=dataset_name,
    description="Evaluating our advanced system prompt.",
    blocking=False,
)
View the evaluation results for experiment:...
results_1.wait()
results_2.wait()
```

--------------------------------

### Store item without indexing

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_49.md

Example of storing an item with indexing disabled. The item remains accessible through get and search but won't have a vector representation.

```python
store.put(("docs",), "report", {"memory": "Will likes ai"}, index=False)
```

--------------------------------

### __init__ Method

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_119.md

Demonstrates the initialization method `__init__`.

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

--------------------------------

### Store item with default indexing

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_49.md

Example of storing an item where indexing behavior depends on the store's configuration. The item will be accessible through get and search operations.

```python
store.put(("docs",), "report", {"memory": "Will likes ai"})
```

--------------------------------

### Initialize store database

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_49.md

Sets up necessary database tables and runs migrations. Must be called before first use of the store.

```python
setup() -> None
```

--------------------------------

### Runnable Invoke Method (Python)

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_47.md

Synchronously invokes the retriever to get relevant documents. Accepts a query string, optional configuration, and additional keyword arguments. Returns a list of Document objects. Example: retriever.invoke('query').

```python
def 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.
    """
    # ... implementation ...
```

--------------------------------

### create_project Async

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_101.md

Create a project.

```APIDOC
## POST /create_project

### Description
Create a project.

### Method
POST

### Endpoint
/create_project

### Parameters
#### Request Body
- **project_name** (str) - Required - The name of the project.

### Request Example
{
  "project_name": "My New Project"
}

```

--------------------------------

### Bind arguments to Runnable in Python

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_112.md

This example shows how to bind additional arguments to a Runnable for chaining, allowing parameters not present in previous outputs. It uses ChatOllama model and StrOutputParser from LangChain libraries. Assumes LangChain and related dependencies are installed.

```Python
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'
```

--------------------------------

### Runnable Async Invoke Method (Python)

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_47.md

Asynchronously invokes the retriever to get relevant documents. Accepts a query string, optional configuration, and additional keyword arguments. Returns a list of Document objects. Example: await retriever.ainvoke('query').

```python
async def 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.
    """
    # ... implementation ...
```

--------------------------------

### Add listeners to Runnable in Python

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_112.md

This example illustrates binding lifecycle listeners to a Runnable for monitoring start, end, and error events. It uses RunnableLambda and Run schema from LangChain. Requires LangChain core libraries and assumes 'import time' is available.

```Python
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)
```

--------------------------------

### POST /datasets/:dataset_id/similar_examples

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_101.md

Retrieve dataset examples whose inputs best match the provided inputs. Requires few-shot indexing to be enabled for the dataset.

```APIDOC
## POST /datasets/:dataset_id/similar_examples

### Description
Retrieve the dataset examples whose inputs best match the current inputs. This method requires few-shot indexing to be enabled for the dataset.

### Method
POST

### Endpoint
`/datasets/:dataset_id/similar_examples`

### Parameters
#### Path Parameters
- **dataset_id** (ID_TYPE) - Required - The ID of the dataset to search over.

#### Query Parameters
- **limit** (int) - Required - The maximum number of examples to return.
- **filter** (str) - Optional - A filter string to apply to the search results. Uses the same syntax as the `filter` parameter in `list_runs()`. Only a subset of operations are supported. Defaults to None.

#### Request Body
- **inputs** (dict) - Required - The inputs to use as a search query. Must match the dataset input schema and be JSON serializable.
- **kwargs** (Any) - Optional - Additional keyword args to pass as part of the request body. Defaults to `{}`.

### Request Example
```json
{
  "inputs": {
    "question": "When would i use the runnable generator"
  },
  "limit": 3,
  "dataset_id": "your_dataset_id"
}
```

### Response
#### Success Response (200)
- **list[ExampleSearch]** - A list of ExampleSearch objects.

#### Response Example
```json
[
    {
        "inputs": {"question": "How do I cache a Chat model? What caches can I use?"},
        "outputs": {"answer": "You can use LangChain's caching layer..."},
        "metadata": null,
        "id": "b2ddd1c4-dff6-49ae-8544-f48e39053398",
        "dataset_id": "01b6ce0f-bfb6-4f48-bbb8-f19272135d40"
    }
]
```
```

--------------------------------

### Create LangChainStringEvaluator - Python

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_161.md

Demonstrates basic instantiation of LangChainStringEvaluator with a simple evaluator type. Shows the minimal required setup for creating an evaluator instance using string evaluator names like 'exact_match'. Requires langchain package to be installed.

```python
>>> evaluator = LangChainStringEvaluator("exact_match")
```

--------------------------------

### Initialize Filesystem Middleware

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_3.md

Initializes the filesystem middleware. Allows configuration of backend, system prompt, custom tool descriptions, and token limit for eviction.

```Python
def __init__(
    *,
    backend: BACKEND_TYPES | None = None,
    system_prompt: str | None = None,
    custom_tool_descriptions: dict[str, str] | None = None,
    tool_token_limit_before_evict: int | None = 20000,
) -> None
```

--------------------------------

### Encode Keys/Values in Langchain Storage (Python)

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_150.md

Demonstrates how to use EncoderBackedStore to wrap a key-value store with JSON encoding/decoding for keys and values. This example requires an abstract store implementation and defines custom encoder/serializer functions. It performs multi-set, get, and delete operations on integer keys and float values.

```python
import json

def key_encoder(key: int) -> str:
    return json.dumps(key)

def value_serializer(value: float) -> str:
    return json.dumps(value)

def value_deserializer(serialized_value: str) -> float:
    return json.loads(serialized_value)

# Create an instance of the abstract store
abstract_store = MyCustomStore()

# Create an instance of the encoder-backed store
store = EncoderBackedStore(
    store=abstract_store,
    key_encoder=key_encoder,
    value_serializer=value_serializer,
    value_deserializer=value_deserializer,
)

# Use the encoder-backed store methods
store.mset([(1, 3.14), (2, 2.718)])
values = store.mget([1, 2])  # Retrieves [3.14, 2.718]
store.mdelete([1, 2])  # Deletes the keys 1 and 2
```

--------------------------------

### Call Method - Map Run and Example to Dictionary

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_157.md

Maps the Run and Example to a dictionary. This special method (`__call__`) takes an Example object as input and returns a dictionary mapping the Run and Example.

```python
def __call__(example: Example) -> dict[str, str]:
  """Maps the Run and Example to a dictionary."""
```

--------------------------------

### Bind async listeners to a Runnable (Python)

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_157.md

Shows how to attach asynchronous start and end callbacks to a LangChain Runnable using the `with_alisteners` method. The example imports necessary modules, defines helper functions, creates a RunnableLambda, binds listeners, and concurrent invocations. Requires Python 3.8+, LangChain core library, and asyncio support.

```Python
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())
```

--------------------------------

### Map Example to Dictionary

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_157.md

Maps an Example or dataset row to a dictionary. This method takes an Example object as input and returns a dictionary representing the mapped data.

```python
def map(example: Example) -> dict[str, str]:
  """Maps the Example, or dataset row to a dictionary."""
```

--------------------------------

### ExampleCreate Schema

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_135.md

Schema for uploading an example, including support for attachments. It inherits from ExampleBase and includes an __init__ method for initialization from a dictionary.

```python
class ExampleCreate(BaseModel):
    def __init__(self, **data):
        ...
```

--------------------------------

### Dict input with args_schema for as_tool

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_112.md

Shows how to use `as_tool` with a dictionary input by specifying the schema via `args_schema`. The example defines a Pydantic `BaseModel` for the schema and a function that processes the dictionary input.

```python
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]})
```

--------------------------------

### LangSmithLoader Example (Python)

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_171.md

Demonstrates the usage of LangSmithLoader for loading LangSmith Dataset examples as Document objects.

```Python
from langchain_core.document_loaders import LangSmithLoader

loader = LangSmithLoader(dataset_id="...", limit=100)
docs = []
for doc in loader.lazy_load():
    docs.append(doc)
```

--------------------------------

### OpenAIFunctionsRouter Initialization and Core Methods

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_154.md

Demonstrates the initialization of OpenAIFunctionsRouter and its core methods for retrieving schema information, such as input and output schemas in Pydantic and JSON formats. It also includes methods for accessing configuration and graph representations.

```python
from langgraph.routing.openai_functions import OpenAIFunctionsRouter
from typing import Any

router = OpenAIFunctionsRouter()

# Get input schema
input_schema_pydantic = router.get_input_schema()
input_schema_json = router.get_input_jsonschema()

# Get output schema
output_schema_pydantic = router.get_output_schema()
output_schema_json = router.get_output_jsonschema()

# Get config schema
config_schema_pydantic = router.config_schema
config_schema_json = router.get_config_jsonschema()

# Get graph representation
graph = router.get_graph()

# Get prompts
prompts = router.get_prompts()

```

--------------------------------

### GET /threads/{thread_id}/state

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_137.md

Get the current state of a thread. This endpoint retrieves the latest state snapshot for a specified thread.

```APIDOC
## GET /threads/{thread_id}/state

### Description
Get the current state of a thread. This endpoint retrieves the latest state snapshot for a specified thread.

### Method
GET

### Endpoint
GET /threads/{thread_id}/state

### Parameters
#### Path Parameters
- **thread_id** (string) - Required - The unique identifier of the thread

#### Query Parameters
- **subgraphs** (boolean) - Optional - Include subgraphs in the state. Default is false

#### Request Body
No request body required for this endpoint.

### Response
#### Success Response (200)
- **StateSnapshot** (object) - The latest state of the thread

#### Response Example
{
  "values": {"key": "value"},
  "next": ["step1", "step2"],
  "checkpoint": {
    "id": "checkpoint_id"
  }
}
```

--------------------------------

### create_dataset Async

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_101.md

Create a dataset.

```APIDOC
## POST /create_dataset

### Description
Create a dataset.

### Method
POST

### Endpoint
/create_dataset

### Parameters
#### Request Body
- **dataset_name** (str) - Required - The name of the dataset.

```

--------------------------------

### Connect and Store Data with PostgresStore

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_49.md

Demonstrates how to establish a connection to a PostgreSQL database using a connection string and the `PostgresStore`. It shows basic operations like setting up the store, putting data, and retrieving data. This is suitable for general key-value storage.

```python
from langgraph.store.postgres import PostgresStore

conn_string = "postgresql://user:pass@localhost:5432/dbname"

with PostgresStore.from_conn_string(conn_string) as store:
    store.setup() # Run migrations. Done once

    # Store and retrieve data
    store.put(("users", "123"), "prefs", {"theme": "dark"})
    item = store.get(("users", "123"), "prefs")
```

--------------------------------

### StateGraph.add_edge

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_5.md

Add a directed edge from the start node (or list of start nodes) to the end node, defining the execution flow in the graph.

```APIDOC
## add_edge

### Description
Add a directed edge from the start node (or list of start nodes) to the end node.

### Method
INSTANCE METHOD

### Parameters
- **start_key** (str | list[str]) - The starting node(s)
- **end_key** (str) - The destination node

### Example
```python
graph.add_edge("A", "B")
graph.add_edge(["A", "C"], "D")
```
```

--------------------------------

### Initialize Tool Emulator - Python

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_114.md

Initializes the tool emulator with optional lists of tools to emulate and a model to use for emulation. If no tools are specified, all tools are emulated. An empty list emulates no tools.

```python
__init__(
    *, 
    tools: list[str | BaseTool] | None = None,
    model: str | BaseChatModel | None = None,
) -> None:
    """Initialize the tool emulator."""
    pass
```

--------------------------------

### Experiment-Level Precision Evaluator

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_161.md

An example of an experiment-level evaluator function in Python designed to calculate precision. It processes sequences of runs and examples.

```python
def precision(runs: Sequence[Run], examples: Sequence[Example]):
    # Experiment-level evaluator for precision.
    # TP / (TP + FP)
    predictions = [run.outputs["output"].lower() for run in runs]
```

--------------------------------

### Async Context Manager Entry Point - Python

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_162.md

The __aenter__ method acts as the entry point for an async context manager, returning a MultiServerMCPClient instance. It raises NotImplementedError since context manager support has been removed. No parameters are required.

```python
async __aenter__() -> MultiServerMCPClient
```

--------------------------------

### ExampleBase Schema

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_135.md

Base model for an example, serving as a foundation for more specific example schemas. It is defined using Pydantic's BaseModel.

```python
class ExampleBase(BaseModel):
    # Configuration class for the schema.
```

--------------------------------

### Create Basic ToolNode in Python

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_147.md

This example demonstrates basic ToolNode creation with a simple tool function for calculations. It uses the langchain.tools module and requires a sequence of tools or functions. Input is a tool list; output is a ToolNode instance. Limitations include dependency on LangChain libraries.

```python
from langchain.tools import ToolNode
from langchain_core.tools import tool

@tool
def calculator(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

tool_node = ToolNode([calculator])
```

--------------------------------

### Initialize LangGraph Client (__init__)

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_137.md

Initializes the LangGraph client, allowing configuration via URL, API key, headers, or pre-existing client instances. It handles the creation of default sync and async clients if not provided. At least one connection parameter (url, client, or sync_client) must be specified.

```python
class LangGraphClient:
    def __init__(
        self,
        assistant_id: str,
        /, 
        *,
        url: str | None = None,
        api_key: str | None = None,
        headers: dict[str, str] | None = None,
        client: LangGraphClient | None = None,
        sync_client: SyncLangGraphClient | None = None,
        config: RunnableConfig | None = None,
        name: str | None = None,
        distributed_tracing: bool = False,
    ):
        pass
```

--------------------------------

### Fetch checkpoint with get (Python)

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_174.md

The get method retrieves a checkpoint from the Postgres database based on a RunnableConfig. It returns the matching Checkpoint object or None if not found. This is a synchronous operation.

```python
def get(config: RunnableConfig) -> Checkpoint | None:
    # implementation omitted
    pass
```

--------------------------------

### Create Agent with Filesystem Middleware (Python)

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_3.md

This example demonstrates how to create an agent using the FilesystemMiddleware. It shows two scenarios: one with ephemeral storage (default) and another with hybrid storage using CompositeBackend for persistent storage.

```python
from deepagents.middleware.filesystem import FilesystemMiddleware
from deepagents.memory.backends import StateBackend, StoreBackend, CompositeBackend
from langchain.agents import create_agent

# Ephemeral storage only (default)
agent = create_agent(middleware=[FilesystemMiddleware()])

# With hybrid storage (ephemeral + persistent /memories/)
backend = CompositeBackend(default=StateBackend(), routesionado={"/memories/": StoreBackend()})
agent = create_agent(middleware=[FilesystemMiddleware(memory_backend=backend)])
```

--------------------------------

### Fetch Checkpoint Synchronously with get

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_174.md

The get method fetches a checkpoint based on the provided RunnableConfig. It returns the requested Checkpoint object if found, otherwise it returns None. This is a synchronous operation.

```python
get(config: RunnableConfig) -> Checkpoint | None
```

--------------------------------

### Start periodic TTL sweeper

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_49.md

Starts a background thread that periodically sweeps expired items. Returns a Future object for controlling the sweeper thread.

```python
start_ttl_sweeper(sweep_interval_minutes: int | None = None) -> Future[None]
```

--------------------------------

### SubAgentMiddleware Initialization and Usage (Python)

Source: https://github.com/tomato007tomats/lang-api-reference/blob/main/todos_md/page_3.md

Demonstrates basic and advanced usage of SubAgentMiddleware for creating agents. It shows how to initialize the middleware with default models and custom middleware for subagents.

```python
from langchain.agents.middleware.subagents import SubAgentMiddleware
from langchain.agents import create_agent

# Basic usage with defaults (no default middleware)agent = create_agent(
    "openai:gpt-4o",
    middleware=[
        SubAgentMiddleware(
            default_model="openai:gpt-4o",
            subagents=[],
        )
    ],
)

# Add custom middleware to subagentsagent = create_agent(
    "openai:gpt-4o",
    middleware=[
        SubAgentMiddleware(
            default_model="openai:gpt-4o",
            default_middleware=[TodoListMiddleware()],
            subagents=[],
        )
    ],
)
```