LangChain + LangGraph Integration

Pair contextweaver with LangChain and LangGraph so chain-based and graph-based agents get budget-aware memory, bounded tool routing, and a context firewall — without giving up the framework's agent loop.

Importable adapter

As of issue #502, contextweaver.adapters.langchain ships a first-class BaseTool → SelectableItem adapter, so you no longer hand-roll the conversion:

from contextweaver.adapters.langchain import load_langchain_catalog

# `tools` is your list of LangChain BaseTool instances.
catalog = load_langchain_catalog(tools, namespace="research")

Install the optional extra for live loading (pip install 'contextweaver[langchain]'); the plain-dict path (langchain_tools_to_catalog) needs no extra and is handy in tests. Feed the returned Catalog to the Router exactly as in the routing examples below.

Why

LangChain's memory classes (ConversationBufferMemory, ConversationSummaryMemory) and LangGraph's stateful checkpoints both accumulate conversation history with no token-budget enforcement. Long sessions blow through the model's context window; large tool outputs (multi-KB JSON, RAG retrievals) bloat the prompt; loading every tool into the system message wastes thousands of tokens.

contextweaver provides three composable replacements:

Pain	contextweaver answer
Unbounded `ConversationBufferMemory`	`ContextManager.build_sync(phase=…)` returns a phase-specific, budgeted prompt
Loading every tool into the LLM	`Router.route(query)` returns a bounded shortlist
Large tool results bloat the prompt	`ContextManager.ingest_tool_result_sync()` firewalls raw bytes to the artifact store

The repo already ships a runnable LangChain memory-replacement demo: examples/langchain_memory_demo.py (installable via pip install 'contextweaver[langchain]'). Read that first — it shows the token-budget delta side by side.

Prerequisites

pip install 'contextweaver[langchain]' langchain langchain-openai langgraph
export OPENAI_API_KEY=sk-...

LangChain Integration

Architecture

User query
   │
   ▼
contextweaver Router         ← all tools (Catalog)
   │ (top-k shortlist)
   ▼
LangChain Agent              ← receives only the shortlist
   │ (tool call)
   ▼
contextweaver Firewall       ← intercepts large results
   │ (summary)
   ▼
contextweaver ContextManager ← phase-specific compilation
   │ (pack.prompt)
   ▼
LLM

Memory replacement

ConversationBufferMemory is the canonical "memory" you want to replace:

# Before — unbounded memory
from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory()

# After — budget-aware memory via contextweaver
from contextweaver.context.manager import ContextManager
from contextweaver.config import ContextBudget
from contextweaver.types import ContextItem, ItemKind, Phase

ctx_mgr = ContextManager(
    budget=ContextBudget(route=500, call=1200, interpret=1500, answer=3000),
)

To get the conversation history that a LangChain agent expects, build a prompt for the relevant phase right before invoking the agent:

def respond(user_query: str, turn: int) -> str:
    # 1. Ingest the user turn into the event log.
    ctx_mgr.ingest_sync(ContextItem(
        id=f"u{turn}", kind=ItemKind.user_turn, text=user_query,
    ))

    # 2. Compile a phase-specific, budgeted prompt.
    pack = ctx_mgr.build_sync(phase=Phase.answer, query=user_query)

    # 3. Invoke the LangChain agent with the compiled prompt as "history".
    result = agent_executor.invoke({"input": user_query, "history": pack.prompt})

    # 4. Ingest the agent's response so the next turn can see it.
    ctx_mgr.ingest_sync(ContextItem(
        id=f"a{turn}", kind=ItemKind.agent_msg, text=result["output"],
    ))
    return result["output"]

pack.stats.included_count / dropped_count tell you exactly what was kept and what was dropped by the budget — surface these in logs.

Firewalling tool results via a callback

LangChain emits a BaseCallbackHandler.on_tool_end() event after every tool finishes. That's the natural hook for the context firewall:

from langchain_core.callbacks import BaseCallbackHandler

class ContextWeaverCallback(BaseCallbackHandler):
    """Route LangChain tool results through contextweaver's firewall."""

    def __init__(self, ctx_mgr: ContextManager) -> None:
        self._mgr = ctx_mgr
        self._call_counter = 0

    def on_tool_end(self, output: str, *, name: str, **_: object) -> None:
        self._call_counter += 1
        tool_call_id = f"tc-{self._call_counter}"
        self._mgr.ingest_tool_result_sync(
            tool_call_id=tool_call_id,
            raw_output=str(output),
            tool_name=name,
        )

Wire it into your AgentExecutor:

from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4")
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[ContextWeaverCallback(ctx_mgr)],
)

Routing to a tool shortlist

Use Router.route() to narrow the tool list before building the agent:

from contextweaver.routing.catalog import Catalog
from contextweaver.routing.router import Router
from contextweaver.routing.tree import TreeBuilder

# (build Catalog of SelectableItems from your tool definitions)
graph = TreeBuilder().build(catalog.all())
router = Router(graph, items=catalog.all(), top_k=3)

shortlist_ids = router.route(user_query).candidate_ids
selected_tools = [t for t in all_tools if t.name in shortlist_ids]
agent = create_openai_functions_agent(llm, selected_tools, prompt)

LangGraph Integration

LangGraph state is the natural place to park a ContextManager because each node sees the same state object.

Architecture

User query
   │
   ▼
[LangGraph State]                ← ctx_mgr lives here
   │
   ▼
node: route                       ← ctx_mgr.build_sync(phase=Phase.route)
   │ (Router.route → shortlist)
   ▼
node: call_tool                   ← framework calls the tool
   │ (ctx_mgr.ingest_tool_result_sync — firewall)
   ▼
node: interpret                   ← ctx_mgr.build_sync(phase=Phase.interpret)
   │
   ▼
node: answer                      ← ctx_mgr.build_sync(phase=Phase.answer)
   │
   ▼
LLM

Stateful 4-node graph

from typing import TypedDict

from langchain_openai import ChatOpenAI
from langgraph.graph import END, StateGraph

from contextweaver.context.manager import ContextManager
from contextweaver.routing.router import Router
from contextweaver.types import ContextItem, ItemKind, Phase


class AgentState(TypedDict):
    user_query: str
    routed_tools: list[str]
    tool_result: str
    answer: str
    turn: int


ctx_mgr = ContextManager()
# `router` and `llm` from the earlier sections


def route_node(state: AgentState) -> AgentState:
    ctx_mgr.ingest_sync(ContextItem(
        id=f"u{state['turn']}", kind=ItemKind.user_turn, text=state["user_query"],
    ))
    routed = router.route(state["user_query"])
    return {**state, "routed_tools": list(routed.candidate_ids)}


def call_tool_node(state: AgentState) -> AgentState:
    tool_id = state["routed_tools"][0]
    raw = execute_tool(tool_id, state["user_query"])   # your runtime
    ctx_mgr.ingest_sync(ContextItem(
        id=f"tc-{state['turn']}", kind=ItemKind.tool_call,
        text=f"{tool_id}(...)", parent_id=f"u{state['turn']}",
    ))
    item, _ = ctx_mgr.ingest_tool_result_sync(
        tool_call_id=f"tc-{state['turn']}",
        raw_output=str(raw),
        tool_name=tool_id,
    )
    return {**state, "tool_result": item.text}


def answer_node(state: AgentState) -> AgentState:
    pack = ctx_mgr.build_sync(phase=Phase.answer, query=state["user_query"])
    answer = ChatOpenAI(model="gpt-4").invoke(pack.prompt)
    return {**state, "answer": str(answer.content)}


graph = StateGraph(AgentState)
graph.add_node("route",     route_node)
graph.add_node("call_tool", call_tool_node)
graph.add_node("answer",    answer_node)
graph.add_edge("route", "call_tool")
graph.add_edge("call_tool", "answer")
graph.add_edge("answer", END)
graph.set_entry_point("route")
app = graph.compile()

result = app.invoke({"user_query": "...", "routed_tools": [], "tool_result": "",
                     "answer": "", "turn": 1})

Putting ctx_mgr in the closure (rather than the state itself) keeps the state pickle-friendly for LangGraph checkpoints.

Migration guide — LangChain memory → contextweaver

LangChain pattern	contextweaver equivalent
`ConversationBufferMemory()`	`ContextManager(budget=ContextBudget(...))`
`ConversationSummaryMemory(llm=...)`	`ContextManager(summarizer=YourSummarizer())` — implement the `Summarizer` protocol
`memory.load_memory_variables({})`	`ctx_mgr.build_sync(phase=Phase.answer, query=user_query).prompt`
`memory.save_context(in, out)`	`ctx_mgr.ingest_sync(ContextItem(kind=ItemKind.user_turn, ...))` + `ctx_mgr.ingest_sync(ContextItem(kind=ItemKind.agent_msg, ...))`

See examples/langchain_memory_demo.py for a runnable side-by-side comparison.

Troubleshooting

AgentExecutor ignores pack.prompt. Make sure the prompt template has a "{history}" placeholder (or whichever variable you injected the pack into) — LangChain does not auto-discover it.
Callbacks not firing. Confirm you passed callbacks=[...] to the AgentExecutor, not just to the LLM. on_tool_end is on the executor.
LangGraph checkpoints fail to pickle. Keep ContextManager outside the TypedDict state — it holds an in-memory event log that isn't serialised by default. Persist mgr.event_log.to_dict() separately alongside the checkpoint if you need cross-session continuity.
Tool not in the shortlist. Inspect result.scores — the TF-IDF retriever may need richer descriptions or tags. You can also use context_hints=[...] to inject conversation context into scoring.

LangChain + LangGraph Integration

Importable adapter

Why

Prerequisites

LangChain Integration

Architecture

Memory replacement

Firewalling tool results via a callback

Routing to a tool shortlist

LangGraph Integration

Architecture

Stateful 4-node graph

Migration guide — LangChain memory → contextweaver

Troubleshooting

See also