Final Project: AI-Powered Interactive System

💻 Implementation Guide

🚀 Core Agent System Example

from crewai import Agent, Task, Crew, Process
from pydantic import BaseModel, Field
import asyncio

class ResponseData(BaseModel):
    """Schema for agent responses."""
    content: str = Field(..., description="The main response content")
    confidence: float = Field(..., description="Confidence score (0-1)")
    suggestions: list = Field(default=[], description="Follow-up suggestions")

# Initialize agents
main_agent = Agent(
    role="Main Coordinator",
    goal="Coordinate user interactions and manage workflow",
    backstory="Expert in managing complex workflows and user interactions",
    tools=[search_tool],
    verbose=True,
)

specialist_agent = Agent(
    role="Domain Specialist",
    goal="Provide expert knowledge in the specific domain",
    backstory="Deep expert in the chosen domain with years of experience",
    verbose=True,
)

evaluator_agent = Agent(
    role="Quality Evaluator",
    goal="Evaluate and improve response quality",
    backstory="Expert in quality assurance and response optimization",
    verbose=True,
)

# Task creation functions
def create_analysis_task(user_input: str) -> Task:
    return Task(
        description=f"Analyze the user request: {user_input}",
        expected_output="Initial analysis and response plan",
        agent=main_agent,
    )

def create_specialist_task(context: str) -> Task:
    return Task(
        description=f"Provide expert analysis for: {context}",
        expected_output="Detailed expert response",
        output_pydantic=ResponseData,
        agent=specialist_agent,
    )

def create_evaluation_task(response: str) -> Task:
    return Task(
        description=f"Evaluate and improve this response: {response}",
        expected_output="Improved response with quality assessment",
        agent=evaluator_agent,
    )

# Main workflow function
async def process_user_request(user_input: str):
    # Create and execute the crew
    crew = Crew(
        agents=[main_agent, specialist_agent, evaluator_agent],
        tasks=[
            create_analysis_task(user_input),
            create_specialist_task(user_input),
            create_evaluation_task("placeholder"),
        ],
        process=Process.sequential,
        verbose=True,
    )
    
    result = await crew.kickoff_async()
    return result

💬 Streamlit Chat Interface with Voice Input

import streamlit as st
import asyncio
import whisper
import tempfile
import os
from streamlit_mic_recorder import mic_recorder
from your_agent_system import process_user_request

st.title("🤖 AI Interactive System")

# Initialize session state
if "messages" not in st.session_state:
    st.session_state.messages = []
    st.session_state.is_processing = False

# Display chat messages
for msg in st.session_state.messages:
    with st.chat_message(msg["role"]):
        st.markdown(msg["content"])

# Input method selection
input_method = st.radio("Choose input method:", ["Text", "Voice"], horizontal=True)

user_input = None

def convert_speech_to_text(audio_bytes):
    """Convert speech to text using OpenAI Whisper."""
    try:
        with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as temp_audio:
            temp_audio.write(audio_bytes)
            temp_audio_path = temp_audio.name

        try:
            # Load Whisper model
            model = whisper.load_model("base")
            result = model.transcribe(temp_audio_path)
            return result["text"]
        finally:
            os.unlink(temp_audio_path)
    except Exception as e:
        st.error(f"Speech recognition error: {str(e)}")
        return None

# Get user input
if input_method == "Text":
    user_input = st.chat_input("Type your message...")
else:
    st.write("Click the microphone to record:")
    audio = mic_recorder(
        start_prompt="🎤 Start recording",
        stop_prompt="⏹️ Stop recording",
        just_once=True,
        use_container_width=True,
    )
    
    if audio:
        with st.spinner("Converting speech to text..."):
            user_input = convert_speech_to_text(audio["bytes"])
            if user_input:
                st.success(f"Recognized: {user_input}")
            else:
                st.error("Could not recognize speech. Please try again.")

# Process user input
if user_input and not st.session_state.is_processing:
    st.session_state.messages.append({"role": "user", "content": user_input})
    st.session_state.is_processing = True
    
    with st.chat_message("assistant"):
        with st.spinner("🤖 Processing your request..."):
            try:
                response = asyncio.run(process_user_request(user_input))
                st.session_state.messages.append(
                    {"role": "assistant", "content": response}
                )
            except Exception as e:
                st.error(f"Error: {str(e)}")
            finally:
                st.session_state.is_processing = False
    
    st.rerun()

# Sidebar configuration
with st.sidebar:
    st.header("Configuration")
    st.selectbox("Agent Mode", ["Standard", "Expert", "Fast"])
    st.slider("Response Length", 1, 10, 5)
    
    if st.button("Clear Chat"):
        st.session_state.messages = []
        st.rerun()

📊 Session Management

class SessionManager:
    def __init__(self):
        self.conversation_history = []
        self.user_preferences = {}
        self.context = {}
    
    def add_message(self, role: str, content: str):
        self.conversation_history.append({
            "role": role,
            "content": content,
            "timestamp": datetime.now()
        })
    
    def get_context(self) -> str:
        """Get conversation context for agents."""
        recent_messages = self.conversation_history[-5:]  # Last 5 messages
        return "\n".join([f"{msg['role']}: {msg['content']}" for msg in recent_messages])
    
    def update_preferences(self, key: str, value: any):
        self.user_preferences[key] = value
    
    def clear_history(self):
        self.conversation_history = []

# Usage in Streamlit
if "session_manager" not in st.session_state:
    st.session_state.session_manager = SessionManager()