Agentic applications require a different approach to data processing—one that brings structure to probabilistic AI systems while maintaining the familiar patterns data engineers already know. Typedef’s Fenic framework solves this by treating AI workloads as traditional data pipelines, using declarative DataFrame APIs that automatically handle LLM orchestration, batching, caching, and rate limiting. This declarative approach decouples heavy batch inference from real-time agent reasoning, making AI systems more debuggable, scalable, and production-ready.
The key insight: agentic workflows are just pipelines. They take inputs, reason over context, generate outputs, and log results—exactly what DataFrame APIs were designed to handle. Rather than building custom orchestration layers, Fenic lets you define what transformations you want declaratively while the framework handles how to execute them efficiently with multiple LLM providers, automatic retries, and cost tracking.
The fundamental problem with traditional agentic architectures
Most agentic applications today suffer from what Typedef calls “pilot paralysis”—87% of enterprise AI projects fail to reach production. The core issue isn’t the agents themselves but the data infrastructure supporting them. Your most valuable data lives in PDFs, audio recordings, images, and video files, requiring preprocessing mazes of OCR models, transcription models, and computer vision. Once you extract text, the hard part begins: managing rate limits across providers, chunking documents for context windows, balancing expensive accurate models against cheaper less reliable ones, and moving data between custom LLM scripts, warehouses, and inference infrastructure.
Traditional frameworks treat LLM calls as opaque User-Defined Functions (UDFs) that query optimizers can’t see into or optimize. This creates bottlenecks, unpredictable costs, and debugging nightmares. Fenic’s innovation is making inference a first-class operation in the query engine itself. When you call semantic operators like semantic.extract() or semantic.join(), the query engine is fully aware that inference is happening and can optimize these operations just like CPU or memory operations—batching requests, caching results, managing rate limits, and tracking costs automatically.
Why DataFrames bring structure to probabilistic systems
DataFrames provide three critical capabilities that make them ideal for agentic applications. Lineage tracking ensures every column and row has traceable origins, even from stochastic model output. You can trace data backwards from results to source rows or forwards to analyze downstream impacts. Columnar consistency keeps data structured and meaningful whether it’s a summary, embedding, or toxicity_score—the schema remains clear even when content is generated by LLMs. Deterministic transformations wrap inference calls in declarative logic where model + prompt + input → output, enabling caching, versioning, and debugging of otherwise probabilistic operations.
This structure is particularly powerful for agent preprocessing. Traditional agentic architectures mix batch inference with real-time decision-making, creating unpredictable latency and resource utilization. Fenic’s recommended pattern separates these concerns:
- Use DataFrames for heavy lifting—extracting structure, enriching context, preparing clean data
- Hand off to agents for real-time reasoning. Your agents focus on decisions, not data prep, resulting in more predictable and responsive systems with better resource utilization through batched LLM calls.
Declarative DataFrame APIs: the core abstraction
Fenic provides a PySpark-inspired DataFrame API specifically designed for AI workloads. The framework is built in Python (87%) with a Rust core (13%) for performance-critical operations. Installation is simple: pip install fenic with support for Python 3.10, 3.11, and 3.12.
The declarative programming model follows lazy evaluation—operations build a logical query plan that’s only executed when an action is called. This allows the query engine to optimize the entire pipeline: reordering operations, batching inference requests across rows, eliminating redundant computations, and caching repeated operations. Developers define what transformations they want; Fenic handles how to execute them efficiently.
A typical pipeline looks like this:
pythonimport fenic as fc from pydantic import BaseModel, Field from typing import Literal from fenic.api.functions import semantic, markdown # Define schema for structured extractionclass TicketSchema(BaseModel): priority: Literal["low", "medium", "high"] description: str = Field(description="Issue description") # Session configurationconfig = fc.SessionConfig( app_name="agent_preprocessing", semantic=fc.SemanticConfig( language_models={ "fast": fc.OpenAILanguageModel( "gpt-4o-mini", rpm=500, tpm=200_000 ), "powerful": fc.AnthropicLanguageModel( "claude-opus-4-0", rpm=100, tpm=100_000 ) }, default_language_model="fast" ) ) session = fc.Session.get_or_create(config) # Load data and process declarativelydf = session.read.csv("customer_tickets.csv") structured_tickets = ( df .with_column("raw_text", fc.col("ticket").cast(fc.MarkdownType)) .with_column( "chunks", markdown.extract_header_chunks(fc.col("raw_text"), header_level=2) ) .explode("chunks") .with_column("extracted", semantic.extract(fc.col("chunks"), TicketSchema)) .unnest("extracted") # Flatten the struct .filter(fc.col("priority") == "high") .with_column("embeddings", semantic.embed(fc.col("description"))) ) # Mock agent for demonstrationclass MockAgent: def process(self, data): print("Agent received structured data:") for d in data: print(d) agent = MockAgent() # Hand off clean, structured data to agentagent.process(structured_tickets.to_pylist())
The API includes all standard DataFrame transformations: select(), filter(), with_column(), join(), group_by(), sort(), union(), explode(), and more. Actions that trigger execution include show(), collect(), count(), and conversions to Pandas, Polars, Arrow, or Python data structures.
Semantic operators: first-class inference operations
Fenic’s breakthrough is treating semantic operations as native DataFrame operations rather than opaque UDFs. Eight core semantic operators enable declarative AI workflows:
semantic.extract transforms unstructured text into structured data using Pydantic schemas for type-safe extraction:
pythonimport fenic as fc from pydantic import BaseModel, Field from typing import List, Literal from fenic.api.functions import semantic # Define schemasclass Issue(BaseModel): category: Literal["bug", "feature", "question"] priority: Literal["low", "medium", "high"] estimated_effort: int = Field(description="Hours to resolve") class Ticket(BaseModel): customer_tier: Literal["free", "pro", "enterprise"] region: Literal["us", "eu", "apac"] issues: List[Issue] # Session config (simple, fast model)config = fc.SessionConfig( app_name="ticket_processing", semantic=fc.SemanticConfig( language_models={ "fast": fc.OpenAILanguageModel("gpt-4o-mini", rpm=100, tpm=50_000) }, default_language_model="fast" ) ) session = fc.Session.get_or_create(config) # Mock dataframe (instead of reading from CSV)data = [ {"raw_ticket": "Enterprise customer in apac reporting a bug: login fails, estimated 4 hours"}, {"raw_ticket": "Pro customer in eu requesting new feature: dark mode, estimated 10 hours"}, {"raw_ticket": "Free customer in us with question: billing, estimated 2 hours"}, {"raw_ticket": "Enterprise customer in apac reporting a high priority bug: payment gateway issue, 12 hours"} ] df = session.create_dataframe(data) # Extraction pipelinetickets = ( df .with_column("extracted", semantic.extract(fc.col("raw_ticket"), Ticket)) .unnest("extracted") # flatten Ticket object into columns .filter(fc.col("region") == "apac") .explode("issues") # expand list of issues into rows) ) # Filter high priority bugshigh_priority_bugs = tickets.filter( (fc.col("issues").category == "bug") & (fc.col("issues").priority == "high") ) # Show resultshigh_priority_bugs.show()
semantic.predicate creates boolean filters using natural language, combining traditional predicates with semantic reasoning:
pythonimport fenic as fc from fenic.api.functions import semantic # Assume df is already a dataframe with columns: years_experience, resume_textqualified_candidates = df.filter( (fc.col("years_experience") > 5) & semantic.predicate( "Does this resume mention agentic AI systems? {{resume}}", resume=fc.col("resume_text") ) ) qualified_candidates.show()
semantic.join matches records based on meaning rather than exact values, enabling fuzzy matching across tables:
pythonpredicate = """Is this candidate a good fit for the job?Candidate Background: {{left_on}}Job Requirements: {{right_on}}Consider: technical skills, experience level, cultural fit"""matched_candidates = ( applicants.semantic.join( other=job_openings, predicate=predicate, left_on=fc.col("resume"), right_on=fc.col("job_description") ) .order_by(fc.desc("match_score")) # Note: order_by typically needs a sort expression .limit(10) )
semantic.map applies natural language transformations to each row, semantic.classify categorizes text into predefined classes with few-shot learning, semantic.reduce aggregates grouped data with LLM operations, semantic.with_cluster_labels clusters rows by semantic similarity using embeddings, and semantic.analyze_sentiment provides built-in sentiment analysis. Each operator is query-engine-aware, enabling automatic optimization.
AI-native data types for text-heavy workloads
Fenic extends beyond typical multimodal types with specialized types for text-heavy AI applications. MarkdownType provides native markdown parsing with structure-aware operations like fc.markdown.extract_header_chunks() for breaking documents into logical sections. TranscriptType handles SRT, WebVTT, and generic formats with speaker and timestamp awareness, essential for meeting or podcast processing. JsonType enables JQ expressions for nested data manipulation. EmbeddingType represents fixed-length embedding vectors with semantic similarity operations. These types make unstructured data first-class citizens in your pipeline.
A practical example processing meeting transcripts:
pythonfrom pydantic import BaseModel, Field from typing import List, Literal import fenic as fc class SegmentAnalysis(BaseModel): speaker: str key_points: List[str] action_items: List[str] sentiment: Literal["positive", "neutral", "negative"] from fenic.api.functions import semantic from fenic.api.functions.builtin import collect_list, length # Read transcript file and cast to TranscriptType# Note: Use read.docs() with content_type or read the file as plain text firstdf = ( session.read.docs("meeting.srt", content_type="markdown") # or handle as plain text .with_column("transcript", fc.col("content").cast(fc.TranscriptType)) ) # Extract insights and aggregate action items by speakermeeting_insights = ( df .with_column("analysis", semantic.extract(fc.col("transcript"), SegmentAnalysis)) .unnest("analysis") # Flatten the struct .filter(length(fc.col("action_items")) > 0) .group_by("speaker") .agg(collect_list("action_items").alias("all_actions")) ) meeting_insights.show()
Building production agentic workflows
The recommended architecture separates concerns between batch preprocessing and real-time agent execution. Fenic handles the data pipeline—loading raw data, extracting structure, enriching with embeddings, classifying and filtering, and preparing clean context. The agent runtime receives structured, pre-processed data and focuses on planning, decision-making, and action execution.
Consider a customer support agent. The traditional approach mixes everything: the agent receives raw tickets, must parse them, extract entities, look up customer history, classify urgency, and only then can reason about responses. This creates unpredictable latency as each ticket requires multiple sequential LLM calls during user interaction.
The Fenic approach preprocesses tickets in batches:
pythonimport fenic as fc from fenic.api.functions import semantic # Batch preprocessing (offline or scheduled)preprocessed = ( tickets_df # Extract structured ticket information .with_column("ticket_data", semantic.extract(fc.col("raw_ticket"), TicketStructure)) .unnest("ticket_data") # Join with customer history using semantic predicate .semantic.join( customer_history_df, predicate="Is this ticket related to previous issue? Ticket: {{ticket_desc}} History: {{history}}", left_on=fc.col("description"), right_on=fc.col("history") ) # Classify urgency (based on tier + issue type) .with_column( "urgency", semantic.classify( fc.col("issue_type"), classes=["low", "medium", "high", "critical"] ) ) # Generate embeddings on ticket description .with_column("embedding", semantic.embed(fc.col("description"))) # Note: vector_search is not shown in docs - this may need adjustment # .with_column( # "similar_resolved_tickets", # fc.vector_search(fc.col("embedding"), resolved_tickets_index, k=5) # )) # Store resultspreprocessed.write.parquet("preprocessed_tickets.parquet") # Agent consumes clean data in real-timefor ticket inpreprocessed.filter(fc.col("urgency") == "high").collect(): response = agent.generate_response( ticket_data=ticket # Already structured and enriched similar_cases=ticket["similar_resolved_tickets"] ) agent.take_action(response)
This pattern delivers 3-5x faster agent response times by moving heavy inference offline, 30-40% cost reduction through intelligent batching and caching, and significantly improved debuggability since data lineage is traceable through the entire pipeline.
Session configuration and multi-provider orchestration
Fenic’s session configuration declaratively defines all LLM providers, rate limits, and execution settings. The framework handles provider-specific APIs, retry logic, and fallbacks automatically:
pythonimport fenic as fc config = fc.SessionConfig( app_name="production_agents", semantic=fc.SemanticConfig( language_models={ "fast": fc.OpenAILanguageModel( "gpt-4o-mini", rpm=500, tpm=200_000 ), "reasoning": fc.AnthropicLanguageModel( "claude-opus-4-0", rpm=100, tpm=100_000, # total tokens per minute profiles={ "quick": fc.AnthropicLanguageModel.Profile( thinking_token_budget=1024 ), "thorough": fc.AnthropicLanguageModel.Profile( thinking_token_budget=4096 ) }, default_profile="quick" ), "gemini": fc.GoogleVertexLanguageModel( "gemini-2.0-flash-lite", rpm=300, tpm=150_000 ) }, default_language_model="fast", embedding_models={ "cohere": fc.CohereEmbeddingModel( "embed-english-v3.0", rpm=100 ) } ) )
You can specify different models for different operations: use fast, cheap models for classification and extraction, reserve powerful models for advanced reasoning, and switch models mid-pipeline based on data characteristics. The framework tracks token usage and costs across all providers automatically.
Production features for reliable systems
Fenic includes critical production capabilities. Caching and persistence avoid redundant computation:
pythonimport fenic as fc from pydantic import BaseModel from typing import List from fenic.api.functions import semantic # Example schemaclass ComplexSchema(BaseModel): doc_id: str text: str metadata: dict # Preprocessing pipeline with persistenceexpensive_preprocessing = ( df .with_column("extracted", semantic.extract(fc.col("documents"), ComplexSchema)) .unnest("extracted") .with_column("embedding", semantic.embed(fc.col("text"))) .persist() # Cache results after first computation) # Subsequent operations reuse cached dataresults_a = expensive_preprocessing.filter(fc.col("metadata")["priority"] == "high").collect() results_b = ( expensive_preprocessing .group_by("doc_id") .agg(fc.collect_list("embedding").alias("all_embeddings")) )
Row-level lineage enables tracing data transformations:
python# Collect data with row-level lineage enabledresult = df.collect(with_lineage=True) # Initialize lineage trackerlineage = df.lineage() # Example: trace backwards from result row(s)source_rows = lineage.backward(row_ids=["result_uuid_123"]) # Example: trace forwards from source row(s)result_rows = lineage.forward(row_ids=["source_uuid_456"]) # Inspect resultsprint("Backwards trace:", source_rows) print("Forwards trace:", result_rows)
Query metrics and cost tracking provide observability:
python# Collect results along with metricsresult = df.collect(with_metrics=True) # Access and print metricsprint(f"Total tokens: {result.metrics.lm_metrics.total_tokens}") print(f"Estimated cost: ${result.metrics.lm_metrics.estimated_cost}") print(f"Query time: {result.metrics.query_metrics.execution_time}s") print(f"Operator breakdown: {result.metrics.operator_metrics}")
Explain plans show query optimization:
pythondf.explain() # Display logical plandf.show(explain_analyze=True) # Show with execution metrics
These features make debugging production issues tractable. When an agent produces unexpected output, you can trace back through the entire data pipeline to identify where structure was lost or inference failed.
Advanced patterns and optimizations
Fuzzy matching before semantic operations reduces costs by using Rust-powered string similarity algorithms (Levenshtein, Jaro-Winkler, Damerau-Levenshtein, Hamming) for initial filtering before expensive LLM calls:
pythonimport fenic as fc from fenic.api.functions.text import compute_fuzzy_ratio from fenic.api.functions import semantic # Block candidates using fuzzy matching firstcandidates = ( df # Fuzzy match against required skills .with_column( "fuzzy_score", compute_fuzzy_ratio(fc.col("resume_skills"), "Python, ML, NLP", method="jaro_winkler") ) # Cheap filter first .filter(fc.col("fuzzy_score") > 70) # Score is 0-100, not 0-1 # Add semantic predicate as boolean column .with_column( "ml_experience_ok", semantic.predicate( "Has 3+ years of Python ML experience? Resume: {{resume}}", resume=fc.col("resume_skills") ) ) # Expensive filter on reduced subset .filter(fc.col("ml_experience_ok")) ) candidates.show()
Jinja templating for dynamic prompts enables data-aware prompt construction:
pythonimport fenic as fc from fenic.api.functions import semantic template = """Analyze this ticket for customer: {{customer_name}}Tier: {{tier}}Previous issues:{% for issue in history %}- {{issue.description}} (resolved: {{issue.resolved}}){% endfor %}Current issue: {{current_ticket}}"""df = df.with_column( "ticket_analysis", semantic.map( template, customer_name=fc.col("name"), tier=fc.col("tier"), history=fc.col("ticket_history"), current_ticket=fc.col("description") ) ) df.show()
User-defined functions extend the framework when needed:
pythonimport fenic as fc # Define custom UDF@fc.udf(return_type=str)def custom_enrichment(text: str, metadata: dict) -> str: # Custom Python logic return f"{text} [Context: {metadata.get('key', 'N/A')}]" # Apply UDF to columns df = df.with_column( "enriched", custom_enrichment(fc.col("text"), fc.col("metadata")) ) df.show()
Model Context Protocol integration
Fenic provides an MCP server at https://mcp.fenic.ai that gives AI assistants deep knowledge of the framework’s API. Configure your AI coding assistant:
bashclaude mcp add -t http fenic-docs https://mcp.fenic.ai
Now your assistant has direct access to Fenic’s complete API documentation and architectural details, providing accurate, context-aware help rather than generic Python advice. Ask questions like “How do I implement semantic join with custom predicates?” or “Show me how to configure multiple LLM providers with different rate limits” and get precise, Fenic-specific guidance.
Real-world impact and use cases
Insurance-tech company Matic uses Fenic to build and deploy semantic extraction pipelines across thousands of policies and transcripts. Their CPO reports: “We’ve dramatically reduced the time it takes to eliminate errors caused by human analysis, significantly cut costs, and lowered our Errors and Omissions risk.”
Content companies have moved beyond static taxonomies to dynamic narrative extraction, analyzing streams of news articles to automatically construct narrative arcs showing how stories evolve over time. Cybersecurity teams process mixed structured and unstructured threat intelligence data. Fintech firms perform transaction enrichment and classification at scale.
Common patterns include semantic feature engineering for recommendation systems, high-precision named entity recognition, automated user-generated content moderation, conversational intelligence platforms, real-time context engineering for AI agents, large-scale content classification, and high-quality data labeling workflows.
Getting started with Fenic
Install Fenic and configure your preferred LLM provider:
bashimport fenic as fc from fenic.api.functions import semantic from fenic.api.functions.builtin import count, desc from pydantic import BaseModel from typing import Literal # <– fix# Define structured schema for error classificationclass ErrorClassification(BaseModel): severity: Literal["low", "medium", "high", "critical"] category: Literal["auth", "database", "network", "application"] user_facing: bool # Session configurationconfig = fc.SessionConfig( app_name="error_analysis", semantic=fc.SemanticConfig( language_models={ "gpt": fc.OpenAILanguageModel("gpt-4o-mini", rpm=100, tpm=50000) } )) # Create or reuse session session = fc.Session.get_or_create(config) # Alternative: session = fc.Session(config) # Load error logs from CSV df = session.read.csv("error_logs.csv") # Apply semantic extraction and aggregationclassified = ( df.filter(fc.col("timestamp") > "2025-01-01") .with_column("classification", semantic.extract(fc.col("error_message"), ErrorClassification)) .unnest("classification") # Flatten the struct .filter(fc.col("severity").isin(["high", "critical"])) .group_by("category") .agg(count("*").alias("count")) .order_by(desc("count"))) # Show results classified.show()
The framework’s examples directory contains 10+ complete applications including meeting transcript processing, podcast summarization, news analysis, semantic join, named entity recognition, document extraction, markdown processing, JSON handling, and feedback clustering. These provide production-ready patterns for common use cases.
Conclusion: The declarative advantage for agentic AI
Building agentic applications with declarative DataFrame APIs fundamentally changes the development experience. You focus on defining what transformations your data needs—extraction schemas, classification criteria, join predicates—while Fenic handles the advanced orchestration of LLM providers, batching, caching, rate limiting, and cost optimization. Inference becomes a first-class operation that the query engine can reason about and optimize, rather than an opaque external call.
The decoupled architecture separating batch preprocessing from real-time agent reasoning creates more predictable, debuggable, and scalable systems. Your agents receive clean, structured, enriched data and focus on decision-making rather than data wrangling. Row-level lineage, comprehensive metrics, and familiar DataFrame operations bring the reliability of traditional data pipelines to probabilistic AI workloads.
Fenic is open source under Apache-2.0, enabling local-first development with the option to deploy to Typedef’s cloud infrastructure with zero code changes. The framework built on open-source columnar and query technologies, while focusing the value on Typedef’s inference-aware engine. For teams moving AI projects from promising prototypes to scalable production workloads, declarative DataFrame APIs provide the missing infrastructure layer that makes agentic applications reliable, maintainable, and cost-effective at scale.
