AI Agents¶
lumen.ai.agents
¶
__all__ = ['Agent', 'AnalysisAgent', 'BaseCodeAgent', 'ChatAgent', 'DbtslAgent', 'DeckGLAgent', 'DocumentListAgent', 'hvPlotAgent', 'SQLAgent', 'TableListAgent', 'ValidationAgent', 'VegaLiteAgent']
module-attribute
¶
Agent
¶
Bases:
Agents are actors responsible for taking a user query and performing a particular task, either by adding context or generating outputs.
Agents have access to an LLM and are given context and can solve tasks by executing a series of prompts or by rendering contents such as forms or widgets to gather user input.
agents = param.List(doc='\n List of agents this agent can invoke.')
class-attribute
instance-attribute
¶
debug = param.Boolean(default=False, doc='\n Whether to enable verbose error reporting.')
class-attribute
instance-attribute
¶
llm = param.ClassSelector(class_=Llm, doc='\n The LLM implementation to query.')
class-attribute
instance-attribute
¶
user = param.String(default='Agent', doc='\n The name of the user that will be respond to the user query.')
class-attribute
instance-attribute
¶
applies(context)
async
classmethod
¶
Additional checks to determine if the agent should be used.
respond(messages, context, step_title=None)
async
¶
Provides a response to the user query.
The type of the response may be a simple string or an object.
Arguments
messages: list[Message] The list of messages corresponding to the user query and any other system messages to be included. context: TContext A mapping containing context for the agent to perform its task. step_title: str | None If the Agent response is part of a longer query this describes the step currently being processed.
AnalysisAgent
¶
Bases:
analyses = param.List([])
class-attribute
instance-attribute
¶
conditions = param.List(default=['Use for custom analysis, advanced analytics, or domain-specific methods', "Use when the user query matches one of the available analyses' name or description below.", "Include the selected analysis' required cols in the instructions", 'NOT for simple queries or basic visualizations'])
class-attribute
instance-attribute
¶
input_schema = AnalysisInputs
class-attribute
instance-attribute
¶
output_schema = AnalysisOutputs
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'template': PROMPTS_DIR / 'AnalysisAgent' / 'main.jinja2', 'response_model': make_analysis_model}})
class-attribute
instance-attribute
¶
purpose = param.String(default='Perform custom analyses that are reliable and repeatable.')
class-attribute
instance-attribute
¶
respond(messages, context, step_title=None)
async
¶
BaseCodeAgent
¶
Bases:
Base class for view agents that can generate and execute code.
Subclasses must:
- Set _executor_class to the appropriate CodeExecutor subclass
- Implement _generate_code_spec() for their specific code generation flow
code_execution = param.Selector(default='disabled', objects=['disabled', 'prompt', 'llm', 'allow'], doc="\n Code execution mode for generating visualizations via code:\n - disabled: No code execution; generate declarative specs only (safe for production)\n - prompt: Generate code, prompt user for permission to execute\n - llm: Generate code, validate with LLM safety check, then execute\n - allow: Generate and execute code without user confirmation\n\n ⚠️ WARNING: The 'prompt', 'llm', and 'allow' modes execute LLM-generated code and\n must NEVER be enabled in production environments with access to secrets, credentials,\n or sensitive data.\n ", allow_refs=True)
class-attribute
instance-attribute
¶
code_execution_enabled
property
¶
Whether code execution is enabled (any mode except 'disabled').
ChatAgent
¶
Bases:
ChatAgent provides general information about available data and other topics to the user. When data is available, it acts as an analyst providing insights and interpretations.
conditions = param.List(default=["Use for general conversation that doesn't require fetching or querying data", 'Use for technical questions about programming, functions, methods, libraries, or APIs', "Use when user asks to 'explain', 'interpret', 'analyze', 'summarize', or 'comment on' existing data in context", "NOT when user asks to 'show', 'get', 'fetch', 'query', 'filter', 'calculate', 'aggregate', or 'transform' data", 'NOT for creating new data transformations - only for explaining data that already exists'])
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'template': PROMPTS_DIR / 'ChatAgent' / 'main.jinja2'}})
class-attribute
instance-attribute
¶
purpose = param.String(default='\n Provides conversational assistance and interprets existing results.\n Handles general questions, technical documentation, and programming help.\n When data has been retrieved, explains findings in accessible terms.')
class-attribute
instance-attribute
¶
respond(messages, context, step_title=None)
async
¶
DbtslAgent
¶
Bases:
Responsible for creating and executing queries against a dbt Semantic Layer to answer user questions about business metrics.
conditions = param.List(default=['Always use this when dbtsl_metaset is available'])
class-attribute
instance-attribute
¶
output_schema = DbtslOutputs
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'response_model': DbtslQueryParams, 'template': PROMPTS_DIR / 'DbtslAgent' / 'main.jinja2'}, 'revise_output': {'response_model': RetrySpec, 'template': PROMPTS_DIR / 'BaseLumenAgent' / 'revise_output.jinja2'}})
class-attribute
instance-attribute
¶
purpose = param.String(default='\n Responsible for displaying data to answer user queries about\n business metrics using dbt Semantic Layers. This agent can compile\n and execute metric queries against a dbt Semantic Layer.')
class-attribute
instance-attribute
¶
requires = param.List(default=['source', 'dbtsl_metaset'], readonly=True)
class-attribute
instance-attribute
¶
source = param.ClassSelector(class_=BaseSQLSource, doc='\n The source associated with the dbt Semantic Layer.')
class-attribute
instance-attribute
¶
user = param.String(default='DBT')
class-attribute
instance-attribute
¶
respond(messages, context, step_title=None)
async
¶
Responds to user messages by generating and executing a dbt Semantic Layer query.
DeckGLAgent
¶
Bases:
Agent for generating DeckGL 3D map visualizations.
Supports two generation modes: - When code_execution is 'disabled': Generate DeckGL JSON specs directly - When code_execution is enabled: Generate PyDeck Python code, execute safely, convert to spec
conditions = param.List(default=['Use for 3D geographic visualizations, map-based data, or when user requests DeckGL/deck.gl', 'Use for large-scale geospatial data with latitude/longitude coordinates', 'Use for hexbin aggregations, heatmaps, or 3D extruded visualizations on maps'])
class-attribute
instance-attribute
¶
default_map_style = 'https://basemaps.cartocdn.com/gl/dark-matter-gl-style/style.json'
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'response_model': DeckGLSpec, 'template': PROMPTS_DIR / 'DeckGLAgent' / 'main.jinja2'}, 'main_pydeck': {'response_model': PyDeckSpec, 'template': PROMPTS_DIR / 'DeckGLAgent' / 'main_pydeck.jinja2'}, 'code_safety': {'response_model': CodeSafetyCheck, 'template': PROMPTS_DIR / 'DeckGLAgent' / 'code_safety.jinja2'}, 'revise_output': {'response_model': RetrySpec, 'template': PROMPTS_DIR / 'DeckGLAgent' / 'revise_output.jinja2'}})
class-attribute
instance-attribute
¶
purpose = param.String(default='Generates DeckGL 3D map visualizations from geographic data.')
class-attribute
instance-attribute
¶
user = param.String(default='DeckGL')
class-attribute
instance-attribute
¶
view_type = DeckGLView
class-attribute
instance-attribute
¶
respond(messages, context, step_title=None)
async
¶
Generate a DeckGL visualization.
DocumentListAgent
¶
Bases:
The DocumentListAgent lists all available documents provided by the user.
conditions = param.List(default=['Use when user asks to list or see all available documents', 'NOT when user asks about specific document content'])
class-attribute
instance-attribute
¶
input_schema = DocumentListInputs
class-attribute
instance-attribute
¶
purpose = param.String(default='\n Displays a list of all available documents.')
class-attribute
instance-attribute
¶
applies(context)
async
classmethod
¶
SQLAgent
¶
Bases:
conditions = param.List(default=['Use for querying, filtering, aggregating, or transforming data with SQL', "Use for calculations that require executing SQL (e.g., 'calculate average', 'sum by category')", "Use when user asks to 'show', 'get', 'fetch', 'query', 'find', 'filter', 'calculate', 'aggregate', or 'transform' data", "NOT when user asks to 'explain', 'interpret', 'analyze', 'summarize', or 'comment on' existing data", 'NOT useful if the user is using the same data for plotting'])
class-attribute
instance-attribute
¶
exclusions = param.List(default=['dbtsl_metaset'])
class-attribute
instance-attribute
¶
exploration_enabled = param.Boolean(default=True, allow_refs=True, doc='\n Whether to enable SQL exploration mode. When False, only attempts oneshot SQL generation.')
class-attribute
instance-attribute
¶
input_schema = SQLInputs
class-attribute
instance-attribute
¶
not_with = param.List(default=['DbtslAgent', 'MetadataLookup', 'TableListAgent'])
class-attribute
instance-attribute
¶
output_schema = SQLEditors
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'response_model': make_sql_model, 'template': PROMPTS_DIR / 'SQLAgent' / 'main.jinja2'}, 'select_tables': {'response_model': make_table_selection_model, 'template': PROMPTS_DIR / 'SQLAgent' / 'select_tables.jinja2'}, 'select_discoveries': {'response_model': make_discovery_model, 'template': PROMPTS_DIR / 'SQLAgent' / 'select_discoveries.jinja2'}, 'check_sufficiency': {'response_model': make_discovery_model, 'template': PROMPTS_DIR / 'SQLAgent' / 'check_sufficiency.jinja2'}, 'revise_output': {'response_model': RetrySpec, 'template': PROMPTS_DIR / 'SQLAgent' / 'revise_output.jinja2'}})
class-attribute
instance-attribute
¶
purpose = param.String(default='\n Creates and executes SQL queries to retrieve, filter, aggregate, or transform data.\n Handles table joins, WHERE clauses, GROUP BY, calculations, and other SQL operations.\n Generates new data pipelines from SQL transformations.')
class-attribute
instance-attribute
¶
user = param.String(default='SQL')
class-attribute
instance-attribute
¶
respond(messages, context, step_title=None)
async
¶
Execute SQL generation with table selection, then one-shot attempt, then exploration if needed.
revise(feedback, messages, context, view=None, spec=None, language=None, errors=None, **kwargs)
async
¶
TableListAgent
¶
Bases:
The TableListAgent lists all available data and lets the user pick one.
conditions = param.List(default=["Use when user explicitly asks to 'list data', 'show available data', or 'what data do you have'", 'NOT for showing actual data contents, querying, or analyzing data', 'NOT for describing data sources or telling what the data is about'])
class-attribute
instance-attribute
¶
input_schema = TableListInputs
class-attribute
instance-attribute
¶
not_with = param.List(default=['DbtslAgent', 'SQLAgent'])
class-attribute
instance-attribute
¶
purpose = param.String(default='\n Displays a list of all available data & datasets. Not useful for identifying which dataset to use for analysis.')
class-attribute
instance-attribute
¶
applies(context)
async
classmethod
¶
ValidationAgent
¶
Bases:
ValidationAgent focuses solely on validating whether the executed plan fully answered the user's original query. It identifies missing elements and suggests next steps when validation fails.
conditions = param.List(default=['Use to validate whether executed plans fully answered user queries', 'Use to identify missing elements from the original user request', 'NOT for data analysis, pattern identification, or technical programming questions'])
class-attribute
instance-attribute
¶
input_schema = ValidationInputs
class-attribute
instance-attribute
¶
output_schema = ValidationOutputs
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'template': PROMPTS_DIR / 'ValidationAgent' / 'main.jinja2', 'response_model': QueryCompletionValidation, 'tools': []}})
class-attribute
instance-attribute
¶
purpose = param.String(default="\n Validates whether executed plans fully answered the user's original query.\n Identifies missing elements, assesses completeness, and suggests next steps\n when validation fails. Acts as a quality gate for plan execution.")
class-attribute
instance-attribute
¶
respond(messages, context, step_title=None)
async
¶
VegaLiteAgent
¶
Bases:
conditions = param.List(default=['Use for publication-ready visualizations or when user specifically requests Vega-Lite charts', 'Use for polished charts intended for presentation or sharing'])
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'response_model': VegaLiteSpec, 'template': PROMPTS_DIR / 'VegaLiteAgent' / 'main.jinja2'}, 'main_altair': {'response_model': AltairSpec, 'template': PROMPTS_DIR / 'VegaLiteAgent' / 'main_altair.jinja2'}, 'code_safety': {'response_model': CodeSafetyCheck, 'template': PROMPTS_DIR / 'VegaLiteAgent' / 'code_safety.jinja2'}, 'interaction_polish': {'response_model': VegaLiteSpecUpdate, 'template': PROMPTS_DIR / 'VegaLiteAgent' / 'interaction_polish.jinja2'}, 'annotate_plot': {'response_model': VegaLiteSpecUpdate, 'template': PROMPTS_DIR / 'VegaLiteAgent' / 'annotate_plot.jinja2'}, 'revise_output': {'response_model': RetrySpec, 'template': PROMPTS_DIR / 'VegaLiteAgent' / 'revise_output.jinja2'}})
class-attribute
instance-attribute
¶
purpose = param.String(default='Generates a vega-lite plot specification from the input data pipeline.')
class-attribute
instance-attribute
¶
user = param.String(default='Vega')
class-attribute
instance-attribute
¶
vector_store_path = param.Path(default=None, check_exists=False, doc='\n Path to a custom vector store for storing and retrieving Vega-Lite examples;\n if not provided a default store will be used depending on the LLM--\n OpenAIEmbeddings for OpenAI LLM or NumpyEmbeddings for all others.')
class-attribute
instance-attribute
¶
view_type = VegaLiteView
class-attribute
instance-attribute
¶
annotate(instruction, messages, context, spec)
async
¶
Apply annotations based on user request.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
instruction
|
|
User's description of what to annotate |
required |
messages
|
|
Chat history for context |
required |
context
|
|
Session context |
required |
spec
|
|
The current VegaLite specification (full dict with 'spec' key) |
required |
Returns:
| Type | Description |
|---|---|
|
Updated specification with annotations |
respond(messages, context, step_title=None)
async
¶
Generates a VegaLite visualization using progressive building approach with real-time updates.
revise(feedback, messages, context, view=None, spec=None, language=None, errors=None, **kwargs)
async
¶
hvPlotAgent
¶
Bases:
conditions = param.List(default=['Use for exploratory data analysis, interactive plots, and dynamic filtering', 'Use for quick, iterative data visualization during analysis'])
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'template': PROMPTS_DIR / 'hvPlotAgent' / 'main.jinja2'}})
class-attribute
instance-attribute
¶
purpose = param.String(default='Generates a plot of the data given a user prompt.')
class-attribute
instance-attribute
¶
view_type = hvPlotUIView
class-attribute
instance-attribute
¶
analysis
¶
AnalysisAgent
¶
Bases:
analyses = param.List([])
class-attribute
instance-attribute
¶
conditions = param.List(default=['Use for custom analysis, advanced analytics, or domain-specific methods', "Use when the user query matches one of the available analyses' name or description below.", "Include the selected analysis' required cols in the instructions", 'NOT for simple queries or basic visualizations'])
class-attribute
instance-attribute
¶
input_schema = AnalysisInputs
class-attribute
instance-attribute
¶
output_schema = AnalysisOutputs
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'template': PROMPTS_DIR / 'AnalysisAgent' / 'main.jinja2', 'response_model': make_analysis_model}})
class-attribute
instance-attribute
¶
purpose = param.String(default='Perform custom analyses that are reliable and repeatable.')
class-attribute
instance-attribute
¶
respond(messages, context, step_title=None)
async
¶
AnalysisOutputs
¶
make_analysis_model(analyses)
¶
base
¶
Agent
¶
Bases:
Agents are actors responsible for taking a user query and performing a particular task, either by adding context or generating outputs.
Agents have access to an LLM and are given context and can solve tasks by executing a series of prompts or by rendering contents such as forms or widgets to gather user input.
agents = param.List(doc='\n List of agents this agent can invoke.')
class-attribute
instance-attribute
¶
debug = param.Boolean(default=False, doc='\n Whether to enable verbose error reporting.')
class-attribute
instance-attribute
¶
llm = param.ClassSelector(class_=Llm, doc='\n The LLM implementation to query.')
class-attribute
instance-attribute
¶
user = param.String(default='Agent', doc='\n The name of the user that will be respond to the user query.')
class-attribute
instance-attribute
¶
applies(context)
async
classmethod
¶
Additional checks to determine if the agent should be used.
respond(messages, context, step_title=None)
async
¶
Provides a response to the user query.
The type of the response may be a simple string or an object.
Arguments
messages: list[Message] The list of messages corresponding to the user query and any other system messages to be included. context: TContext A mapping containing context for the agent to perform its task. step_title: str | None If the Agent response is part of a longer query this describes the step currently being processed.
base_code
¶
Base class for agents that execute LLM-generated code.
This provides a clean inheritance hierarchy: BaseViewAgent └── BaseCodeAgent ├── VegaLiteAgent └── (future code-executing agents)
BaseCodeAgent
¶
Bases:
Base class for view agents that can generate and execute code.
Subclasses must:
- Set _executor_class to the appropriate CodeExecutor subclass
- Implement _generate_code_spec() for their specific code generation flow
code_execution = param.Selector(default='disabled', objects=['disabled', 'prompt', 'llm', 'allow'], doc="\n Code execution mode for generating visualizations via code:\n - disabled: No code execution; generate declarative specs only (safe for production)\n - prompt: Generate code, prompt user for permission to execute\n - llm: Generate code, validate with LLM safety check, then execute\n - allow: Generate and execute code without user confirmation\n\n ⚠️ WARNING: The 'prompt', 'llm', and 'allow' modes execute LLM-generated code and\n must NEVER be enabled in production environments with access to secrets, credentials,\n or sensitive data.\n ", allow_refs=True)
class-attribute
instance-attribute
¶
code_execution_enabled
property
¶
Whether code execution is enabled (any mode except 'disabled').
base_list
¶
base_lumen
¶
BaseLumenAgent
¶
Bases:
prompts = param.Dict(default={'revise_output': {'response_model': RetrySpec, 'template': PROMPTS_DIR / 'BaseLumenAgent' / 'revise_output.jinja2'}})
class-attribute
instance-attribute
¶
user = param.String(default='Lumen')
class-attribute
instance-attribute
¶
revise(instruction, messages, context, view=None, spec=None, language=None, errors=None, **kwargs)
async
¶
Retry the output by line, allowing the user to provide instruction on why the output was not satisfactory, or an error.
base_view
¶
BaseViewAgent
¶
Bases:
input_schema = ViewInputs
class-attribute
instance-attribute
¶
output_schema = ViewOutputs
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'template': PROMPTS_DIR / 'BaseViewAgent' / 'main.jinja2'}})
class-attribute
instance-attribute
¶
respond(messages, context, step_title=None)
async
¶
Generates a visualization based on user messages and the current data pipeline.
ViewInputs
¶
chat
¶
ChatAgent
¶
Bases:
ChatAgent provides general information about available data and other topics to the user. When data is available, it acts as an analyst providing insights and interpretations.
conditions = param.List(default=["Use for general conversation that doesn't require fetching or querying data", 'Use for technical questions about programming, functions, methods, libraries, or APIs', "Use when user asks to 'explain', 'interpret', 'analyze', 'summarize', or 'comment on' existing data in context", "NOT when user asks to 'show', 'get', 'fetch', 'query', 'filter', 'calculate', 'aggregate', or 'transform' data", 'NOT for creating new data transformations - only for explaining data that already exists'])
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'template': PROMPTS_DIR / 'ChatAgent' / 'main.jinja2'}})
class-attribute
instance-attribute
¶
purpose = param.String(default='\n Provides conversational assistance and interprets existing results.\n Handles general questions, technical documentation, and programming help.\n When data has been retrieved, explains findings in accessible terms.')
class-attribute
instance-attribute
¶
respond(messages, context, step_title=None)
async
¶
dbtsl
¶
DbtslAgent
¶
Bases:
Responsible for creating and executing queries against a dbt Semantic Layer to answer user questions about business metrics.
conditions = param.List(default=['Always use this when dbtsl_metaset is available'])
class-attribute
instance-attribute
¶
output_schema = DbtslOutputs
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'response_model': DbtslQueryParams, 'template': PROMPTS_DIR / 'DbtslAgent' / 'main.jinja2'}, 'revise_output': {'response_model': RetrySpec, 'template': PROMPTS_DIR / 'BaseLumenAgent' / 'revise_output.jinja2'}})
class-attribute
instance-attribute
¶
purpose = param.String(default='\n Responsible for displaying data to answer user queries about\n business metrics using dbt Semantic Layers. This agent can compile\n and execute metric queries against a dbt Semantic Layer.')
class-attribute
instance-attribute
¶
requires = param.List(default=['source', 'dbtsl_metaset'], readonly=True)
class-attribute
instance-attribute
¶
source = param.ClassSelector(class_=BaseSQLSource, doc='\n The source associated with the dbt Semantic Layer.')
class-attribute
instance-attribute
¶
user = param.String(default='DBT')
class-attribute
instance-attribute
¶
respond(messages, context, step_title=None)
async
¶
Responds to user messages by generating and executing a dbt Semantic Layer query.
DbtslQueryParams
¶
Bases:
Model for dbtsl.client.query() parameters.
chain_of_thought = Field(description="You are a world-class dbt Semantic Layer expert. Think step by step about\n what metrics are needed, what dimensions to group by, what time granularity\n to use, and any filters that should be applied; if filters are applied, include those\n filtered dimensions in group_by. If there are errors, mention how you'll address the errors.\n ")
class-attribute
instance-attribute
¶
expr_slug = Field(description='Give the query a concise, but descriptive, slug that includes the metrics\n and dimensions used, e.g. monthly_revenue_by_region. The slug must be unique.')
class-attribute
instance-attribute
¶
group_by = Field(default_factory=list, description="A list of dimensions to group by, e.g. ['metric_time__month'], must include dimensions from where.")
class-attribute
instance-attribute
¶
limit = Field(default=None, description='The maximum number of rows to return.')
class-attribute
instance-attribute
¶
metrics = Field(default_factory=list, description="A list of metrics to include in the query, e.g. ['revenue']")
class-attribute
instance-attribute
¶
order_by = Field(default_factory=list, description="A list of columns or expressions to order the results by, e.g. ['metric_time__month']")
class-attribute
instance-attribute
¶
where = Field(default_factory=list, description="A list of conditions to filter the results; dimensions referenced here must also be in group_by, e.g. ['metric_time__month >= date_trunc('month', '2024-09-30'::date)']")
class-attribute
instance-attribute
¶
deck_gl
¶
DeckGL Agent for generating 3D map visualizations.
This agent generates deck.gl visualizations either: - Directly as JSON specs (when code_execution is disabled) - Via PyDeck Python code execution (when code_execution is enabled)
DeckGLAgent
¶
Bases:
Agent for generating DeckGL 3D map visualizations.
Supports two generation modes: - When code_execution is 'disabled': Generate DeckGL JSON specs directly - When code_execution is enabled: Generate PyDeck Python code, execute safely, convert to spec
conditions = param.List(default=['Use for 3D geographic visualizations, map-based data, or when user requests DeckGL/deck.gl', 'Use for large-scale geospatial data with latitude/longitude coordinates', 'Use for hexbin aggregations, heatmaps, or 3D extruded visualizations on maps'])
class-attribute
instance-attribute
¶
default_map_style = 'https://basemaps.cartocdn.com/gl/dark-matter-gl-style/style.json'
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'response_model': DeckGLSpec, 'template': PROMPTS_DIR / 'DeckGLAgent' / 'main.jinja2'}, 'main_pydeck': {'response_model': PyDeckSpec, 'template': PROMPTS_DIR / 'DeckGLAgent' / 'main_pydeck.jinja2'}, 'code_safety': {'response_model': CodeSafetyCheck, 'template': PROMPTS_DIR / 'DeckGLAgent' / 'code_safety.jinja2'}, 'revise_output': {'response_model': RetrySpec, 'template': PROMPTS_DIR / 'DeckGLAgent' / 'revise_output.jinja2'}})
class-attribute
instance-attribute
¶
purpose = param.String(default='Generates DeckGL 3D map visualizations from geographic data.')
class-attribute
instance-attribute
¶
user = param.String(default='DeckGL')
class-attribute
instance-attribute
¶
view_type = DeckGLView
class-attribute
instance-attribute
¶
respond(messages, context, step_title=None)
async
¶
Generate a DeckGL visualization.
DeckGLSpec
¶
Bases:
Response model for DeckGL JSON spec generation (declarative mode).
chain_of_thought = Field(description='Explain your design choices for the 3D visualization:\n - What geographic patterns does this data reveal?\n - Which layer type best represents the data (HexagonLayer, ScatterplotLayer, etc.)?\n - What visual encodings (elevation, color, radius) highlight key insights?\n Keep response to 1-2 sentences.', examples=['The data shows population density clustering around urban centers—HexagonLayer with elevation encoding count will create an intuitive 3D cityscape effect.', 'Point locations with varying magnitudes suggest ScatterplotLayer with radius encoding value and color gradient for intensity.'])
class-attribute
instance-attribute
¶
json_spec = Field(description='A DeckGL JSON specification. Must include:\n - initialViewState with latitude, longitude, zoom, pitch, bearing\n - layers array with @@type for each layer\n Do NOT include \'data\' in layers - it will be injected automatically.\n Use @@= syntax for accessors, e.g. "getPosition": "@@=[longitude, latitude]"\n ')
class-attribute
instance-attribute
¶
PyDeckSpec
¶
Bases:
Response model for PyDeck code generation.
chain_of_thought = Field(default='', description='Explain your design choices for the 3D visualization:\n - What geographic patterns does this data reveal?\n - Which layer type best represents the data?\n - What visual encodings highlight key insights?\n Keep response to 1-2 sentences.', examples=['The data shows CO2 emissions by plant location—HexagonLayer aggregating emissions with elevation shows concentration hotspots.', 'Wind turbine locations with capacity data—ScatterplotLayer with radius proportional to capacity reveals infrastructure distribution.'])
class-attribute
instance-attribute
¶
code = Field(description='Python code that creates a PyDeck visualization.\n Requirements:\n - Import pydeck as `pdk`\n - Data is available as `df` (pandas DataFrame)\n - Column names have spaces replaced with underscores and non-alphanumeric chars removed\n - Must assign final deck to variable `deck`\n - Do NOT call .to_html(), .show(), or any I/O methods\n ```\n ')
class-attribute
instance-attribute
¶
document_list
¶
DocumentListAgent
¶
Bases:
The DocumentListAgent lists all available documents provided by the user.
conditions = param.List(default=['Use when user asks to list or see all available documents', 'NOT when user asks about specific document content'])
class-attribute
instance-attribute
¶
input_schema = DocumentListInputs
class-attribute
instance-attribute
¶
purpose = param.String(default='\n Displays a list of all available documents.')
class-attribute
instance-attribute
¶
applies(context)
async
classmethod
¶
hvplot
¶
hvPlotAgent
¶
Bases:
conditions = param.List(default=['Use for exploratory data analysis, interactive plots, and dynamic filtering', 'Use for quick, iterative data visualization during analysis'])
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'template': PROMPTS_DIR / 'hvPlotAgent' / 'main.jinja2'}})
class-attribute
instance-attribute
¶
purpose = param.String(default='Generates a plot of the data given a user prompt.')
class-attribute
instance-attribute
¶
view_type = hvPlotUIView
class-attribute
instance-attribute
¶
sql
¶
DistinctQuery
¶
Bases:
Universal column analysis with optional pattern matching - handles join keys, categories, date ranges.
column = Field(description='Column to analyze unique values')
class-attribute
instance-attribute
¶
offset = Field(default=0, description='Number of distinct values to skip (0 for initial, 10 for follow-up)')
class-attribute
instance-attribute
¶
pattern = Field(default='', description="Optional pattern to search for (e.g., 'chin' for China/Chinese variations). Leave empty for all distinct values.")
class-attribute
instance-attribute
¶
query = Field(default="SELECT DISTINCT {column} FROM {slug[table]} WHERE {column} ILIKE '%{pattern}%' LIMIT 10 OFFSET {offset}")
class-attribute
instance-attribute
¶
format_result(df)
¶
Format query results for display.
generate_sql(source)
¶
Generate SQL for this query type.
get_description()
¶
Get query description for logging.
model_post_init(__context)
¶
Adjust query template based on whether pattern is provided.
SQLAgent
¶
Bases:
conditions = param.List(default=['Use for querying, filtering, aggregating, or transforming data with SQL', "Use for calculations that require executing SQL (e.g., 'calculate average', 'sum by category')", "Use when user asks to 'show', 'get', 'fetch', 'query', 'find', 'filter', 'calculate', 'aggregate', or 'transform' data", "NOT when user asks to 'explain', 'interpret', 'analyze', 'summarize', or 'comment on' existing data", 'NOT useful if the user is using the same data for plotting'])
class-attribute
instance-attribute
¶
exclusions = param.List(default=['dbtsl_metaset'])
class-attribute
instance-attribute
¶
exploration_enabled = param.Boolean(default=True, allow_refs=True, doc='\n Whether to enable SQL exploration mode. When False, only attempts oneshot SQL generation.')
class-attribute
instance-attribute
¶
input_schema = SQLInputs
class-attribute
instance-attribute
¶
not_with = param.List(default=['DbtslAgent', 'MetadataLookup', 'TableListAgent'])
class-attribute
instance-attribute
¶
output_schema = SQLEditors
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'response_model': make_sql_model, 'template': PROMPTS_DIR / 'SQLAgent' / 'main.jinja2'}, 'select_tables': {'response_model': make_table_selection_model, 'template': PROMPTS_DIR / 'SQLAgent' / 'select_tables.jinja2'}, 'select_discoveries': {'response_model': make_discovery_model, 'template': PROMPTS_DIR / 'SQLAgent' / 'select_discoveries.jinja2'}, 'check_sufficiency': {'response_model': make_discovery_model, 'template': PROMPTS_DIR / 'SQLAgent' / 'check_sufficiency.jinja2'}, 'revise_output': {'response_model': RetrySpec, 'template': PROMPTS_DIR / 'SQLAgent' / 'revise_output.jinja2'}})
class-attribute
instance-attribute
¶
purpose = param.String(default='\n Creates and executes SQL queries to retrieve, filter, aggregate, or transform data.\n Handles table joins, WHERE clauses, GROUP BY, calculations, and other SQL operations.\n Generates new data pipelines from SQL transformations.')
class-attribute
instance-attribute
¶
user = param.String(default='SQL')
class-attribute
instance-attribute
¶
respond(messages, context, step_title=None)
async
¶
Execute SQL generation with table selection, then one-shot attempt, then exploration if needed.
revise(feedback, messages, context, view=None, spec=None, language=None, errors=None, **kwargs)
async
¶
SQLEditors
¶
SQLInputs
¶
SQLQuery
¶
Bases:
A single SQL query with its associated metadata.
query = Field(description="\n One, correct, valid SQL query that answers the user's question;\n should only be one query and do NOT add extraneous comments; no multiple semicolons.\n No limits unless explicitly requested.")
class-attribute
instance-attribute
¶
table_slug = Field(description='\n Provide a unique, descriptive table slug for the SQL expression that clearly indicates the key transformations and source tables involved.\n Include 1 or 2 elements of data lineage in the slug, such as the main transformation and original table names,\n e.g. top_5_athletes_in_2020 or distinct_years_from_wx_table.\n Ensure the slug does not duplicate any existing table names or slugs.\n ')
class-attribute
instance-attribute
¶
SampleQuery
¶
Bases:
See actual data content - reveals format issues and patterns.
TableQuery
¶
Bases:
Wildcard search using native source metadata to discover tables and columns by pattern.
pattern = Field(description="Search pattern (e.g., 'revenue', 'customer')")
class-attribute
instance-attribute
¶
scope = Field(default='both', description="Search scope: 'columns' for column names, 'tables' for table names, 'both' for either")
class-attribute
instance-attribute
¶
format_result(matches)
¶
Format search results for display.
get_description()
¶
Get query description for logging.
search_metadata(source)
¶
Search tables and columns using native source metadata with fuzzy matching.
Returns: List of (table_name, column_name) tuples sorted by match quality. column_name is None for table matches.
TableSelection
¶
Bases:
Select tables relevant to answering the user's query.
reasoning = Field(description='Brief explanation of why these tables are needed to answer the query.')
class-attribute
instance-attribute
¶
tables = Field(description='List of table slugs needed to answer the query. Select only tables that are directly relevant.')
class-attribute
instance-attribute
¶
make_discovery_model(sources)
¶
make_source_table_model(sources)
¶
make_sql_model(sources)
¶
Create a SQL query model with source/table validation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
sources
|
|
List of (source_name, table_name) tuples for tables with full schemas. |
required |
make_table_model(sources)
¶
Create a table model with constrained table choices.
make_table_selection_model(available_tables)
¶
Create a table selection model with constrained table choices.
table_list
¶
TableListAgent
¶
Bases:
The TableListAgent lists all available data and lets the user pick one.
conditions = param.List(default=["Use when user explicitly asks to 'list data', 'show available data', or 'what data do you have'", 'NOT for showing actual data contents, querying, or analyzing data', 'NOT for describing data sources or telling what the data is about'])
class-attribute
instance-attribute
¶
input_schema = TableListInputs
class-attribute
instance-attribute
¶
not_with = param.List(default=['DbtslAgent', 'SQLAgent'])
class-attribute
instance-attribute
¶
purpose = param.String(default='\n Displays a list of all available data & datasets. Not useful for identifying which dataset to use for analysis.')
class-attribute
instance-attribute
¶
applies(context)
async
classmethod
¶
validation
¶
QueryCompletionValidation
¶
Bases:
Validation of whether the executed plan answered the user's query
chain_of_thought = Field(description='Restate intent and results succinctly; then explain your reasoning as to why you will be answering yes or no.')
class-attribute
instance-attribute
¶
correct = Field(description='True if query correctly solves user request, otherwise False.')
class-attribute
instance-attribute
¶
missing_elements = Field(default_factory=list, description="List of specific elements from the user's query that weren't addressed")
class-attribute
instance-attribute
¶
suggestions = Field(default_factory=list, description='Suggestions for additional steps that could complete the query if not fully answered')
class-attribute
instance-attribute
¶
ValidationAgent
¶
Bases:
ValidationAgent focuses solely on validating whether the executed plan fully answered the user's original query. It identifies missing elements and suggests next steps when validation fails.
conditions = param.List(default=['Use to validate whether executed plans fully answered user queries', 'Use to identify missing elements from the original user request', 'NOT for data analysis, pattern identification, or technical programming questions'])
class-attribute
instance-attribute
¶
input_schema = ValidationInputs
class-attribute
instance-attribute
¶
output_schema = ValidationOutputs
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'template': PROMPTS_DIR / 'ValidationAgent' / 'main.jinja2', 'response_model': QueryCompletionValidation, 'tools': []}})
class-attribute
instance-attribute
¶
purpose = param.String(default="\n Validates whether executed plans fully answered the user's original query.\n Identifies missing elements, assesses completeness, and suggests next steps\n when validation fails. Acts as a quality gate for plan execution.")
class-attribute
instance-attribute
¶
respond(messages, context, step_title=None)
async
¶
ValidationInputs
¶
vega_lite
¶
AltairSpec
¶
Bases:
Response model for Altair code generation.
chain_of_thought = Field(default='', description="Explain your design choices based on visualization theory:\n - What story does this data tell?\n - What's the most compelling insight or trend (for the title)?\n - Which visual encodings (position, color, size) best reveal patterns?\n Keep response to 1-2 sentences.", examples=['The data reveals US dominance in Winter Olympic hosting—a horizontal bar chart sorted descending makes comparison immediate, with the leader highlighted in a distinct color.', 'This time series shows a 40% revenue spike in Q3 2024—a line chart with point markers reveals the trend clearly.'])
class-attribute
instance-attribute
¶
code = Field(description="Python code that creates an Altair chart.\n Requirements:\n - Import altair as `alt`\n - Data is available as `df` (pandas DataFrame)\n - Must assign final chart to variable `chart`\n - Do NOT call .to_dict(), .save(), .display() or any I/O methods\n - Use 'container' for width to make charts responsive\n ")
class-attribute
instance-attribute
¶
VegaLiteAgent
¶
Bases:
conditions = param.List(default=['Use for publication-ready visualizations or when user specifically requests Vega-Lite charts', 'Use for polished charts intended for presentation or sharing'])
class-attribute
instance-attribute
¶
prompts = param.Dict(default={'main': {'response_model': VegaLiteSpec, 'template': PROMPTS_DIR / 'VegaLiteAgent' / 'main.jinja2'}, 'main_altair': {'response_model': AltairSpec, 'template': PROMPTS_DIR / 'VegaLiteAgent' / 'main_altair.jinja2'}, 'code_safety': {'response_model': CodeSafetyCheck, 'template': PROMPTS_DIR / 'VegaLiteAgent' / 'code_safety.jinja2'}, 'interaction_polish': {'response_model': VegaLiteSpecUpdate, 'template': PROMPTS_DIR / 'VegaLiteAgent' / 'interaction_polish.jinja2'}, 'annotate_plot': {'response_model': VegaLiteSpecUpdate, 'template': PROMPTS_DIR / 'VegaLiteAgent' / 'annotate_plot.jinja2'}, 'revise_output': {'response_model': RetrySpec, 'template': PROMPTS_DIR / 'VegaLiteAgent' / 'revise_output.jinja2'}})
class-attribute
instance-attribute
¶
purpose = param.String(default='Generates a vega-lite plot specification from the input data pipeline.')
class-attribute
instance-attribute
¶
user = param.String(default='Vega')
class-attribute
instance-attribute
¶
vector_store_path = param.Path(default=None, check_exists=False, doc='\n Path to a custom vector store for storing and retrieving Vega-Lite examples;\n if not provided a default store will be used depending on the LLM--\n OpenAIEmbeddings for OpenAI LLM or NumpyEmbeddings for all others.')
class-attribute
instance-attribute
¶
view_type = VegaLiteView
class-attribute
instance-attribute
¶
annotate(instruction, messages, context, spec)
async
¶
Apply annotations based on user request.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
instruction
|
|
User's description of what to annotate |
required |
messages
|
|
Chat history for context |
required |
context
|
|
Session context |
required |
spec
|
|
The current VegaLite specification (full dict with 'spec' key) |
required |
Returns:
| Type | Description |
|---|---|
|
Updated specification with annotations |
respond(messages, context, step_title=None)
async
¶
Generates a VegaLite visualization using progressive building approach with real-time updates.
revise(feedback, messages, context, view=None, spec=None, language=None, errors=None, **kwargs)
async
¶
VegaLiteSpec
¶
Bases:
chain_of_thought = Field(description="Explain your design choices based on visualization theory:\n - What story does this data tell?\n - What's the most compelling insight or trend (for the title)?\n - What additional context adds value without repeating the title (for the subtitle)?\n - Which visual encodings (position, color, size) best reveal patterns?\n - Should color highlight specific insights or remain neutral?\n - What makes this plot engaging and useful for the user?\n Keep response to 1-2 sentences.", examples=["The data reveals US dominance in Winter Olympic hosting (4 times vs France's 3)—title should emphasize this leadership. Position encoding via horizontal bars sorted descending makes comparison immediate, neutral blue keeps focus on counts rather than categories, and the subtitle can note the 23-country spread to add context without redundancy.", "This time series shows a 40% revenue spike in Q3 2024—the key trend for the title. A line chart with position encoding (time→x, revenue→y) reveals the pattern, endpoint labels eliminate need for constant grid reference making it cleaner, and color remains neutral since there's one series; the subtitle should explain what drove the spike (e.g., 'Three offshore projects') to add insight."])
class-attribute
instance-attribute
¶
yaml_spec = Field(description='A basic vega-lite YAML specification with core plot elements only (mark, basic x/y encoding). Skip $schema and data fields.')
class-attribute
instance-attribute
¶
VegaLiteSpecUpdate
¶
Bases: