context

context

Context injection — retrieve relevant memory and inject into prompts.

Classes

ContextConfig dataclass

ContextConfig(enabled: bool = True, top_k: int = 5, min_score: float = 0.0, max_context_tokens: int = 2048)

Controls how retrieved context is injected into prompts.

Functions

format_context

format_context(results: List[RetrievalResult]) -> str

Format retrieval results into a context block.

Each result is prefixed with its source attribution.

Source code in src/openjarvis/tools/storage/context.py

def format_context(results: List[RetrievalResult]) -> str:
    """Format retrieval results into a context block.

    Each result is prefixed with its source attribution.
    """
    if not results:
        return ""

    lines = []
    for r in results:
        source_tag = f"[Source: {r.source}]" if r.source else ""
        if source_tag:
            lines.append(f"{source_tag} {r.content}")
        else:
            lines.append(r.content)

    return "\n\n".join(lines)

build_context_message

build_context_message(results: List[RetrievalResult], facts: Sequence[Fact] = ()) -> Message

Create a system message with formatted context.

Source code in src/openjarvis/tools/storage/context.py

def build_context_message(
    results: List[RetrievalResult],
    facts: Sequence[Fact] = (),
) -> Message:
    """Create a system message with formatted context."""
    # Defensive filtering here protects direct callers as well as the normal
    # inject_context() path. Quarantined facts must never become instructions
    # merely because a caller skipped the budget-selection helper.
    facts = _trusted_facts(facts)
    sections = []
    if facts:
        fact_text = "\n".join(f"- {fact.text}" for fact in facts)
        sections.append(
            "The following durable facts were remembered from prior "
            "conversations. Use them when relevant to the user's request:\n\n"
            + fact_text
        )
    if results:
        sections.append(
            "The following context was retrieved from the knowledge"
            " base. Use it to inform your response, citing sources"
            " where applicable:\n\n" + format_context(results)
        )
    content = "\n\n".join(sections)
    return Message(
        role=Role.SYSTEM,
        content=content,
        metadata={"memory_context": True},
    )

inject_context

inject_context(query: str, messages: List[Message], backend: Optional[MemoryBackend], *, config: Optional[ContextConfig] = None, facts: Sequence[Fact] = ()) -> List[Message]

Retrieve relevant context and prepend it to messages.

Returns a new list — the original list is not mutated. Automatic-memory facts are included independently of the retrieval backend, so persisted facts remain recallable even when the document store is empty. If no facts or results are available, returns the original messages unchanged.

PARAMETER	DESCRIPTION
query	The user query to search for. TYPE: str
messages	The existing message list. TYPE: List[Message]
backend	The memory backend to search, or None when only facts are available. TYPE: Optional[MemoryBackend]
config	Context injection settings (uses defaults if None). TYPE: Optional[ContextConfig] DEFAULT: None
facts	Durable facts captured by the automatic memory service. Quarantined provenance tiers are excluded before budgeting or prompt construction. TYPE: Sequence[Fact] DEFAULT: ()

Source code in src/openjarvis/tools/storage/context.py

def inject_context(
    query: str,
    messages: List[Message],
    backend: Optional[MemoryBackend],
    *,
    config: Optional[ContextConfig] = None,
    facts: Sequence[Fact] = (),
) -> List[Message]:
    """Retrieve relevant context and prepend it to *messages*.

    Returns a **new** list — the original list is not mutated.
    Automatic-memory facts are included independently of the retrieval
    backend, so persisted facts remain recallable even when the document
    store is empty. If no facts or results are available, returns the original
    messages unchanged.

    Parameters
    ----------
    query:
        The user query to search for.
    messages:
        The existing message list.
    backend:
        The memory backend to search, or ``None`` when only facts are available.
    config:
        Context injection settings (uses defaults if ``None``).
    facts:
        Durable facts captured by the automatic memory service. Quarantined
        provenance tiers are excluded before budgeting or prompt construction.
    """
    cfg = config or ContextConfig()
    if not cfg.enabled:
        return messages

    results = backend.retrieve(query, top_k=cfg.top_k) if backend is not None else []

    # Filter by minimum score
    results = [r for r in results if r.score >= cfg.min_score]

    # When both sources have data, cap facts at half the total budget so they
    # cannot starve query-specific document retrieval. Unused fact budget is
    # still available to documents. Newest facts win within the fact budget.
    fact_budget = cfg.max_context_tokens
    if results:
        fact_budget //= 2
    selected_facts: List[Fact] = []
    total_tokens = 0
    for fact in reversed(_trusted_facts(facts)):
        tokens = _count_tokens(fact.text)
        if total_tokens + tokens > fact_budget:
            continue
        selected_facts.append(fact)
        total_tokens += tokens

    # Fill the remaining context budget with retrieved documents.
    truncated: List[RetrievalResult] = []
    for r in results:
        tokens = _count_tokens(r.content)
        if total_tokens + tokens > cfg.max_context_tokens:
            # A large top result should not disappear solely because facts
            # consumed their reserved share. Prefer that result when it fits
            # the total budget on its own.
            if not truncated and selected_facts and tokens <= cfg.max_context_tokens:
                selected_facts = []
                total_tokens = 0
            else:
                break
        if total_tokens + tokens > cfg.max_context_tokens:
            break
        truncated.append(r)
        total_tokens += tokens

    if not selected_facts and not truncated:
        return messages

    # Publish event
    bus = get_event_bus()
    bus.publish(
        EventType.MEMORY_RETRIEVE,
        {
            "context_injection": True,
            "query": query,
            "num_results": len(truncated),
            "num_facts": len(selected_facts),
            "total_tokens": total_tokens,
        },
    )

    # Build context message and prepend
    ctx_msg = build_context_message(truncated, selected_facts)
    return _merge_context_message(messages, ctx_msg)