ContextFlow uses explicit Pydantic ContextItem schemas instead of arbitrary dictionaries ({"role": "user"}). This enables strong typing, priority ranking, and exact token counting.
class ContextItem(BaseModel):
role: Literal["system", "user", "assistant", "tool"]
content: str
priority: int = 0
tokens: Optional[int] = None
metadata: Dict[str, Any] = Field(default_factory=dict)- Safety: Passing typo'd dictionaries (
{"rola": "user"}) silently fails in production. Pydantic guarantees boundary integrity before theContextPipelineeven begins compressing the array. - Algorithmic Importance (
priority): Raw OpenAI dictionaries have no concept of "importance." By injecting apriorityinteger, our TokenBudget can mathematically droppriority=0"noise" items first when the array overflows Context Window limits. - Internal Telemetry: Storing exact
tiktokenbyte-pair lengths natively inside the array objects prevents duplicate loops across pipeline algorithms.
When the execution reaches the final Provider.arun(), the pipeline seamlessly strips the Pydantic metadata and converts the array natively to the OpenAI dictionary specification using item.to_llm_dict().