Production agents need more than a longer prompt. They need a repeatable way to turn raw documents, logs, tables, and conversation traces into small, typed, auditable context surfaces. A declarative context construction pipeline treats context as data: hydrate sources, shape them with deterministic and semantic operators, store the shaped result, and expose bounded views or tools to the agent runtime.
This pattern is useful when agents otherwise spend their prompt budget parsing PDFs, extracting facts, summarizing histories, searching large tool outputs, or carrying entire records forward between turns. Those operations are real work, but they do not all need to happen inside the reasoning model’s context window. Moving extraction, embedding, summarization, filtering, and table construction into an external pipeline lets the agent receive compact results while the system retains provenance and operational control.
The Fenic project illustrates the mechanism with a DataFrame-style API for context engineering. It combines ordinary relational transforms such as select, filter, join, group, window, explode, and unnest with semantic transforms such as schema-backed extraction, embedding, similarity joins, summarization, and document parsing. The resulting tables can then be exposed as direct Python functions or as Model Context Protocol tools with row limits and typed parameters.
Instead of treating context as a pile of strings, the pipeline models it as rows with explicit columns: source path, chunk id, question, answer, timestamp, user id, embedding, relevance score, sensitivity label, or extracted entity. This makes context queryable before it becomes model input. It also makes review easier because engineers can inspect the intermediate tables rather than reverse-engineering a prompt after failure.
Deterministic operators handle the parts that should behave like normal data processing: filtering, projection, aggregation, sorting, pagination, joins, and window functions. Semantic operators call language or embedding models for tasks such as extracting structured facts, parsing documents, creating embeddings, classifying records, mapping free text into summaries, or reducing a group of rows into a compact note.
The agent should not receive every table or every row. A good context pipeline publishes narrow operations: recall facts for one user, return the top K policy answers, search document chunks with citations, or retrieve a profile plus the last few events. Each tool has typed inputs, row limits, descriptions, and result shaping so the agent can ask for context without flooding itself.
Moving extraction and summarization outside the agent prompt saves context budget, but the pipeline still needs versioning, metrics, limits, provenance, and rollback. External context construction should make behavior more inspectable, not more mysterious.
The lifecycle starts by hydrating sources from files, databases, APIs, or in-memory rows. The shaping phase applies deterministic and semantic transforms to produce typed context tables. The serving phase exposes selected tables or parameterized queries as tools. The operating phase records metrics, versions outputs, monitors cost, and rolls back when a context build proves harmful.
import fenic as fc
from pydantic import BaseModel, Field
class QAPair(BaseModel):
question: str = Field(description="Question answered by the source")
answer: str = Field(description="Grounded answer from the source")
session = fc.Session.get_or_create(
fc.SessionConfig(
app_name="policy_context",
semantic=fc.SemanticConfig(
language_models={"gpt": fc.OpenAILanguageModel(model_name="gpt-5-nano")},
embedding_models={"embed": fc.OpenAIEmbeddingModel(model_name="text-embedding-3-small")},
default_embedding_model="embed",
),
)
)
policy_context = (
session.read.pdf_metadata("policies/*.pdf")
.select(
fc.col("file_path").alias("source"),
fc.semantic.parse_pdf(fc.col("file_path")).alias("text"),
)
.select(
"source",
fc.semantic.extract(fc.col("text").cast(fc.StringType), QAPair).alias("qa"),
)
.unnest("qa")
.select(
"source",
"question",
"answer",
fc.semantic.embed(fc.col("question")).alias("embedding"),
)
)
policy_context.write.save_as_table("policy_qa", mode="overwrite")Once the shaped table exists, the agent runtime calls a small retrieval function instead of reading entire documents. The function can enforce user scope, top-K size, output columns, and citation fields before returning anything to the model.
async def qa_neighbors(query: str, k: int = 3):
q = session.create_dataframe([{"q": query}])
results = q.semantic.sim_join(
session.table("policy_qa"),
left_on=fc.semantic.embed(fc.col("q")),
right_on=fc.col("embedding"),
k=k,
similarity_score_column="relevance",
)
return results.select("question", "answer", "source", "relevance")._planFor MCP-based runtimes, a context platform can register parameterized tools or generate system tools from catalog tables. That turns shaped context into a protocol surface while keeping execution, row limits, and table descriptions outside the model’s free-form text.
A support agent can answer questions from policy PDFs by retrieving pre-extracted Q/A rows with source citations. The agent no longer needs to parse PDFs during the conversation, and the context table can be rebuilt when policies change.
Conversation traces can be converted into typed facts, profile blocks, and recent event timelines. At runtime, the agent receives only a scoped memory snapshot or semantic recall result, not every historical message.
Teams can process articles, transcripts, CRM records, or product catalogs by combining structure-aware chunking, schema-backed extraction, embeddings, and aggregate summaries. The same shaped tables can support search, recommendations, evaluation, and agent tools.
Heavy parsing, extraction, embedding, and summarization should usually run in batch or background jobs. Online agent calls should retrieve small, already-shaped records. This separation improves latency and makes context builds reproducible.
# Batch path: expensive semantic operations.
context_table = (
raw_docs
.with_column("chunks", fc.text.recursive_word_chunk("body", chunk_size=500))
.explode("chunks")
.with_column("embedding", fc.semantic.embed(fc.col("chunks")))
.with_column("metadata", fc.semantic.extract("chunks", ChunkMetadata))
)
context_table.write.save_as_table("doc_context_v2026_04_24", mode="overwrite")
# Runtime path: bounded query over prepared context.
def retrieve_context(query: str, k: int):
return semantic_top_k("doc_context_v2026_04_24", query, min(k, 8))Store enough columns to debug the pipeline: source id, transform version, extraction schema version, chunk id, model alias, confidence, and timestamps. When an agent gives a bad answer, engineers should be able to inspect whether retrieval failed, extraction was wrong, or the agent ignored good context.
Raw stores often contain sensitive fields, verbose text, or columns that are useful only to builders. Publish stable views or registered tools that select the fields an agent can safely consume. Use row caps and typed parameters as part of the contract.