Struct DataAgent 

pub struct DataAgent { /* private fields */ }

DataAgent: A specialized agent for data analysis and processing tasks This agent is built on top of the TemplateAgent and provides data-specific functionality

Implementations

impl DataAgent

pub async fn new(_config: Config) -> Result<Self, KowalskiError>

Creates a new DataAgent with the specified configuration

Examples found in repository

examples/csv_analysis.rs (line 17)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Initialize logging
    env_logger::init();

    // Load configuration
    let config = Config::default();
    let mut data_agent = DataAgent::new(config).await?;

    // Start a conversation
    println!("📊 Starting CSV Analysis...");
    let conversation_id = data_agent.start_conversation("llama3.2");
    println!("Data Agent Conversation ID: {}", conversation_id);

    // Set up the role for data analysis
    let role = Role::new(
        "Data Analysis Assistant",
        "You are an expert at analyzing and interpreting data from CSV files.",
    )
    .with_audience(Audience::new(
        "Data Scientist",
        "You are speaking to a data scientist who needs detailed analysis.",
    ))
    .with_preset(Preset::new(
        "Analysis",
        "Provide comprehensive analysis with insights and recommendations.",
    ));

    // Sample CSV data for analysis
    let csv_data = r#"name,age,city,salary,department
John Doe,30,New York,75000,Engineering
Jane Smith,28,San Francisco,85000,Marketing
Bob Johnson,35,Chicago,65000,Sales
Alice Brown,32,Boston,70000,Engineering
Charlie Wilson,29,Seattle,80000,Engineering
Diana Davis,31,Austin,72000,Marketing
Eve Miller,27,Denver,68000,Sales
Frank Garcia,33,Portland,75000,Engineering
Grace Lee,26,Atlanta,65000,Marketing
Henry Taylor,34,Dallas,78000,Engineering"#;

    println!("\n📈 Processing CSV Data:");
    println!("{}", csv_data);

    // Process the CSV data
    let analysis_result = data_agent.process_csv(csv_data).await?;

    println!("\n📊 CSV Analysis Results:");
    println!("Headers: {:?}", analysis_result.headers);
    println!("Total Rows: {}", analysis_result.total_rows);
    println!("Total Columns: {}", analysis_result.total_columns);
    println!(
        "Summary: {}",
        serde_json::to_string_pretty(&analysis_result.summary)?
    );

    // Ask the agent to analyze the data
    let analysis_prompt = format!(
        "Please analyze this CSV data and provide insights:\n\n{}\n\nAnalysis results:\n{}",
        csv_data,
        serde_json::to_string_pretty(&analysis_result.summary)?
    );

    let mut response = data_agent
        .chat_with_history(&conversation_id, &analysis_prompt, Some(role))
        .await?;

    println!("\n🤖 AI Analysis:");

    // Process the streaming response
    let mut buffer = String::new();
    while let Some(chunk) = response.chunk().await? {
        match data_agent
            .process_stream_response(&conversation_id, &chunk)
            .await
        {
            Ok(Some(message)) => {
                // Print the content if it exists
                if !message.content.is_empty() {
                    print!("{}", message.content);
                    io::stdout().flush()?;
                    buffer.push_str(&message.content);
                }

                // Handle tool calls if they exist
                if let Some(tool_calls) = &message.tool_calls {
                    for tool_call in tool_calls {
                        print!("\n[Tool Call] {}(", tool_call.function.name);
                        if let Some(obj) = tool_call.function.arguments.as_object() {
                            for (key, value) in obj {
                                print!("{}: {}, ", key, value);
                            }
                        }
                        println!(")");
                        io::stdout().flush()?;
                    }
                }
            }
            Ok(None) => {
                data_agent
                    .add_message(&conversation_id, "assistant", &buffer)
                    .await;
                println!("\n✅ Analysis complete!\n");
                break;
            }
            Err(e) => {
                eprintln!("\n❌ Error processing stream: {}", e);
                break;
            }
        }
    }

    // Ask a follow-up question about specific insights
    let follow_up = "What are the key insights about salary distribution across departments?";
    let mut follow_up_response = data_agent
        .chat_with_history(&conversation_id, follow_up, None)
        .await?;

    println!("\n🔍 Follow-up Analysis:");
    let mut buffer = String::new();
    while let Some(chunk) = follow_up_response.chunk().await? {
        match data_agent
            .process_stream_response(&conversation_id, &chunk)
            .await
        {
            Ok(Some(message)) => {
                // Print the content if it exists
                if !message.content.is_empty() {
                    print!("{}", message.content);
                    io::stdout().flush()?;
                    buffer.push_str(&message.content);
                }

                // Handle tool calls if they exist
                if let Some(tool_calls) = &message.tool_calls {
                    for tool_call in tool_calls {
                        print!("\n[Tool Call] {}(", tool_call.function.name);
                        if let Some(obj) = tool_call.function.arguments.as_object() {
                            for (key, value) in obj {
                                print!("{}: {}, ", key, value);
                            }
                        }
                        println!(")");
                        io::stdout().flush()?;
                    }
                }
            }
            Ok(None) => {
                data_agent
                    .add_message(&conversation_id, "assistant", &buffer)
                    .await;
                println!("\n");
                break;
            }
            Err(e) => {
                eprintln!("\n❌ Error processing stream: {}", e);
                break;
            }
        }
    }

    Ok(())
}

pub async fn process_csv( &self, csv_content: &str, ) -> Result<CsvAnalysisResult, KowalskiError>

Processes a CSV file

Examples found in repository

examples/csv_analysis.rs (line 55)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Initialize logging
    env_logger::init();

    // Load configuration
    let config = Config::default();
    let mut data_agent = DataAgent::new(config).await?;

    // Start a conversation
    println!("📊 Starting CSV Analysis...");
    let conversation_id = data_agent.start_conversation("llama3.2");
    println!("Data Agent Conversation ID: {}", conversation_id);

    // Set up the role for data analysis
    let role = Role::new(
        "Data Analysis Assistant",
        "You are an expert at analyzing and interpreting data from CSV files.",
    )
    .with_audience(Audience::new(
        "Data Scientist",
        "You are speaking to a data scientist who needs detailed analysis.",
    ))
    .with_preset(Preset::new(
        "Analysis",
        "Provide comprehensive analysis with insights and recommendations.",
    ));

    // Sample CSV data for analysis
    let csv_data = r#"name,age,city,salary,department
John Doe,30,New York,75000,Engineering
Jane Smith,28,San Francisco,85000,Marketing
Bob Johnson,35,Chicago,65000,Sales
Alice Brown,32,Boston,70000,Engineering
Charlie Wilson,29,Seattle,80000,Engineering
Diana Davis,31,Austin,72000,Marketing
Eve Miller,27,Denver,68000,Sales
Frank Garcia,33,Portland,75000,Engineering
Grace Lee,26,Atlanta,65000,Marketing
Henry Taylor,34,Dallas,78000,Engineering"#;

    println!("\n📈 Processing CSV Data:");
    println!("{}", csv_data);

    // Process the CSV data
    let analysis_result = data_agent.process_csv(csv_data).await?;

    println!("\n📊 CSV Analysis Results:");
    println!("Headers: {:?}", analysis_result.headers);
    println!("Total Rows: {}", analysis_result.total_rows);
    println!("Total Columns: {}", analysis_result.total_columns);
    println!(
        "Summary: {}",
        serde_json::to_string_pretty(&analysis_result.summary)?
    );

    // Ask the agent to analyze the data
    let analysis_prompt = format!(
        "Please analyze this CSV data and provide insights:\n\n{}\n\nAnalysis results:\n{}",
        csv_data,
        serde_json::to_string_pretty(&analysis_result.summary)?
    );

    let mut response = data_agent
        .chat_with_history(&conversation_id, &analysis_prompt, Some(role))
        .await?;

    println!("\n🤖 AI Analysis:");

    // Process the streaming response
    let mut buffer = String::new();
    while let Some(chunk) = response.chunk().await? {
        match data_agent
            .process_stream_response(&conversation_id, &chunk)
            .await
        {
            Ok(Some(message)) => {
                // Print the content if it exists
                if !message.content.is_empty() {
                    print!("{}", message.content);
                    io::stdout().flush()?;
                    buffer.push_str(&message.content);
                }

                // Handle tool calls if they exist
                if let Some(tool_calls) = &message.tool_calls {
                    for tool_call in tool_calls {
                        print!("\n[Tool Call] {}(", tool_call.function.name);
                        if let Some(obj) = tool_call.function.arguments.as_object() {
                            for (key, value) in obj {
                                print!("{}: {}, ", key, value);
                            }
                        }
                        println!(")");
                        io::stdout().flush()?;
                    }
                }
            }
            Ok(None) => {
                data_agent
                    .add_message(&conversation_id, "assistant", &buffer)
                    .await;
                println!("\n✅ Analysis complete!\n");
                break;
            }
            Err(e) => {
                eprintln!("\n❌ Error processing stream: {}", e);
                break;
            }
        }
    }

    // Ask a follow-up question about specific insights
    let follow_up = "What are the key insights about salary distribution across departments?";
    let mut follow_up_response = data_agent
        .chat_with_history(&conversation_id, follow_up, None)
        .await?;

    println!("\n🔍 Follow-up Analysis:");
    let mut buffer = String::new();
    while let Some(chunk) = follow_up_response.chunk().await? {
        match data_agent
            .process_stream_response(&conversation_id, &chunk)
            .await
        {
            Ok(Some(message)) => {
                // Print the content if it exists
                if !message.content.is_empty() {
                    print!("{}", message.content);
                    io::stdout().flush()?;
                    buffer.push_str(&message.content);
                }

                // Handle tool calls if they exist
                if let Some(tool_calls) = &message.tool_calls {
                    for tool_call in tool_calls {
                        print!("\n[Tool Call] {}(", tool_call.function.name);
                        if let Some(obj) = tool_call.function.arguments.as_object() {
                            for (key, value) in obj {
                                print!("{}: {}, ", key, value);
                            }
                        }
                        println!(")");
                        io::stdout().flush()?;
                    }
                }
            }
            Ok(None) => {
                data_agent
                    .add_message(&conversation_id, "assistant", &buffer)
                    .await;
                println!("\n");
                break;
            }
            Err(e) => {
                eprintln!("\n❌ Error processing stream: {}", e);
                break;
            }
        }
    }

    Ok(())
}

pub async fn analyze_data( &self, csv_content: &str, ) -> Result<Value, KowalskiError>

Analyzes data statistics

Trait Implementations

impl Agent for DataAgent

fn new<'async_trait>( config: Config, ) -> Pin<Box<dyn Future<Output = Result<Self, KowalskiError>> + Send + 'async_trait>>

where Self: 'async_trait,

Creates a new agent with the specified configuration.

fn start_conversation(&mut self, model: &str) -> String

Starts a new conversation

fn get_conversation(&self, id: &str) -> Option<&Conversation>

Gets a conversation by ID

fn list_conversations(&self) -> Vec<&Conversation>

Lists all conversations

fn delete_conversation(&mut self, id: &str) -> bool

Deletes a conversation

fn chat_with_history<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, conversation_id: &'life1 str, content: &'life2 str, role: Option<Role>, ) -> Pin<Box<dyn Future<Output = Result<Response, KowalskiError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Chats with history

fn process_stream_response<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, conversation_id: &'life1 str, chunk: &'life2 [u8], ) -> Pin<Box<dyn Future<Output = Result<Option<Message>, KowalskiError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Processes a stream response

fn add_message<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 mut self, conversation_id: &'life1 str, role: &'life2 str, content: &'life3 str, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Adds a message to a conversation

fn name(&self) -> &str

Gets the agent’s name

fn description(&self) -> &str

Gets the agent’s description