Use ReasonBlocks with LangChain

rb.middleware() returns a ReasonBlocksMiddleware — a langchain.agents.middleware.AgentMiddleware subclass that also acts as a context manager. You attach it to create_agent like any other middleware. It hooks before_agent, before_model, wrap_model_call, and after_agent to score each step, evaluate trajectory monitors server-side, inject steering text into the system message, and (optionally) route the model based on the FSM state.

Install

pip install reasonblocks "langchain>=1.0" langchain-anthropic langgraph

Initialize ReasonBlocks

Create one ReasonBlocks per process and call rb.middleware() per run. Each middleware instance is single-use.

import os
from reasonblocks import ReasonBlocks

rb = ReasonBlocks(
    api_key=os.environ["REASONBLOCKS_API_KEY"],
    base_url=os.environ.get("REASONBLOCKS_BASE_URL"),  # omit for the hosted API
)

The constructor does not auto-read REASONBLOCKS_API_KEY — your code must pass it explicitly. The base_url default is read from REASONBLOCKS_BASE_URL at import time by reasonblocks._settings; if you set that env var before importing reasonblocks, you can omit base_url here.

Wrap your agent in the context manager

The recommended pattern is with mw:. The context manager guarantees that the run_finish telemetry event fires on success, exception, or explicit failure marking.

from langchain.agents import create_agent

mw = rb.middleware(
    agent_name="bugfixer",
    task="fix the TypeError in handler.py",
    framework="langchain",
    model="anthropic:claude-haiku-4-5-20251001",
    codebase_id="my-org/my-repo",
)

agent = create_agent(
    model="anthropic:claude-haiku-4-5-20251001",
    tools=[search_codebase, read_file, edit_file, run_tests],
    system_prompt="You are a senior software engineer. Fix bugs by searching, reading, editing, and testing.",
    middleware=[mw],
)

with mw:
    result = agent.invoke(
        {"messages": [("user", "There's a TypeError in the request handler. Find and fix it.")]},
    )

Tag runs for the dashboard

Every parameter on rb.middleware() is optional. Anything not consumed by a named field rides along in metadata on the run row.

mw = rb.middleware(
    run_id="pr-42-attempt-3",                    # auto-generated UUID if omitted
    agent_name="bugfixer",
    task="fix the TypeError in handler.py",
    framework="langchain",                        # default
    model="anthropic:claude-haiku-4-5-20251001",  # display only
    codebase_id="my-org/my-repo",
    org_id="6d3f...",                             # "default" if omitted
    project_id="a91b...",                         # "default" if omitted
    metadata={"pr_number": 42, "experiment": "v2"},
)

When your api_key is a per-org rb_live_* key, the API overrides org_id / project_id with the key’s authoritative scope. Leaving them at "default" is fine.

Mark explicit failures

If the agent returns normally but the run was logically a failure (wrong answer, tests still failing), call mark_failure(reason=...) before the with block exits. Exceptions that escape the block are recorded automatically as failure: <ExceptionType>.

with mw:
    result = agent.invoke({"messages": [("user", "Fix the bug.")]})
    if not tests_pass(result):
        mw.mark_failure(reason="tests_still_failing")

Inspect step_log

mw.step_log is a list[StepLogEntry] populated as the agent runs. Each entry holds the FSM state, difficulty, resolved model id (when routed), monitors that fired, full intervention text, tool calls, tokens, and latency.

for entry in mw.step_log:
    print(
        f"step {entry.step}: state={entry.fsm_state} "
        f"difficulty={entry.difficulty} model_id={entry.model_id} "
        f"tokens={entry.tokens}"
    )
    if entry.monitors_fired:
        print(f"  monitors fired: {entry.monitors_fired}")
    for src, text in zip(entry.injection_sources, entry.intervention_texts):
        print(f"  [{src}] {text[:160]}")

Call entry.as_dict() for a JSON-shaped view.

Add CodebaseMemory tools

make_langchain_tools wraps a CodebaseMemory (and optionally an ImportGraph) into @tool-decorated callables. The agent uses them to recall prior findings before re-reading files and to persist new findings during a run.

from reasonblocks import CodebaseMemory, ImportGraph
from reasonblocks.integrations.langchain_tools import make_langchain_tools

memory = CodebaseMemory(
    codebase_id="my-org/my-repo",
    api_key=os.environ["REASONBLOCKS_API_KEY"],
)
graph = ImportGraph().build_from_files({path: open(path).read() for path in py_files})

rb_tools = make_langchain_tools(
    memory,
    graph,
    recall_top_k=5,
    recall_threshold=0.25,
    enable_recall=True,
    enable_store=True,
    enable_impact=True,
)

agent = create_agent(
    model="anthropic:claude-haiku-4-5-20251001",
    tools=[*rb_tools, search_codebase, read_file, edit_file, run_tests],
    system_prompt="Call recall_findings before reading any file.",
    middleware=[mw],
)

The factory adds up to three tools:

recall_findings — semantic search over prior findings.
store_finding — persist a new finding for future runs (hidden when enable_store=False).
impact_analysis — blast-radius lookup via ImportGraph (only added when a graph is passed).

ImportGraph.build_from_files requires networkx. Install with pip install networkx.

See Persist agent findings with CodebaseMemory for the underlying API.

Complete example

import os
from langchain.agents import create_agent
from langchain_core.tools import tool
from reasonblocks import CodebaseMemory, ReasonBlocks
from reasonblocks.integrations.langchain_tools import make_langchain_tools

rb = ReasonBlocks(api_key=os.environ["REASONBLOCKS_API_KEY"])
memory = CodebaseMemory(
    codebase_id="my-org/my-repo",
    api_key=os.environ["REASONBLOCKS_API_KEY"],
)
rb_tools = make_langchain_tools(memory)

@tool
def search_codebase(query: str) -> str:
    """Search the codebase for files matching a query."""
    ...

@tool
def read_file(path: str) -> str:
    """Read the contents of a file."""
    ...

mw = rb.middleware(
    agent_name="bugfixer",
    task="fix the TypeError in handler.py",
    codebase_id="my-org/my-repo",
)

agent = create_agent(
    model="anthropic:claude-haiku-4-5-20251001",
    tools=[*rb_tools, search_codebase, read_file],
    system_prompt="You are a senior software engineer. Fix bugs methodically.",
    middleware=[mw],
)

with mw:
    result = agent.invoke({"messages": [("user", "Fix the bug.")]})

Each mw is single-use — call rb.middleware(...) again for the next run.

Getting Started

Concepts

Using ReasonBlocks

Connectors and sync

Use ReasonBlocks with LangChain

Add CodebaseMemory tools

Complete example

​Add CodebaseMemory tools

​Complete example

Add CodebaseMemory tools

Complete example