feat(context): Phase 4 - Context Assembly Pipeline #82

New Issue

cardosofelipe · 2026-01-04T00:51:16Z

cardosofelipe commented

2026-01-04 00:51:16 +00:00

Overview

Implement the full context assembly pipeline: Gather → Count → Score → Rank → Compress → Format.

Parent Issue

#61: Context Management Engine

Implementation Tasks

1. Create `compression/truncation.py`

Create Compressor class
Implement smart truncation (preserve sentence boundaries)
Implement code-aware truncation (preserve function signatures)
Make compression ratio configurable

class Compressor:
    async def compress(
        self,
        contexts: list[BaseContext],
        budget: TokenBudget
    ) -> list[BaseContext]:
        """Compress contexts to fit within budget."""
        ...

    def _truncate_text(
        self,
        text: str,
        max_tokens: int,
        preserve_start: bool = True
    ) -> str:
        """Truncate text at sentence boundaries."""
        ...

2. Create `assembly/pipeline.py`

Create ContextPipeline class
Implement pipeline stages in order
Add timing measurement
Handle assembly timeout

class ContextPipeline:
    async def assemble(
        self,
        contexts: list[BaseContext],
        query: str,
        model: str,
        max_tokens: int
    ) -> AssembledContext:
        """
        Pipeline: Gather → Count → Score → Rank → Compress → Format
        """
        start = time.perf_counter()
        ...
        elapsed_ms = (time.perf_counter() - start) * 1000
        ...

3. Create `assembly/formatter.py`

Create ContextFormatter class
Implement model detection (Claude vs OpenAI vs default)
Route to appropriate adapter

class ContextFormatter:
    def format(
        self,
        contexts: list[BaseContext],
        model: str
    ) -> str:
        """Format contexts for specific model."""
        adapter = self._get_adapter(model)
        return adapter.format(contexts)

4. Create `AssembledContext` dataclass

Define result structure with content, token_count, contexts_included, assembly_time_ms
Implement JSON serialization for caching

Files to Create

backend/app/services/context/
├── compression/
│   ├── __init__.py
│   └── truncation.py
└── assembly/
    ├── __init__.py
    ├── pipeline.py
    └── formatter.py

Acceptance Criteria

Pipeline executes all stages in correct order
Compression preserves sentence/code boundaries
Assembly completes in <100ms for typical requests
Timeout handling prevents hung requests
AssembledContext contains all metadata
Unit tests for compression and pipeline

Dependencies

#69 (Phase 1 - Foundation)
#70 (Phase 2 - Token Budget)
#71 (Phase 3 - Scoring & Ranking)

Labels

phase-2, context, backend

## Overview Implement the full context assembly pipeline: Gather → Count → Score → Rank → Compress → Format. ## Parent Issue - #61: Context Management Engine --- ## Implementation Tasks ### 1. Create `compression/truncation.py` - [ ] Create `Compressor` class - [ ] Implement smart truncation (preserve sentence boundaries) - [ ] Implement code-aware truncation (preserve function signatures) - [ ] Make compression ratio configurable ```python class Compressor: async def compress( self, contexts: list[BaseContext], budget: TokenBudget ) -> list[BaseContext]: """Compress contexts to fit within budget.""" ... def _truncate_text( self, text: str, max_tokens: int, preserve_start: bool = True ) -> str: """Truncate text at sentence boundaries.""" ... ``` ### 2. Create `assembly/pipeline.py` - [ ] Create `ContextPipeline` class - [ ] Implement pipeline stages in order - [ ] Add timing measurement - [ ] Handle assembly timeout ```python class ContextPipeline: async def assemble( self, contexts: list[BaseContext], query: str, model: str, max_tokens: int ) -> AssembledContext: """ Pipeline: Gather → Count → Score → Rank → Compress → Format """ start = time.perf_counter() ... elapsed_ms = (time.perf_counter() - start) * 1000 ... ``` ### 3. Create `assembly/formatter.py` - [ ] Create `ContextFormatter` class - [ ] Implement model detection (Claude vs OpenAI vs default) - [ ] Route to appropriate adapter ```python class ContextFormatter: def format( self, contexts: list[BaseContext], model: str ) -> str: """Format contexts for specific model.""" adapter = self._get_adapter(model) return adapter.format(contexts) ``` ### 4. Create `AssembledContext` dataclass - [ ] Define result structure with content, token_count, contexts_included, assembly_time_ms - [ ] Implement JSON serialization for caching --- ## Files to Create ``` backend/app/services/context/ ├── compression/ │ ├── __init__.py │ └── truncation.py └── assembly/ ├── __init__.py ├── pipeline.py └── formatter.py ``` --- ## Acceptance Criteria - [ ] Pipeline executes all stages in correct order - [ ] Compression preserves sentence/code boundaries - [ ] Assembly completes in <100ms for typical requests - [ ] Timeout handling prevents hung requests - [ ] AssembledContext contains all metadata - [ ] Unit tests for compression and pipeline --- ## Dependencies - #69 (Phase 1 - Foundation) - #70 (Phase 2 - Token Budget) - #71 (Phase 3 - Scoring & Ranking) ## Labels `phase-2`, `context`, `backend`

cardosofelipe referenced this issue

2026-01-04 00:52:21 +00:00

feat(context): Phase 6 - Caching Layer #84

cardosofelipe referenced this issue

2026-01-04 00:52:21 +00:00

feat(context): Phase 7 - Main Engine & Integration #85

cardosofelipe referenced this issue

2026-01-04 00:54:11 +00:00

feat(mcp): Context Management Engine #61

cardosofelipe referenced this issue from a commit

2026-01-04 14:23:32 +00:00

feat(context): implement assembly pipeline and compression (#82)