feat(context): Phase 3 - Context Scoring & Ranking #81

Closed
opened 2026-01-04 00:51:16 +00:00 by cardosofelipe · 0 comments

Overview

Implement context scoring algorithms and ranking logic for optimal context selection.

Parent Issue

  • #61: Context Management Engine

Implementation Tasks

1. Create scoring/relevance.py

  • Create RelevanceScorer class
  • Implement embedding-based similarity scoring (via Knowledge Base)
  • Handle contexts without embeddings (fallback to keyword matching)
  • Cache relevance scores
class RelevanceScorer:
    def __init__(self, mcp_manager: MCPClientManager):
        self.mcp = mcp_manager
        self._cache: dict[str, float] = {}

    async def score(
        self,
        context: BaseContext,
        query: str
    ) -> float:
        """Score context relevance to query using embeddings."""
        ...

2. Create scoring/recency.py

  • Create RecencyScorer class
  • Implement time-based decay scoring
  • Make decay rate configurable
class RecencyScorer:
    def __init__(self, decay_hours: float = 24.0):
        self.decay_hours = decay_hours

    def score(self, context: BaseContext) -> float:
        """Score based on recency with exponential decay."""
        age_hours = (datetime.utcnow() - context.timestamp).total_seconds() / 3600
        return math.exp(-age_hours / self.decay_hours)

3. Create scoring/composite.py

  • Create CompositeScorer class
  • Implement weighted scoring combination
  • Make weights configurable per context type

4. Create prioritization/ranker.py

  • Create ContextRanker class
  • Implement parallel scoring using asyncio.gather
  • Implement greedy selection within budget
  • Handle priority overrides (priority > 0 forces inclusion)
class ContextRanker:
    async def rank(
        self,
        contexts: list[BaseContext],
        query: str,
        budget: TokenBudget,
        scorer: CompositeScorer
    ) -> list[BaseContext]:
        """Rank contexts by score, respecting budget."""
        ...

Files to Create

backend/app/services/context/
├── scoring/
│   ├── __init__.py
│   ├── relevance.py
│   ├── recency.py
│   └── composite.py
└── prioritization/
    ├── __init__.py
    └── ranker.py

Acceptance Criteria

  • Relevance scoring uses Knowledge Base embeddings
  • Recency scoring implements exponential decay
  • Composite scorer combines weighted scores
  • Ranker respects budget constraints
  • Parallel scoring improves performance
  • Priority overrides work correctly
  • Unit tests for all scoring algorithms

Dependencies

  • #69 (Phase 1 - Foundation)
  • #70 (Phase 2 - Token Budget)

Labels

phase-2, context, backend

## Overview Implement context scoring algorithms and ranking logic for optimal context selection. ## Parent Issue - #61: Context Management Engine --- ## Implementation Tasks ### 1. Create `scoring/relevance.py` - [ ] Create `RelevanceScorer` class - [ ] Implement embedding-based similarity scoring (via Knowledge Base) - [ ] Handle contexts without embeddings (fallback to keyword matching) - [ ] Cache relevance scores ```python class RelevanceScorer: def __init__(self, mcp_manager: MCPClientManager): self.mcp = mcp_manager self._cache: dict[str, float] = {} async def score( self, context: BaseContext, query: str ) -> float: """Score context relevance to query using embeddings.""" ... ``` ### 2. Create `scoring/recency.py` - [ ] Create `RecencyScorer` class - [ ] Implement time-based decay scoring - [ ] Make decay rate configurable ```python class RecencyScorer: def __init__(self, decay_hours: float = 24.0): self.decay_hours = decay_hours def score(self, context: BaseContext) -> float: """Score based on recency with exponential decay.""" age_hours = (datetime.utcnow() - context.timestamp).total_seconds() / 3600 return math.exp(-age_hours / self.decay_hours) ``` ### 3. Create `scoring/composite.py` - [ ] Create `CompositeScorer` class - [ ] Implement weighted scoring combination - [ ] Make weights configurable per context type ### 4. Create `prioritization/ranker.py` - [ ] Create `ContextRanker` class - [ ] Implement parallel scoring using asyncio.gather - [ ] Implement greedy selection within budget - [ ] Handle priority overrides (priority > 0 forces inclusion) ```python class ContextRanker: async def rank( self, contexts: list[BaseContext], query: str, budget: TokenBudget, scorer: CompositeScorer ) -> list[BaseContext]: """Rank contexts by score, respecting budget.""" ... ``` --- ## Files to Create ``` backend/app/services/context/ ├── scoring/ │ ├── __init__.py │ ├── relevance.py │ ├── recency.py │ └── composite.py └── prioritization/ ├── __init__.py └── ranker.py ``` --- ## Acceptance Criteria - [ ] Relevance scoring uses Knowledge Base embeddings - [ ] Recency scoring implements exponential decay - [ ] Composite scorer combines weighted scores - [ ] Ranker respects budget constraints - [ ] Parallel scoring improves performance - [ ] Priority overrides work correctly - [ ] Unit tests for all scoring algorithms --- ## Dependencies - #69 (Phase 1 - Foundation) - #70 (Phase 2 - Token Budget) ## Labels `phase-2`, `context`, `backend`
Sign in to join this conversation.