Crate llm_pipeline 

LLM Pipeline

Reusable node payloads for LLM workflows, with optional sequential chaining.

This crate provides the building blocks for LLM-powered workflows: payloads that execute LLM calls, parsing utilities for messy model output, and a chain helper for sequential composition.

Orchestration (routing, loops, concurrency, checkpoints) belongs in your graph runtime (e.g. LangGraph). This crate provides what runs inside each node.

Core Concepts

• Payload — object-safe trait for executable units. Takes a serde_json::Value input, returns a PayloadOutput.
• ExecCtx — shared execution context (HTTP client, endpoint, template vars, cancellation, optional event handler).
• LlmCall — the primary payload: renders prompts, calls Ollama, parses responses.
• Chain — sequential composition of payloads.
• PayloadOutput — Value-based output with parse_as::<T>() for typed extraction at workflow edges.

Quick Start (Payload API)

use llm_pipeline::{LlmCall, Chain, ExecCtx};
use llm_pipeline::payload::Payload;
use serde::Deserialize;
use serde_json::json;

#[derive(Debug, Deserialize)]
struct Analysis { summary: String }

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let ctx = ExecCtx::builder("http://localhost:11434").build();

    let chain = Chain::new("analyze")
        .push(Box::new(
            LlmCall::new("draft", "Analyze: {input}")
                .with_config(llm_pipeline::LlmConfig::default().with_json_mode(true))
        ))
        .push(Box::new(
            LlmCall::new("refine", "Refine this analysis: {input}")
                .with_config(llm_pipeline::LlmConfig::default().with_json_mode(true))
        ));

    let output = chain.execute(&ctx, json!("Your text here")).await?;
    let result: Analysis = output.parse_as()?;
    println!("{}", result.summary);
    Ok(())
}

Pipeline API (compatibility)

The original Pipeline<T> API is still available and works as before. Internally it now uses LlmCall payloads.

use llm_pipeline::{Pipeline, Stage, PipelineInput};
use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, Serialize, Deserialize)]
struct Analysis { summary: String, insights: Vec<String> }

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = reqwest::Client::new();
    let pipeline = Pipeline::<Analysis>::builder()
        .add_stage(Stage::new("analyze", "Analyze: {input}").with_json_mode(true))
        .add_stage(Stage::new("refine", "Refine: {input}").with_json_mode(true))
        .build()?;

    let result = pipeline.execute(
        &client, "http://localhost:11434", PipelineInput::new("Your text"),
    ).await?;
    println!("{}", result.final_output.summary);
    Ok(())
}

Re-exports

pub use backend::BackoffConfig;

pub use backend::MockBackend;

pub use backend::OllamaBackend;

pub use backend::RecordingBackend;

pub use chain::Chain;

pub use diagnostics::ParseDiagnostics;

pub use exec_ctx::ExecCtx;

pub use exec_ctx::ExecCtxBuilder;

pub use limits::PipelineLimits;

pub use llm_call::LlmCall;

pub use output_strategy::OutputStrategy;

pub use payload::BoxFut;

pub use payload::Payload;

pub use payload::PayloadOutput;

pub use retry::RetryConfig;

pub use retry_policy::SemanticRetryPolicy;

pub use retry_policy::TransportRetryPolicy;

pub use streaming::StreamingDecoder;

pub use tool_loop::ToolInvocation;

pub use tool_loop::ToolLoopChoice;

pub use tool_loop::ToolLoopRequest;

pub use tool_loop::ToolLoopResponse;

pub use tool_loop::ToolLoopRunner;

pub use trace::TraceId;Deprecated

pub use client::LlmConfig;

pub use error::PipelineError;

pub use error::Result;

pub use pipeline::Pipeline;

pub use pipeline::PipelineBuilder;

pub use stage::Stage;

pub use stage::StageBuilder;

pub use types::BudgetDebitV1;

pub use types::ExecutionOutcome;

pub use types::PipelineContext;

pub use types::PipelineExecutionReceiptV1;

pub use types::PipelineInput;

pub use types::PipelineProgress;

pub use types::PipelineResult;

pub use types::ProviderCallReceiptV1;

pub use types::RetrievedContextProvenanceV1;

pub use types::RetryCause;

pub use types::RetryDecision;

pub use types::RetryDecisionReceiptV1;

pub use types::StageOutput;

Modules

backend

Backend trait and normalized request/response types.

chain

Sequential chain of payloads.

client

diagnostics

Parse diagnostics and telemetry for output parsing.

error

events

Event system for payload lifecycle and streaming hooks.

exec_ctx

Execution context shared across payload invocations.

limits

Resource limits for pipeline operations.

llm_call

LLM call payload — the primary execution unit.

output_parser

Re-exports from the standalone [llm-output-parser] crate.

output_strategy

Output strategy for configuring how raw LLM text is parsed.

parsing

Parsing utilities for LLM responses.

payload

Core payload trait and output types.

pipeline

prompt

retry

Semantic retry with error feedback for LLM output parsing.

retry_policy

Named retry policy types for transport and semantic retries.

stage

streaming

Buffered streaming decoder for newline-delimited JSON streams.

tool_loop

trace

Cross-crate trace identifier for correlating operations.

types

Structs

TraceCtx

In-process trace context for cross-crate correlation.