Skip to Content

How do I trace a document processing pipeline built with LangChain?

Rius traces a LangChain document pipeline by hooking LangChain’s callbacks, so chains, tools, retrievers and model calls arrive as one nested trace. You add a few observe wrappers for the run root and each stage, a retrieval span for any search LangChain does not run, and a session and user scope for the run. Model calls carry tokens, and Rius computes cost from them.

How it works

Once init() runs, the LangChain integration records anything that runs through LangChain’s callback system: LCEL chains, agents and LangGraph nodes. Each step becomes a span typed by what it is. Chains are CHAIN spans, tool invocations are TOOL spans, retrievers are RETRIEVER spans, and model calls are LLM spans with model, token usage and messages. The nesting follows your chain structure. The full setup and blind spots are on the LangChain integration page, and the other integrations are listed under Integrations.

A document pipeline usually has code around the chains: loading a file, splitting it, searching an index, writing results somewhere. LangChain’s callbacks do not see that code, so you wrap it with the tools on the tracing page:

  • A run root. An observe wrapper around the function that processes one document gives every stage a common parent, so one document reads as one trace.
  • A step per stage. Wrap parsing, chunking, extraction and write-back with named observe calls. A default observe makes a CHAIN span named after the function.
  • Retrieval you run yourself. A search that does not go through a LangChain retriever gets a RETRIEVER span from start_as_current_span, with the data source and top_k. set_retrieved_documents() records ids and scores, not document text.
  • Your own tool functions. A function that LangChain does not already record as a tool, such as an upload to an archive, uses observe with kind=SpanKind.TOOL. A bare observe makes a CHAIN span, which the Tool kind filter in the Spans tab and the Tool Loop alert never see.

Text-completion models (BaseLLM subclasses, such as langchain_openai.OpenAI) are counted twice in tokens and cost. Chat models are counted once.

Set it up

  1. Install the LangChain instrumentation and call init() once at startup. The install page covers this in full; this is the short form:

    pip install "glassflow-rius[langchain]" langchain

  2. Wrap the stages and scope the session and user. The search function, chain, rerank_docs, job_id, requester_id and the other helpers below are your own code. Rius needs no changes to your chains.

    import rius
from rius import SpanKind
 
rius.init(api_key="ri_...", service_name="doc-pipeline")
 
@rius.observe(name="parse_document", capture_output=False)
def parse(path: str) -> str:
    ...
 
@rius.observe(name="extract_fields")
def extract(text: str) -> dict:
    return chain.invoke({"text": text})  # your LangChain chain
 
@rius.observe(name="rerank", kind=SpanKind.RERANKER, capture_output=False)
def rerank_docs(docs: list[str]) -> list[str]:
    ...
 
@rius.observe(kind=SpanKind.TOOL, tool_name="archive_upload")
def archive(result: dict) -> str:
    ...
 
@rius.observe(name="process_document")
def process(path: str) -> dict:
    text = parse(path)
    with rius.start_as_current_span(
        kind=SpanKind.RETRIEVER, data_source_id="policy-kb", top_k=5
    ) as obs:
        obs.set_input(text[:500])
        docs = search(text)
        obs.set_retrieved_documents([{"id": d.id, "score": d.score} for d in docs])
    result = extract(text)
    archive(result)
    return result
 
with rius.session(job_id), rius.user(requester_id):
    process("contracts/acme.pdf")

  3. Choose the session. Use your own id, such as a job or batch id, so every document in that unit of work shares a session.id. With no session scope, Rius groups each trace as its own session. For a script that handles one batch per process, RIUS_SESSION_ID sets one id for the whole process. In a server that handles many requests, scope each request instead.

  4. Choose the user. rius.user stamps user.id on every span in scope, including the spans the LangChain integration creates. Pass an opaque id rather than an email. Outside a user scope, spans are anonymous. There is no environment variable for the user.

  5. In TypeScript, install @arizeai/openinference-instrumentation-langchain, wrap each stage with observe around a named function, and use withSession and withUser the same way. A short script awaits rius.shutdown() before it exits.

  6. Keep large documents out of span content where you do not need them. capture_input=False and capture_output=False skip the arguments and return value of a stage, and captured values are truncated at 32 KB.

What you get

  • One trace per document, with the run root on top, a span per stage beneath it, and your LangChain chains, retrievers and model calls nested inside the stage that called them.
  • Tokens on each model call, and a cost per call that Rius computes. Never send a cost value yourself.
  • Retrieval spans that show which source was searched, how many documents were requested, and the ids and scores that came back.
  • Sessions that group all documents of one job, and a Users view by the user.id you set.
  • Failures that land on the stage that raised them, because exceptions are recorded on the span with ERROR status and re-raised.

To check, run one document through, open the trace in the console, and confirm that your stages appear as nested spans with the model call carrying model, tokens and cost.

When something else fits

If your pipeline is a single chain with no code around it, the integration alone is enough and you need no wrappers. If you call a model outside LangChain, the tracing page covers manual generations. Turn off the matching integration first, or the call is counted twice.

Still stuck? Get help

Last updated on
Quickstart