[Agentic Phase 4] Pipeline orchestration with pydantic-graph

[bashandbone]

Phase 4: pydantic-graph Pipeline Orchestration

Parent Epic: #123
Depends On: #126 (Context Agent Tooling)
Target: v0.3
Risk Level: Medium

Integrate pydantic-graph for structured pipeline orchestration, enabling complex workflow composition with clear execution paths and strategy-based routing.

Goals

• Graph-based pipeline definitions
• Strategy-based execution paths
• Complex workflow composition
• Visual pipeline representation
• Maintainable orchestration logic

Background

pydantic-graph provides:

• Declarative workflow definitions
• Type-safe node connections
• Conditional execution paths
• State management
• Visualization capabilities

Implementation Checklist

pydantic-graph Integration

• Add pydantic-graph dependency
• Create graph definition framework
• Implement node types for CodeWeaver operations
  • Search nodes
  • Agent reasoning nodes
  • Data provider nodes
  • Analysis nodes
  • Synthesis nodes
• Build graph execution engine

Pipeline Definitions

• Simple search pipeline
  • Query → Search → Results
• Agent-enhanced pipeline
  • Query → Agent Analysis → Search → Agent Synthesis
• Multi-source pipeline
  • Query → Parallel (Code Search + Data Providers) → Merge → Synthesis
• Context agent pipeline
  • Query → Decomposition → Multi-step Search → Aggregation → Synthesis

Strategy System

• Strategy selection logic
  • Query complexity assessment
  • Cost/latency optimization
  • Quality requirements
• Conditional routing
  • Simple queries → Fast path
  • Complex queries → Agent path
  • External context needed → Data provider path
• Fallback strategies
  • Agent unavailable → Direct search
  • Timeout → Partial results
  • Error → Graceful degradation

Pipeline Composition

• Reusable pipeline components
• Pipeline templates
• Dynamic pipeline construction
• Pipeline validation

Visualization & Debugging

• Pipeline visualization tools
• Execution trace visualization
• Performance profiling per node
• Debug mode with step-through

Testing

• Unit tests for graph nodes
• Integration tests for pipelines
• Strategy selection tests
• Performance benchmarks
• Edge case handling

Configuration

• Pipeline selection rules
• Node timeout configuration
• Resource limits per node
• Caching strategies
• Monitoring/observability hooks

Success Criteria

• Pipelines are declarative and maintainable
• Strategy selection optimizes cost/latency/quality
• Complex workflows compose cleanly
• Visualization aids debugging
• Performance acceptable
• Tests passing
• Documentation complete

Example Pipelines

Simple Search Pipeline

Query → PreprocessNode → SearchNode → ResultsNode

Agent-Enhanced Pipeline

Query → AgentQueryAnalysis → StrategySelection
         ↓                           ↓
    SemanticSearch              LexicalSearch
         ↓                           ↓
         → ResultsMerge → AgentSynthesis → Response

Context Agent Pipeline

Query → AgentDecomposition → ParallelSearch[]
                                    ↓
                              Aggregation
                                    ↓
                            IterativeRefinement → Synthesis

Benefits

• Clarity: Visual pipeline representation
• Maintainability: Declarative definitions
• Flexibility: Easy to add/modify nodes
• Testability: Each node testable independently
• Observability: Built-in execution tracing
• Performance: Optimize critical paths

Reference

• pydantic-graph: https://github.com/pydantic/pydantic-graph (when available)
• Related: [Agentic Phase 3] Internal context agent for orchestrated search/response #126 (Context agent workflows structured with graphs)
• Related: [Agentic Phase 4] Pipeline orchestration with pydantic-graph #127 (Evaluation framework measures pipeline quality)

[github-actions]

Claude finished Adam Poulemanos (@bashandbone)'s task —— View job

---

Analysis Complete: Pipeline Orchestration with pydantic-graph

I've completed a comprehensive analysis of this issue. Here are the key findings:

---

Summary

Issue #127 proposes integrating pydantic-graph for structured pipeline orchestration in CodeWeaver's agentic search capabilities. This is Phase 4 of a 5-phase epic (#123) to transform CodeWeaver from a semantic search system into an intelligent code understanding platform.

Key Findings

1. Constitutional Alignment ✅

This proposal exceptionally aligns with CodeWeaver's constitution:

• Proven Patterns (Principle II): Leverages pydantic ecosystem
• Evidence-Based Development (Principle III): pydantic-graph is production-ready
• Simplicity Through Architecture (Principle V): Declarative graphs transform complexity into clarity
• AI-First Context (Principle I): Enables sophisticated context curation workflows

2. Technical Status

Already in place:

• ✅ pydantic-graph dependency declared (pyproject.toml:46, version ≥1.4.0)
• ✅ Agent infrastructure scaffolded (src/codeweaver/providers/agent/)
• ✅ pydantic-ai fully integrated with 20+ model providers

Not yet implemented:

• ❌ No pipeline orchestration code exists
• ❌ No graph-based workflow definitions
• ❌ No strategy selection logic

3. pydantic-graph Overview

pydantic-graph is a beta feature within pydantic-ai that provides:

• Declarative workflow definitions using GraphBuilder
• Type-safe node connections with StepContext
• Parallel execution with .map() and join operations
• Conditional branching with .decision() and .match()
• State management and persistence
• Independent testable components

Example Pipeline Pattern:

from pydantic_graph.beta import GraphBuilder, StepContext

@dataclass
class SearchState:
    query: str
    results: list[CodeMatch] = field(default_factory=list)

g = GraphBuilder(state_type=SearchState, output_type=list[CodeMatch])

@g.step
async def analyze_query(ctx: StepContext[SearchState, None, None]) -> str:
    return await agent.analyze_complexity(ctx.state.query)

@g.step
async def fast_search(ctx: StepContext[SearchState, None, None]) -> list[CodeMatch]:
    return await lexical_search(ctx.state.query)

@g.step
async def deep_search(ctx: StepContext[SearchState, None, None]) -> list[CodeMatch]:
    return await semantic_search(ctx.state.query)

# Strategy-based routing
g.add(
    g.edge_from(g.start_node).to(analyze_query),
    g.edge_from(analyze_query).to(
        g.decision()
        .branch(g.match(str, matches=lambda x: x == "simple").to(fast_search))
        .branch(g.match(str, matches=lambda x: x == "complex").to(deep_search))
    ),
    g.edge_from(fast_search, deep_search).to(g.end_node),
)

4. Implementation Difficulty

Estimated: Medium (5-9 weeks, assuming #126 complete)

Favorable factors:

• Library already integrated, no dependency management
• Comprehensive examples in pydantic-ai documentation
• Existing infrastructure (search services, agents, providers)

Challenges:

• Beta API status (may evolve)
• Depends on [Agentic Phase 3] Internal context agent for orchestrated search/response #126 (context agent tooling)
• Visualization tooling unclear from docs

5. Impact Assessment

High positive impact on:

• Maintainability ⭐⭐⭐⭐⭐ - Declarative > imperative code
• Testability ⭐⭐⭐⭐⭐ - Each node independently testable
• Flexibility ⭐⭐⭐⭐⭐ - Add/modify pipelines easily
• Observability ⭐⭐⭐⭐ - Built-in execution tracing

Risks to monitor:

• Beta API stability (mitigate: abstract behind interfaces)
• Performance overhead (mitigate: benchmark early)
• Over-engineering simple flows (mitigate: use graphs only for complex workflows)

Recommendations

✅ Approve and Prioritize

This is a strategic, high-value proposal that:

1. Follows Constitutional principles perfectly
2. Uses proven patterns from pydantic ecosystem
3. Enables sophisticated reasoning ([Agentic Phase 3] Internal context agent for orchestrated search/response #126) and evaluation ([Agentic Phase 5] Evaluation framework with pydantic-eval #128)
4. Positions CodeWeaver as a platform, not just a tool

📋 Pre-Implementation Checklist

Before starting:

• Complete [Agentic Phase 3] Internal context agent for orchestrated search/response #126 (context agent tooling) - provides justifying use cases
• Document strategy patterns: which pipelines benefit from graphs vs. direct code
• Establish performance baselines to measure overhead
• Clarify visualization requirements (pydantic-graph docs don't show tooling)

🏷️ Label Recommendations

Current labels are good. Consider adding:

• pydantic-ai - pydantic-graph is part of pydantic-ai
• architecture - Significant architectural decision
• v0.3 - Target version milestone

Related Issues Analysis

• [EPIC] Agentic enhancements: intelligent search, orchestration, and evaluation #123 (Epic): Overall vision - this is Phase 4 of 5
• [Agentic Phase 3] Internal context agent for orchestrated search/response #126 (Dependency): Context agent tooling - must complete first
• [Agentic Phase 1] Integrate agents into find_code pipeline #124: Agent integration - provides nodes for graphs
• [Agentic Phase 2] Integrate data providers into pipeline #125: Data providers - provides additional nodes
• [Agentic Phase 5] Evaluation framework with pydantic-eval #128: Evaluation framework - will measure pipeline quality

Final Verdict

Strong recommendation to proceed after #126 completion. This represents exactly the kind of "deep quality as competitive moat" thinking outlined in ARCHITECTURE.md Strategic Design Principle #1. Graph-based orchestration will make complex workflows maintainable, testable, and observable—critical for an AI-first context platform.

The constitutional alignment is exceptional, and the pydantic ecosystem integration exemplifies the "Proven Patterns" principle. The beta status is the only concern, but given pydantic-ai's maturity (v1.4.0+) and active community, this risk is acceptable.

---

Branch: claude/issue-127-20251119-2123