Skip to content

LangChain and LangGraph

The optional adapter, installed with pip install "openbtk[langchain]". See the guide for worked examples.

openbtk.integrations.langchain.embeddings.OpenBTKEmbeddings 

OpenBTKEmbeddings(provider: BaseEmbeddingProvider)

Bases: Embeddings

Expose a BaseEmbeddingProvider to LangChain.

Documents are embedded in slices of the provider's own batch_size so a long list is never sent as one oversized request. The provider's sends_data_offsite flag is unchanged by wrapping: an offsite provider stays offsite, and a local-only policy is enforced where the provider is constructed -- see :meth:from_registry.

Parameters:

Name Type Description Default
provider BaseEmbeddingProvider

The provider to wrap.

 required
Example

import numpy as np from openbtk.core.base import BaseEmbeddingProvider class Fixed(BaseEmbeddingProvider): ... sends_data_offsite = False ... dimension = 2 ... def embed(self, texts): ... return np.ones((len(texts), 2), dtype=np.float32) OpenBTKEmbeddings(Fixed()).embed_query("hello") [1.0, 1.0]

Source code in src/openbtk/integrations/langchain/embeddings.py 
40
41
def __init__(self, provider: BaseEmbeddingProvider) -> None:
    self._provider = provider

provider property 

provider: BaseEmbeddingProvider

The wrapped Openbtk provider.

from_registry classmethod 

from_registry(key: str, *, policy: PolicyConfig | None = None, **params: Any) -> OpenBTKEmbeddings

Build the provider registered under key and wrap it.

policy is forwarded to the registry, which refuses an offsite provider unless policy.allow_offsite_providers is true -- wrapping a provider for LangChain does not bypass the local-only guard.

Raises:

Type Description
PolicyError

If the provider sends data offsite and policy does not allow it.
Source code in src/openbtk/integrations/langchain/embeddings.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@classmethod
def from_registry(
    cls, key: str, *, policy: PolicyConfig | None = None, **params: Any
) -> OpenBTKEmbeddings:
    """Build the provider registered under ``key`` and wrap it.

    ``policy`` is forwarded to the registry, which refuses an offsite
    provider unless ``policy.allow_offsite_providers`` is true -- wrapping
    a provider for LangChain does not bypass the local-only guard.

    Raises:
        PolicyError: If the provider sends data offsite and ``policy``
            does not allow it.
    """
    return cls(EMBEDDING_REGISTRY.create(key, policy=policy, **params))

openbtk.integrations.langchain.chat_models.OpenBTKChatModel

Bases: BaseChatModel

Expose a BaseLLMProvider as a LangChain chat model.

Only system, user and assistant messages with plain-text content are supported -- anything else raises ProviderError rather than being flattened into a lossy prompt. Streaming is not overridden: Openbtk's BaseLLMProvider.stream takes a prompt string, not a message list, so .stream() here falls back to LangChain's default of yielding the complete response as a single chunk.

Construct it with provider= -- the BaseLLMProvider to wrap (a pydantic field, as every LangChain chat model's parameters are).

Example

from openbtk.core.base import BaseLLMProvider from openbtk.core.schemas import LLMResponse class Echo(BaseLLMProvider): ... sends_data_offsite = False ... def generate(self, prompt, kw): return LLMResponse(text=prompt) ... def stream(self, prompt, kw): yield prompt ... def chat(self, messages, **kw): ... return LLMResponse(text=messages[-1].content.upper()) OpenBTKChatModel(provider=Echo()).invoke("hi").content 'HI'

openbtk.integrations.langchain.vectorstores.OpenBTKVectorStore 

OpenBTKVectorStore(store: BaseVectorStore, embedding: Embeddings, *, text_key: str = _DEFAULT_TEXT_KEY)

Bases: VectorStore

Expose a BaseVectorStore and an Embeddings as a LangChain VectorStore.

Openbtk stores hold vectors and metadata, not documents, so each text is kept in the metadata under text_key ("text" -- the same key :class:~openbtk.pipelines.rag.RAGPipeline reads, so one index serves both APIs). A caller's own metadata may not use that key.

Scores are the underlying store's own: their scale and direction (L2 distance falls as similarity rises; inner product rises) depend on how the store was built, which this adapter cannot know. For that reason similarity_search_with_relevance_scores (a normalised 0-1 score) is unsupported rather than guessed at.

Parameters:

Name Type Description Default
store BaseVectorStore

The Openbtk vector store.

 required
embedding Embeddings

Embeds documents and queries. Wrap an Openbtk provider with :class:~openbtk.integrations.langchain.OpenBTKEmbeddings.

 required
text_key str

Metadata key holding each document's text.

 _DEFAULT_TEXT_KEY
Example

import numpy as np from openbtk.core.base import BaseVectorStore from openbtk.core.schemas import SearchResult from langchain_core.embeddings import DeterministicFakeEmbedding class Mem(BaseVectorStore): ... def init(self): self.rows = {} ... def upsert(self, ids, vectors, metadata): ... self.rows.update(zip(ids, metadata)) ... def query(self, vector, top_k=5, filter=None): ... return [SearchResult(id=i, score=1.0, metadata=m) ... for i, m in list(self.rows.items())[:top_k]] ... def delete(self, ids): pass vs = OpenBTKVectorStore(Mem(), DeterministicFakeEmbedding(size=4)) _ = vs.add_texts(["note A"], ids=["a"]) vs.similarity_search("anything")[0].page_content 'note A'

Source code in src/openbtk/integrations/langchain/vectorstores.py 
65
66
67
68
69
70
71
72
73
74
def __init__(
    self,
    store: BaseVectorStore,
    embedding: Embeddings,
    *,
    text_key: str = _DEFAULT_TEXT_KEY,
) -> None:
    self._store = store
    self._embedding = embedding
    self._text_key = text_key

store property 

store: BaseVectorStore

The wrapped Openbtk store.

delete 

delete(ids: list[str] | None = None, **kwargs: Any) -> bool | None

Delete by id. Deleting everything (ids=None) is unsupported: BaseVectorStore has no clear operation.

Source code in src/openbtk/integrations/langchain/vectorstores.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def delete(
    self,
    ids: list[str] | None = None,
    **kwargs: Any,  # noqa: ARG002 -- LangChain's signature
) -> bool | None:
    """Delete by id. Deleting everything (``ids=None``) is unsupported:
    ``BaseVectorStore`` has no clear operation."""
    if ids is None:
        raise NotImplementedError(
            "OpenBTKVectorStore.delete() needs explicit ids; "
            "BaseVectorStore has no delete-all."
        )
    self._store.delete(ids)
    return True

from_texts classmethod 

from_texts(texts: list[str], embedding: Embeddings, metadatas: list[dict[str, Any]] | None = None, *, ids: list[str] | None = None, store: BaseVectorStore | None = None, **kwargs: Any) -> OpenBTKVectorStore

Build over store and index texts.

store is required: this adapter does not choose a backend (FAISS, Chroma, Qdrant...) on the caller's behalf.

Source code in src/openbtk/integrations/langchain/vectorstores.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
@classmethod
def from_texts(
    cls,
    texts: list[str],
    embedding: Embeddings,
    metadatas: list[dict[str, Any]] | None = None,
    *,
    ids: list[str] | None = None,
    store: BaseVectorStore | None = None,
    **kwargs: Any,  # noqa: ARG003 -- LangChain's signature
) -> OpenBTKVectorStore:
    """Build over ``store`` and index ``texts``.

    ``store`` is required: this adapter does not choose a backend (FAISS,
    Chroma, Qdrant...) on the caller's behalf.
    """
    if store is None:
        raise ConfigError(
            "OpenBTKVectorStore.from_texts() needs store=<a BaseVectorStore>; "
            "it does not pick a backend for you."
        )
    instance = cls(store, embedding)
    instance.add_texts(texts, metadatas, ids=ids)
    return instance

openbtk.integrations.langchain.documents.chunk_to_document 

chunk_to_document(chunk: ClinicalTextChunk) -> Document

Convert a chunk to a LangChain Document.

Raises:

Type Description
ProcessingError

If chunk.metadata uses a key this adapter reserves (chunk_id, record_id, section, token_count, span_start, span_end, entities).
Example

from openbtk.core.schemas import TextSpan chunk = ClinicalTextChunk( ... chunk_id="n1:0", record_id="n1", text="Plan: rest.", ... span=TextSpan(start=0, end=11, label="chunk", confidence=1.0), ... token_count=4, ... ) doc = chunk_to_document(chunk) doc.page_content, doc.id, doc.metadata["record_id"] ('Plan: rest.', 'n1:0', 'n1')

Source code in src/openbtk/integrations/langchain/documents.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def chunk_to_document(chunk: ClinicalTextChunk) -> Document:
    """Convert a chunk to a LangChain ``Document``.

    Raises:
        ProcessingError: If ``chunk.metadata`` uses a key this adapter
            reserves (``chunk_id``, ``record_id``, ``section``,
            ``token_count``, ``span_start``, ``span_end``, ``entities``).

    Example:
        >>> from openbtk.core.schemas import TextSpan
        >>> chunk = ClinicalTextChunk(
        ...     chunk_id="n1:0", record_id="n1", text="Plan: rest.",
        ...     span=TextSpan(start=0, end=11, label="chunk", confidence=1.0),
        ...     token_count=4,
        ... )
        >>> doc = chunk_to_document(chunk)
        >>> doc.page_content, doc.id, doc.metadata["record_id"]
        ('Plan: rest.', 'n1:0', 'n1')
    """
    clash = sorted(_RESERVED & chunk.metadata.keys())
    if clash:
        raise ProcessingError(
            "ClinicalTextChunk.metadata uses keys reserved by the LangChain "
            f"adapter: {clash}. Rename them before converting.",
            context={"chunk_id": chunk.chunk_id, "reserved_keys": clash},
        )
    metadata: dict[str, Any] = dict(chunk.metadata)
    metadata[_CHUNK_ID] = chunk.chunk_id
    metadata[_RECORD_ID] = chunk.record_id
    metadata[_TOKEN_COUNT] = chunk.token_count
    metadata[_SPAN_START] = chunk.span.start
    metadata[_SPAN_END] = chunk.span.end
    if chunk.section is not None:
        metadata[_SECTION] = chunk.section
    if chunk.entities:
        metadata[_ENTITIES] = [e.model_dump(mode="json") for e in chunk.entities]
    return Document(id=chunk.chunk_id, page_content=chunk.text, metadata=metadata)

openbtk.integrations.langchain.documents.document_to_chunk 

document_to_chunk(document: Document, *, record_id: str | None = None, token_counter: Callable[[str], int] | None = None) -> ClinicalTextChunk

Convert a Document back to a chunk.

A Document produced by :func:chunk_to_document round-trips exactly. For any other Document, record_id and token_counter fill in what the Document cannot say; the chunk id falls back to Document.id, and the span to the whole text.

Parameters:

Name Type Description Default
document Document

The Document to convert.

 required
record_id str | None

Overrides/fills metadata["record_id"].

 None
token_counter Callable[[str], int] | None

Returns the exact subword count of a text. Required unless the Document carries metadata["token_count"].

 None

Raises:

Type Description
ProcessingError

If the chunk id, record id or token count cannot be determined, or the metadata is not valid for a chunk.
Source code in src/openbtk/integrations/langchain/documents.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def document_to_chunk(
    document: Document,
    *,
    record_id: str | None = None,
    token_counter: Callable[[str], int] | None = None,
) -> ClinicalTextChunk:
    """Convert a ``Document`` back to a chunk.

    A Document produced by :func:`chunk_to_document` round-trips exactly.
    For any other Document, ``record_id`` and ``token_counter`` fill in what
    the Document cannot say; the chunk id falls back to ``Document.id``, and
    the span to the whole text.

    Args:
        document: The Document to convert.
        record_id: Overrides/fills ``metadata["record_id"]``.
        token_counter: Returns the exact subword count of a text. Required
            unless the Document carries ``metadata["token_count"]``.

    Raises:
        ProcessingError: If the chunk id, record id or token count cannot be
            determined, or the metadata is not valid for a chunk.
    """
    meta = dict(document.metadata)
    ctx: dict[str, Any] = {"document_id": document.id}

    chunk_id = meta.pop(_CHUNK_ID, None) or document.id
    resolved_record = record_id if record_id is not None else meta.pop(_RECORD_ID, None)
    meta.pop(_RECORD_ID, None)
    if not chunk_id:
        raise ProcessingError(
            "Cannot build a chunk: the Document has neither an id nor "
            "metadata['chunk_id'].",
            context=ctx,
        )
    if not resolved_record:
        raise ProcessingError(
            "Cannot build a chunk: pass record_id= or set metadata['record_id'].",
            context=ctx,
        )
    token_count = meta.pop(_TOKEN_COUNT, None)
    if token_count is None:
        if token_counter is None:
            raise ProcessingError(
                "Cannot build a chunk: the Document has no metadata['token_count'] "
                "and no token_counter was given. ClinicalTextChunk needs an exact "
                "subword count; it is never estimated.",
                context=ctx,
            )
        token_count = token_counter(document.page_content)

    text = document.page_content
    start = meta.pop(_SPAN_START, 0)
    end = meta.pop(_SPAN_END, len(text))
    section = meta.pop(_SECTION, None)
    entities = [LinkedEntity.model_validate(e) for e in meta.pop(_ENTITIES, [])]
    try:
        return ClinicalTextChunk(
            chunk_id=str(chunk_id),
            record_id=str(resolved_record),
            text=text,
            span=TextSpan(start=start, end=end, label="chunk", confidence=1.0),
            section=section,
            token_count=token_count,
            entities=entities,
            metadata=meta,
        )
    except ValueError as e:
        raise ProcessingError(
            f"Document is not a valid chunk: {type(e).__name__}.", context=ctx
        ) from e

openbtk.integrations.langchain.runnables.as_runnable 

as_runnable(component: Component | DeidEngine) -> Runnable[Any, Any]

Wrap component as a LangChain Runnable (see the module table).

Raises:

Type Description
ConfigError

If component is not one of the supported kinds.
Example

import contextlib, io with contextlib.redirect_stdout(io.StringIO()): # registry debug logs ... from openbtk.guardrails.phi_leakage import PHILeakageGuardrail ... result = as_runnable(PHILeakageGuardrail()).invoke("Plan: rest.") result.passed True

Source code in src/openbtk/integrations/langchain/runnables.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def as_runnable(component: Component | DeidEngine) -> Runnable[Any, Any]:
    """Wrap ``component`` as a LangChain ``Runnable`` (see the module table).

    Raises:
        ConfigError: If ``component`` is not one of the supported kinds.

    Example:
        >>> import contextlib, io
        >>> with contextlib.redirect_stdout(io.StringIO()):  # registry debug logs
        ...     from openbtk.guardrails.phi_leakage import PHILeakageGuardrail
        ...     result = as_runnable(PHILeakageGuardrail()).invoke("Plan: rest.")
        >>> result.passed
        True
    """
    name = type(component).__name__
    fn: Any
    if isinstance(component, BaseLLMProvider):
        llm = component

        def fn(prompt: str) -> str:
            return llm.generate(prompt).text

    elif isinstance(component, BaseEmbeddingProvider):
        embeddings = OpenBTKEmbeddings(component)
        fn = embeddings.embed_documents
    elif isinstance(component, BasePreprocessor):
        fn = component.process
    elif isinstance(component, BaseChunker):
        chunker = component

        def fn(record: Any) -> list[Any]:
            return list(chunker.chunk(record))

    elif isinstance(component, BaseGuardrail):
        fn = component.check
    elif isinstance(component, DeidEngine):
        engine = component

        def fn(payload: dict[str, str]) -> Any:
            try:
                text, patient_id = payload["text"], payload["patient_id"]
            except (KeyError, TypeError) as e:
                raise ConfigError(
                    'A DeidEngine runnable takes {"text": ..., "patient_id": ...}.'
                ) from e
            return engine.deidentify(text, patient_id=patient_id)

    else:
        raise ConfigError(
            f"as_runnable() does not support {name}. Supported: {_SUPPORTED}.",
            context={"component": name},
        )
    return RunnableLambda(fn, name=name)

openbtk.integrations.langchain.runnables.from_runnable 

from_runnable(runnable: Runnable[str, str]) -> RunnablePreprocessor

Wrap a text-to-text Runnable as an Openbtk record preprocessor.

Source code in src/openbtk/integrations/langchain/runnables.py 
135
136
137
def from_runnable(runnable: Runnable[str, str]) -> RunnablePreprocessor:
    """Wrap a text-to-text ``Runnable`` as an Openbtk record preprocessor."""
    return RunnablePreprocessor(runnable)

openbtk.integrations.langchain.graph.as_langgraph_node 

as_langgraph_node(component: Component | DeidEngine, *, input_key: str, output_key: str, name: str | None = None) -> Callable[[Mapping[str, Any]], dict[str, Any]]

Wrap component as a LangGraph node.

The node reads state[input_key], runs the component exactly as :func:~openbtk.integrations.langchain.as_runnable documents, and returns {output_key: result} -- a partial update LangGraph merges into state.

Parameters:

Name Type Description Default
component Component | DeidEngine

Any component as_runnable supports.

 required
input_key str

State key holding the component's input.

 required
output_key str

State key the result is written to.

 required
name str | None

The node's __name__ (LangGraph's default node name). Defaults to the component's class name.

 None

Raises:

Type Description
ConfigError

At call time, if input_key is missing from the state (the message names the key, never the state's contents).
Example

import contextlib, io with contextlib.redirect_stdout(io.StringIO()): # registry debug logs ... from openbtk.guardrails.phi_leakage import PHILeakageGuardrail ... node = as_langgraph_node( ... PHILeakageGuardrail(), input_key="draft", output_key="check" ... ) ... update = node({"draft": "Plan: rest."}) update["check"].passed True

Source code in src/openbtk/integrations/langchain/graph.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def as_langgraph_node(
    component: Component | DeidEngine,
    *,
    input_key: str,
    output_key: str,
    name: str | None = None,
) -> Callable[[Mapping[str, Any]], dict[str, Any]]:
    """Wrap ``component`` as a LangGraph node.

    The node reads ``state[input_key]``, runs the component exactly as
    :func:`~openbtk.integrations.langchain.as_runnable` documents, and returns
    ``{output_key: result}`` -- a partial update LangGraph merges into state.

    Args:
        component: Any component ``as_runnable`` supports.
        input_key: State key holding the component's input.
        output_key: State key the result is written to.
        name: The node's ``__name__`` (LangGraph's default node name).
            Defaults to the component's class name.

    Raises:
        ConfigError: At call time, if ``input_key`` is missing from the state
            (the message names the key, never the state's contents).

    Example:
        >>> import contextlib, io
        >>> with contextlib.redirect_stdout(io.StringIO()):  # registry debug logs
        ...     from openbtk.guardrails.phi_leakage import PHILeakageGuardrail
        ...     node = as_langgraph_node(
        ...         PHILeakageGuardrail(), input_key="draft", output_key="check"
        ...     )
        ...     update = node({"draft": "Plan: rest."})
        >>> update["check"].passed
        True
    """
    runnable = as_runnable(component)

    def node(state: Mapping[str, Any]) -> dict[str, Any]:
        if input_key not in state:
            raise ConfigError(
                f"Graph state has no {input_key!r} key for node {node.__name__!r}.",
                context={"input_key": input_key},
            )
        return {output_key: runnable.invoke(state[input_key])}

    node.__name__ = name or type(component).__name__
    return node