Pipeline Loop — Improvements & Roadmap

What the node-based pipeline system could benefit from, organized by priority.

Current State ✅

Architecture

┌─────────────┐
│   START     │
└──────┬──────┘
       │
┌──────▼──────┐
│  Retrieval  │◄──────────────┐
└──────┬──────┘               │
       │                      │
┌──────▼──────┐          (retry)
│ Verification│               │
└──────┬──────┘          ┌────┴────┐
       │                 │ Retry   │
  (pass)│(fail)          └─────────┘
┌──────▼──────┐
│  Generation │
└──────┬──────┘
       │
┌──────▼──────┐
│ Post-Gen    │
│ Check       │
└──────┬──────┘
       │
┌──────▼──────┐
│   END       │
└─────────────┘

Components

Note: Effort estimates below are rough approximations for a single developer. Actual time depends on testing, code review, and integration complexity.

🔴 High Priority

1. Query Expansion Node

Implementation:

• New node: QueryExpansionNode in pipeline_graph.py
• Uses LLM or synonym dictionary to generate 2–3 query variants
• Runs retrieval for each variant, merges results via RRF
• Insert before retrieval: query_expansion → retrieval → verification → ...

Estimated effort: 1–2 hours

Graph change:

query_expansion → retry_retrieval → verification → generation → post_check → success
                                                    ↓ (fail)
                                                  abstain

2. Streaming Pipeline Progress

Status: implemented. Nodes execute in real-time instead of waiting for the entire pipeline to complete.

Implementation (as shipped):

• Backend: Server-Sent Events (SSE) on /api/chat/stream
• Each node emits progress events: { node: "retrieval", status: "running" }, { node: "retrieval", status: "done", elapsed_ms: 120 }
• Frontend: a fetch-based SSE line parser (in services/api.ts) updates PipelineExecution in real-time
• Nodes pulse/highlight when active

🟡 Medium Priority

3. LangGraph StateGraph Integration

Why: Lets agents use the pipeline as LangGraph tools for agentic workflows.

Implementation:

• Wrap build_rag_pipeline() output as a LangGraph StateGraph
• Each PipelineNode maps to a LangGraph node
• Conditional edges map to LangGraph conditional routing
• Expose via get_legal_rag_tool() for agent use

Estimated effort: 2–3 hours

4. Pipeline Caching

Why: Cache retrieval results for repeated/similar queries — saves embedding + search time.

Implementation:

• Add CacheNode that checks a query cache before retrieval
• Cache key: SHA256 of normalized query text
• Cache storage: SQLite or in-memory LRU with TTL
• Cache invalidation: on ingestion or manual --clear-cache
• Hit/miss stats in the dashboard

Estimated effort: 2–3 hours

5. Custom Node Composition

Why: Let users define custom pipeline workflows via config (YAML/JSON).

Implementation:

• Define a pipeline schema:

yaml  pipeline:
    nodes:
      - name: expand
        type: query_expansion
      - name: retrieve
        type: retrieval
        retries: 2
      - name: verify
        type: verification
      - name: generate
        type: generation
    edges:
      expand: [retrieve]
      retrieve: [verify]
      verify:
        pass: [generate]
        fail: [abstain]

• PipelineBuilder.from_config(yaml_str) parses and builds the graph
• Store pipeline configs in data/pipelines/

Estimated effort: 3–4 hours

🟢 Nice to Have

6. Pipeline Debug Endpoint

Why: Full execution trace for debugging pipeline behavior.

Implementation:

• New endpoint: GET /api/pipeline/trace?query=...
• Returns full context dump: every node's input/output, timing, errors
• Frontend: "Debug" button in the pipeline section opens a modal with the trace

Estimated effort: 1–2 hours

7. Node Timeout Handling

Why: Kill stuck nodes (e.g., LLM hanging) after configurable timeout.

Implementation:

• Add timeout_ms to NodeConfig (already exists but unused)
• Pipeline.run() wraps node.run() in a threading.Timer or asyncio.wait_for
• On timeout: log error, mark node as FAILURE, continue pipeline

Estimated effort: 1–2 hours

8. A/B Testing Support

Why: Compare different pipeline configurations side-by-side.

Implementation:

• Run the same query through two pipeline configs
• Compare: timing, answer quality, source diversity
• Dashboard shows side-by-side comparison
• Store comparison results for later analysis

Estimated effort: 2–3 hours

9. Pipeline Metrics Export

Why: Production monitoring with Prometheus/Grafana.

Implementation:

• Expose metrics: pipeline_query_total, pipeline_node_duration_seconds, pipeline_node_errors_total
• Use prometheus_client library
• Add /metrics endpoint to FastAPI

Estimated effort: 2–3 hours

10. Pipeline Versioning

Why: Track pipeline configuration changes over time.

Implementation:

• Hash pipeline config (node types, edges, parameters)
• Store version alongside execution logs
• Dashboard shows which pipeline version produced each result
• Rollback support

Estimated effort: 2–3 hours

Effort Summary

Total estimated effort: ~20–28 hours

Recommended Implementation Order

1. Query expansion node — Highest impact on retrieval quality
2. Node timeout handling — Quick win, prevents hung pipelines
3. Debug endpoint — Makes development/debugging much easier
4. Pipeline caching — Performance improvement for repeated queries
5. Streaming progress — Best UX improvement
6. LangGraph integration — Enables agentic workflows
7. Custom node composition — Power user feature
8. Metrics export — Production readiness
9. A/B testing — Research/optimization tool
10. Pipeline versioning — Long-term maintainability