Building Lumen Apps¶

import lumen.ai as lmai

ui = lmai.ExplorerUI(data='data.csv')
ui.servable()

import lumen.ai as lmai
from lumen.ai.agents import SourceAgent
from lumen.ai.analyses import Analysis
from lumen.ai.controls import CodeSourceControls, UploadSourceControls
from lumen.ai.coordinator import Planner
from lumen.ai.embeddings import OpenAIEmbeddings
from lumen.ai.vector_store import DuckDBVectorStore

# Custom tool
def calculate_growth(current: float, previous: float) -> float:
    """Calculate percentage growth between two values."""
    return ((current - previous) / previous) * 100 if previous else 0.0

# Custom analysis
class CorrelationAnalysis(Analysis):
    async def analyze(self, pipeline, **kwargs):
        return pipeline.data.select_dtypes(include='number').corr()

# Custom data fetcher
def fetch_sales(region: str = "US", year: int = 2024):
    """Fetch sales data."""
    return pd.read_csv(f"sales_{region}_{year}.csv")

# LLM with per-agent model overrides
llm = lmai.llm.OpenAI(
    model_kwargs={
        "default": {"model": "gpt-4.1-mini", "temperature": 0.2},
        "sql": {"model": "gpt-4.1", "temperature": 0.1},
    }
)

# Semantic search (optional)
vector_store = DuckDBVectorStore(uri='embeddings.db', embeddings=OpenAIEmbeddings())

ui = lmai.ExplorerUI(
    # Data sources (file paths, Source, or Pipeline objects)
    data=['customers.csv', 'orders.csv'],

    # LLM configuration
    llm=llm,

    # Agents (extend default agents: TableListAgent, ChatAgent, SQLAgent, etc.)
    agents=[SourceAgent()],

    # Source controls for fetching external data
    source_controls=[
        CodeSourceControls(functions={"Fetch Sales": fetch_sales}),
        UploadSourceControls(),
    ],

    # Custom tools the LLM can invoke
    tools=[calculate_growth],

    # Custom analyses for the AnalysisAgent
    analyses=[CorrelationAnalysis],

    # Coordinator for planning responses
    coordinator=Planner,  # Mostly unnecessary unless you want to customize the planning algorithm
    coordinator_params={},

    # Vector stores for semantic search
    vector_store=vector_store,
    document_vector_store=None,  # Separate store for document search, else uses same vector store

    # UI customization
    title='Sales Analytics',
    suggestions=[
        ('search', 'What data is available?'),
        ('bar_chart', 'Show sales by region'),
    ],  # Predefined suggestions for users to click on

    # Code execution: 'hidden', 'disabled', 'prompt', 'llm', 'allow'
    code_execution='prompt',

    # Export configuration
    notebook_preamble='# Generated by Sales Analytics App',
    export_functions={},

    # Logging
    log_level='INFO',                    # 'DEBUG', 'INFO', 'WARNING', 'ERROR'
    logfire_tags=['sales', 'analytics'],  # Tags for logfire tracing
    logs_db_path=None,                   # Path to store LLM message logs

    # File upload settings
    filedropper_kwargs={'max_file_size': '100MB'},
    upload_handlers={},  # Custom handlers by file extension

    # Page configuration (panel-material-ui Page component)
    page_config={},
)
ui.servable()

llm = lmai.llm.OpenAI(
    model_kwargs={
        "default": {"model": "gpt-4.1-mini", "temperature": 0.2},
        "sql": {"model": "gpt-4.1", "temperature": 0.1},
        "vega_lite": {"model": "gpt-4.1"},
    }
)

from lumen.ai.controls import UploadSourceControls

ui = lmai.ExplorerUI(source_controls=[UploadSourceControls()])

import param
from lumen.ai.agents import SourceAgent
from lumen.ai.controls import CodeSourceControls, UploadSourceControls

def fetch_sales(region: str = "US", year: int = 2024):
    """Fetch sales data for a region and year."""
    return pd.read_csv(f"sales_{region}_{year}.csv")

ui = lmai.ExplorerUI(
    agents=[SourceAgent()],
    source_controls=[
        CodeSourceControls(
            functions={"Fetch Sales": fetch_sales},
            param_overrides={
                "Fetch Sales": {
                    "region": param.Selector(default="US", objects=["US", "EU", "APAC"]),
                    "year": param.Selector(default=2024, objects=[2024, 2023, 2022]),
                },
            },
        ),
        UploadSourceControls(),
    ],
)

from lumen.ai.agents import Agent

class SummaryAgent(Agent):
    purpose = "Creates executive summaries of data analysis results"

    async def respond(self, messages, context, **kwargs):
        system = await self._render_prompt("main", messages, context)
        response = await self.llm.invoke(messages, system=system)
        return [response], {"summary": str(response)}

ui = lmai.ExplorerUI(data='data.csv', agents=[SummaryAgent()])

def calculate_growth(current: float, previous: float) -> float:
    """Calculate percentage growth between two values."""
    if previous == 0:
        return 0.0
    return ((current - previous) / previous) * 100

ui = lmai.ExplorerUI(data='data.csv', tools=[calculate_growth])

from lumen.ai.analyses import Analysis

class CorrelationAnalysis(Analysis):
    async def analyze(self, pipeline, **kwargs):
        df = pipeline.data
        return df.select_dtypes(include='number').corr()

ui = lmai.ExplorerUI(data='data.csv', analyses=[CorrelationAnalysis])

from lumen.ai.agents import SQLAgent

sql_agent = SQLAgent(
    template_overrides={
        "main": {"instructions": "{{ super() }}\nAlways use explicit JOIN syntax."}
    }
)

ui = lmai.ExplorerUI(data='data.csv', agents=[sql_agent])

from lumen.ai.embeddings import OpenAIEmbeddings
from lumen.ai.vector_store import DuckDBVectorStore

vector_store = DuckDBVectorStore(uri='embeddings.db', embeddings=OpenAIEmbeddings())

ui = lmai.ExplorerUI(data='data.csv', vector_store=vector_store)

ui = lmai.ExplorerUI(
    data='data.csv',
    code_execution='prompt',  # Options: 'hidden', 'disabled', 'prompt', 'llm', 'allow'
)