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

Rius traces a CrewAI document pipeline by capturing the model calls through its openai integration and adding observe spans for the run root, each stage and retrieval. Install glassflow-rius[openai] next to crewai, call rius.init(), and every model call arrives with model, tokens and cost. Scope each document with rius.session() and rius.user() so you can find its traces later. The CrewAI integration page has the full setup.

How it works

CrewAI 1.x drives the OpenAI client natively, so the openai integration captures each model call as a generation with model, tokens, cost and time to first token. If LiteLLM is installed, CrewAI uses it instead, and you use the litellm integration. Integrations lists every framework Rius covers.

Crews and tasks produce no spans of their own, and generation spans carry the generic name ChatCompletion. A document pipeline usually has stages outside the crew too, such as parsing, classification and lookup. You add those spans with observe and the span helpers:

• Run root: wrap the entrypoint that processes one document with observe(name=..., kind=SpanKind.AGENT). It becomes the root AGENT span.
• Retrieval: open a span with kind=SpanKind.RETRIEVER, a data_source_id for the index you search and a top_k. Call set_retrieved_documents() with each document’s id and score. These are identifiers and scores, not document text.
• A step per stage: a bare observe on a stage function makes a CHAIN span named after the function. Because crew and task structure is not visible in the trace, wrap per-task entrypoints with observe if you need to see which task a step belongs to.
• Tools: wrap each tool body with observe(kind=SpanKind.TOOL), with @tool outermost. A bare observe on a tool would make a CHAIN span, not a tool span.
• Session and user: rius.session() groups the traces of one job, and rius.user() ties them to the person who submitted it.

Set it up

1. Install the integration and initialize the SDK. The CrewAI integration page covers this in full.

   pip install "glassflow-rius[openai]" crewai

2. Add the spans your pipeline needs around the crew. This example parses a document, retrieves context, then runs a crew for the extraction stage. parse_document, search, parse_fields and the ids are your own code and values.

   import rius
   from crewai import Agent, Crew, Task
   from crewai.tools import tool
   from rius import SpanKind
    
   rius.init(api_key="ri_...", service_name="doc-pipeline")
    
   @tool("extract_fields")
   @rius.observe(kind=SpanKind.TOOL)
   def extract_fields(doc_id: str) -> str:
       """Pull the structured fields out of a parsed document."""
       return parse_fields(doc_id)
    
   analyst = Agent(
       role="Document analyst",
       goal="Extract accurate fields from documents.",
       backstory="You review contracts and invoices.",
       tools=[extract_fields],
       llm="gpt-5-mini",
   )
    
   @rius.observe
   def parse_stage(doc_id: str) -> str:
       return parse_document(doc_id)
    
   @rius.observe(name="extract-stage")
   def extract_stage(doc_id: str, context: str) -> str:
       task = Task(
           description=f"Extract the key fields from document {doc_id}. Context: {context}",
           expected_output="A short list of fields and values.",
           agent=analyst,
       )
       return str(Crew(agents=[analyst], tasks=[task]).kickoff())
    
   @rius.observe(name="run-document-pipeline", kind=SpanKind.AGENT)
   def process_document(doc_id: str, query: str) -> str:
       parse_stage(doc_id)
       with rius.start_as_current_span(
           kind=SpanKind.RETRIEVER, data_source_id="contracts-kb", top_k=5
       ) as obs:
           obs.set_input(query)
           docs = search(query)
           obs.set_retrieved_documents([{"id": d.id, "score": d.score} for d in docs])
       return extract_stage(doc_id, "\n".join(d.text for d in docs))
    
   with rius.user("user-1042"):
       with rius.session("job-8841"):
           process_document("doc-77", "payment terms")

3. Pass your own job id to rius.session() and an opaque user id to rius.user(). Both scopes stamp every span started inside them, including the spans the integration creates. Without a session, each trace is its own session.

4. Set CREWAI_TRACING_ENABLED=false where no terminal is attached, such as CI or containers. CrewAI otherwise asks about its own tracing on first run.

5. Run one document that calls a tool, then open the trace in the console. You should see the root AGENT span, your stage and retrieval spans, the LLM generations and the named TOOL span.

What you get

• A root span for each document, with the stages, retrieval and tool calls nested beneath it.
• Every model call with model, tokens and cost. Rius computes cost from the token counts as a list-price estimate in US dollars, so you never send a cost value.
• A retrieval span that records which documents came back and their scores, without storing document text.
• All traces of one job grouped in a session, and all of one person’s work on the Users page.
• Tool spans you can filter by the Tool kind, because they use the TOOL kind.
• Streamed calls carry token counts only when usage is enabled on the underlying client. See the OpenAI integration for the include_usage note.

When something else fits

• If your pipeline has no crew and calls a provider client directly, the OpenAI or Anthropic integration with observe wrappers is enough.
• If you run CrewAI through LiteLLM, use the litellm integration for the model calls.
• If a call has no integration, trace it as a manual generation, as described in Tracing your code.

Related