Context Engineering в продакшене: паттерны и оптимизация

from dataclasses import dataclass, field

@dataclass
class ContextComponent:
    name: str
    content: str
    tokens: int
    priority: int

@dataclass
class ContextBudget:
    total: int
    output_reserve: int
    lane_caps: dict[str, int] = field(default_factory=dict)

    @property
    def available(self) -> int:
        return self.total - self.output_reserve

class ContextAssembler:
    def __init__(self, budget: ContextBudget):
        self.budget = budget

    async def assemble(self, query: str, intent: str, sources: dict[str, str]) -> list[ContextComponent]:
        components: list[ContextComponent] = []

        if "system" in sources:
            components.append(
                ContextComponent("system", sources["system"], count_tokens(sources["system"]), 1)
            )

        if intent in {"knowledge", "support"} and "retrieval" in sources:
            retrieved = truncate_to_tokens(
                sources["retrieval"],
                self.budget.lane_caps.get("retrieval", 4000),
            )
            components.append(
                ContextComponent("retrieval", retrieved, count_tokens(retrieved), 3)
            )

        if "memory" in sources:
            memory = truncate_to_tokens(
                sources["memory"],
                self.budget.lane_caps.get("memory", 2000),
            )
            components.append(
                ContextComponent("memory", memory, count_tokens(memory), 2)
            )

        if "tools" in sources:
            tools = truncate_to_tokens(
                sources["tools"],
                self.budget.lane_caps.get("tools", 1500),
            )
            components.append(
                ContextComponent("tools", tools, count_tokens(tools), 2)
            )

        return components

def truncate_by_priority(components: list[ContextComponent], max_tokens: int) -> list[ContextComponent]:
    total = sum(c.tokens for c in components)
    if total <= max_tokens:
        return components

    overflow = total - max_tokens
    ordered = sorted(components, key=lambda c: -c.priority)
    kept: list[ContextComponent] = []

    for component in ordered:
        if overflow > 0 and component.priority >= 3:
            if component.tokens <= overflow:
                overflow -= component.tokens
                continue
            new_size = component.tokens - overflow
            component.content = truncate_to_tokens(component.content, new_size)
            component.tokens = new_size
            overflow = 0
        kept.append(component)

    return sorted(kept, key=lambda c: c.priority)

import anthropic

client = anthropic.Anthropic()

SYSTEM_PROMPT = """Ты ассистент поддержки.
Политика ответа:
- не выдумывай факты
- соблюдай policy
- при нехватке данных говори об этом явно
"""

response = client.messages.create(
    model="YOUR_CURRENT_CLAUDE_MODEL_ID",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": SYSTEM_PROMPT,
            "cache_control": {"type": "ephemeral"},
        }
    ],
    messages=[{"role": "user", "content": "Как изменить тариф?"}],
)

from openai import OpenAI

client = OpenAI()

response = client.responses.create(
    model="gpt-5.4-mini",
    input=[
        {
            "role": "developer",
            "content": "Длинный стабильный system/policy prefix ...",
        },
        {
            "role": "user",
            "content": "Почему с меня списали двойную комиссию?",
        },
    ],
)

def track_context_metrics(trace_id: str, utilization: float, cache_hit: bool, truncated: bool):
    langfuse.score(trace_id=trace_id, name="context_utilization", value=utilization)
    langfuse.score(trace_id=trace_id, name="cache_hit", value=1 if cache_hit else 0)
    langfuse.score(trace_id=trace_id, name="truncated", value=1 if truncated else 0)