updated for deployment

.github/workflows/main.yaml

@@ -0,0 +1,51 @@
+name: Sync to Hugging Face Space
+
+on:
+  push:
+    branches:
+      - main
+      - master
+  workflow_dispatch:
+
+jobs:
+  sync-to-hub:
+    runs-on: ubuntu-latest
+    
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+          lfs: true
+
+      - name: Setup Git LFS
+        run: |
+          git lfs install
+          git lfs pull
+          git lfs checkout
+
+      - name: Configure Git
+        run: |
+          git config --global user.email "github-actions[bot]@users.noreply.github.com"
+          git config --global user.name "github-actions[bot]"
+
+      - name: Push to Hugging Face Space
+        env:
+          HF_TOKEN: ${{ secrets.HF_TOKEN }}
+          HF_USERNAME: nivakaran
+          HF_SPACE: sparrowagenticai
+        run: |
+          git branch -M main
+          git push https://$HF_USERNAME:$HF_TOKEN@huggingface.co/spaces/$HF_USERNAME/$HF_SPACE main --force
+          if [ $? -ne 0 ]; then 
+            echo "Push failed - check token and permissions"
+            exit 1
+          fi
+
+      - name: Verify Sync
+        if: success()
+        run: echo "successfully synced to Hugging Face Space!"
+
+      - name: Sync Failed
+        if: failure()
+        run: echo "Failed to sync to Hugging Face Space. Check logs above."

.gitignore

@@ -0,0 +1,12 @@
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+
+# Virtual environments
+.venv
+
+.env

.python-version

@@ -0,0 +1 @@
+3.11

Dockerfile

@@ -0,0 +1,28 @@
+# Read the doc: https://huggingface.co/docs/hub/spaces-sdks-docker
+FROM python:3.11
+
+# Create non-root user
+RUN useradd -m -u 1000 user
+USER user
+
+# Set environment variables
+ENV PATH="/home/user/.local/bin:$PATH"
+ENV PYTHONUNBUFFERED=1
+
+# Set working directory
+WORKDIR /app
+
+# Copy requirements first for better caching
+COPY --chown=user ./requirements.txt requirements.txt
+
+# Install dependencies
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
+
+# Copy application code
+COPY --chown=user . /app
+
+# Expose port 7860 (required by Hugging Face Spaces)
+EXPOSE 7860
+
+# Run the application
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -0,0 +1,88 @@
+# Sparrow Agent Chatbot - Parcel Consolidation And Tracking Agentic AI System
+
+## Overview
+Welcome to Sparrow, a cutting-edge agentic AI chatbot engineered with LangGraph and powered by Groq's high-performance LLM. Designed for a parcel consolidation and tracking platform, Sparrow exemplifies scalable, intelligent automation, delivering seamless user assistance through sophisticated natural language processing and dynamic workflow orchestration. This project is a testament to modern AI engineering, blending modularity, efficiency, and extensibility.
+
+## Key Features
+- **Clarify Agent**: Analyzes and refines user queries into precise, summarized inputs for downstream agents.
+- **Worker Agent**: Executes tasks with a robust toolkit including parcel tracking, user information retrieval, ETA estimation, and a "think" capability for complex reasoning (future-proof for additional tools).
+- **Master Agent**: Orchestrates parallel worker tasks dynamically, synthesizing outputs into cohesive responses.
+- **Final Graph**: Integrates the entire workflow into a streamlined, end-to-end solution.
+
+## Technical Highlights
+- **Framework & Libraries**: Built with Python, Flask, LangChain, and GroqLLM for state-of-the-art AI performance.
+- **Graph Technology**: Leverages LangGraph for a modular, graph-based agent architecture.
+- **Asynchronous Processing**: Utilizes async programming for high-efficiency task handling.
+- **Code Quality**: Maintains a clean, well-documented codebase with modular file structures (states, nodes, graph builder).
+- **Scalability**: Designed to handle increasing loads with dynamic worker allocation.
+
+## System Architecture
+The Sparrow system features a sophisticated workflow, visualized through the following diagrams:
+
+![Workflow 1 - Clarify User Query Flow](./assets/Screenshot%202025-08-26%20034929.png)
+![Workflow 3 - Worker Agent Subgraph Flow](./assets/Screenshot%202025-08-26%20033447.png)
+![Workflow 4 - Master Agent Subgraph Flow](./assets/Screenshot%202025-08-26%20033458.png)
+![Workflow 5 - Final Graph Flow](./assets/Screenshot%202025-08-26%20033424.png)
+
+These graphs illustrate the journey from query intake to final response, showcasing agent-tool interactions and parallel processing.
+
+## Project Structure
+The repository is organized for clarity and scalability:
+
+```
+sparrow-agent/
+├── src/
+│   ├── graphs/
+│   │   ├── __init__.py
+│   │   ├── queryGraph.py         # Query clarification logic
+│   │   ├── actionGraph.py         # Worker task execution with tools
+│   │   ├── masterGraph.py         # Orchestration and output synthesis
+│   │   ├── finalAgentGraph.py     # End-to-end workflow integration
+│   ├── states/
+│   │   ├── __init__.py
+│   │   ├── conversationState.py   # Manages conversation thread states
+│   │   ├── taskState.py           # Tracks task execution states
+│   ├── llms/
+│   │   ├── __init__.py
+│   │   ├── groqllm.py   # Manages conversation thread states
+│   ├── nodes/
+│   │   ├── __init__.py
+│   │   ├── actionNode.py            # Tool execution node
+│   │   ├── masterNode.py        # Response compression node
+│   │   ├── queryNode.py        
+│   ├── utils/
+│   │   ├── __init__.py
+│   │   ├── actionState.py              # Custom logging setup
+│   │   ├── masterState.py              
+│   │   ├── queryState.py            
+├── app.py                         # Main Flask application entry point
+├── requirements.txt               # Dependency list
+├── templates/
+│   ├── index.html                 # Chat interface template
+├── README.md                      # This file
+
+```
+
+## Setup & Installation
+1. **Clone the Repository**: `git clone https://github.com/Nivakaran-S/sparrow-agent.git`
+2. **Install Dependencies**: `pip install -r requirements.txt` (includes Flask, LangChain, LangGraph, GroqLLM, etc.)
+3. **Configure Environment**: Set `export FLASK_SECRET_KEY='your-secret-key'` and API keys for Groq.
+4. **Run the Application**: `python app.py` (defaults to port 5000, adjustable via `PORT` env variable).
+
+## Usage
+- **Chat Interface**: Access at `http://localhost:5000`.
+- **API Endpoints**:
+  - `/chat` (POST): Send messages with JSON `{ "message": "your query" }`.
+  - `/new_conversation` (POST): Reset to a new thread.
+  - `/health` (GET): Check server status.
+- **Interaction**: Real-time responses powered by GroqLLM and agent workflows.
+
+## Contributions
+- **Enhancements**: Add new tools to `workerAgent.py` or optimize graph logic.
+- **Performance**: Improve async handling or worker scalability.
+- **UI/UX**: Upgrade `index.html`
+
+
+## Acknowledgements
+- Powered by Groq for high-speed LLM inference.
+- Built with LangGraph for advanced agent orchestration.

app.py

@@ -0,0 +1,184 @@
+from flask import Flask, request, jsonify, render_template, session
+import uuid
+import logging
+from datetime import datetime
+import os
+import threading
+import time
+from src.graphs.finalAgentGraph import sparrowAgent
+from langchain_core.messages import HumanMessage, AIMessage
+
+app = Flask(__name__)
+app.secret_key = os.environ.get('FLASK_SECRET_KEY', 'your-secret-key-here')
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+conversations = {}
+conversations_lock = threading.Lock()
+
+
+def ensure_langchain_message(message):
+    """Ensure a message is a proper LangChain message object"""
+    if isinstance(message, (HumanMessage, AIMessage)):
+        return message
+    elif isinstance(message, dict):
+        content = message.get('content', str(message))
+        message_type = message.get('type', 'ai')
+        if message_type == 'human':
+            return HumanMessage(content=content)
+        else:
+            return AIMessage(content=content)
+    elif isinstance(message, str):
+        return AIMessage(content=message)
+    else:
+        return AIMessage(content=str(message))
+
+
+def clean_messages_list(messages):
+    """Ensure all messages are valid LangChain message objects"""
+    return [ensure_langchain_message(msg) for msg in messages]
+
+
+@app.route('/')
+def index():
+    """Serve main chat interface"""
+    return "<h1>Sparrow Agent API</h1><p>POST to <code>/chat</code> to talk with the agent.</p>"
+
+
+@app.route('/chat', methods=['POST'])
+def chat():
+    """Handle chat messages"""
+    try:
+        data = request.get_json(force=True)
+        user_message = data.get('message', '').strip()
+
+        if not user_message:
+            return jsonify({'success': False, 'error': 'Empty message'})
+
+        thread_id = session.get('thread_id')
+        if not thread_id:
+            thread_id = str(uuid.uuid4())
+            session['thread_id'] = thread_id
+
+        with conversations_lock:
+            if thread_id not in conversations:
+                conversations[thread_id] = {
+                    'messages': [],
+                    'notes': [],
+                    'query_brief': '',
+                    'final_message': '',
+                    'created_at': datetime.now(),
+                    'last_updated': datetime.now()
+                }
+            conversation = conversations[thread_id]
+
+        human_message = HumanMessage(content=user_message)
+        conversation['messages'].append(human_message)
+        conversation['last_updated'] = datetime.now()
+
+        cleaned_messages = clean_messages_list(conversation['messages'])
+
+        sparrow_input = {
+            'messages': cleaned_messages,
+            'notes': conversation.get('notes', []),
+            'query_brief': conversation.get('query_brief', ''),
+            'final_message': conversation.get('final_message', '')
+        }
+
+        logger.info(f"[{thread_id}] Processing message: {user_message[:100]}")
+
+        result = sparrowAgent.invoke(sparrow_input)
+
+        response_message = ""
+        ai_message = None
+
+        if result.get('final_message'):
+            response_message = result['final_message']
+            ai_message = AIMessage(content=response_message)
+        else:
+            result_messages = clean_messages_list(result.get('messages', []))
+            last_user_index = max(
+                (i for i, msg in enumerate(result_messages) if isinstance(msg, HumanMessage)),
+                default=-1
+            )
+            for i in range(last_user_index + 1, len(result_messages)):
+                msg = result_messages[i]
+                if isinstance(msg, AIMessage) and msg.content.strip():
+                    response_message = msg.content
+                    ai_message = msg
+                    break
+
+        if not response_message:
+            response_message = "I'm processing your request. Could you provide more details?"
+            ai_message = AIMessage(content=response_message)
+
+        status_info = ""
+        if result.get('execution_jobs'):
+            status_info = f"Executed: {', '.join(result['execution_jobs'])}"
+        elif result.get('notes') and isinstance(result['notes'], list) and result['notes']:
+            status_info = str(result['notes'][-1])
+
+        with conversations_lock:
+            if result.get('messages'):
+                conversation['messages'] = clean_messages_list(result['messages'])
+            else:
+                conversation['messages'].append(ai_message)
+
+            # Deduplicate
+            seen = set()
+            unique_messages = []
+            for msg in conversation['messages']:
+                key = (type(msg).__name__, getattr(msg, "content", str(msg)))
+                if key not in seen:
+                    seen.add(key)
+                    unique_messages.append(msg)
+            conversation['messages'] = unique_messages
+
+            conversation['notes'] = result.get('notes', conversation['notes'])
+            conversation['query_brief'] = result.get('query_brief', conversation['query_brief'])
+            conversation['final_message'] = result.get('final_message', conversation['final_message'])
+            conversation['last_updated'] = datetime.now()
+
+        logger.info(f"[{thread_id}] Response generated: {response_message[:100]}")
+
+        return jsonify({
+            'success': True,
+            'response': response_message,
+            'status': status_info,
+            'thread_id': thread_id
+        })
+
+    except Exception as e:
+        logger.exception("Error in /chat")
+        return jsonify({'success': False, 'error': str(e)})
+
+
+@app.route('/health')
+def health():
+    """Health check endpoint"""
+    with conversations_lock:
+        active = len(conversations)
+    return jsonify({'status': 'healthy', 'active_conversations': active, 'timestamp': datetime.now().isoformat()})
+
+
+def cleanup_conversations():
+    """Remove old conversations older than 24 hours"""
+    while True:
+        time.sleep(3600)
+        cutoff = datetime.now().timestamp() - 24 * 3600
+        with conversations_lock:
+            old = [tid for tid, conv in conversations.items() if conv['last_updated'].timestamp() < cutoff]
+            for tid in old:
+                del conversations[tid]
+            if old:
+                logger.info(f"Cleaned up {len(old)} old conversations")
+
+
+if __name__ == '__main__':
+    cleanup_thread = threading.Thread(target=cleanup_conversations, daemon=True)
+    cleanup_thread.start()
+
+    port = int(os.environ.get("PORT", 7860))  # ✅ HF expects 7860
+    logger.info(f"Starting Sparrow Agent Flask app on port {port}")
+    app.run(host="0.0.0.0", port=port)

app2.py

@@ -0,0 +1,276 @@
+import gradio as gr
+import uuid
+import logging
+from datetime import datetime
+import os
+
+from src.graphs.finalAgentGraph import sparrowAgent
+from langchain_core.messages import HumanMessage, AIMessage
+
+# Setup logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+def ensure_langchain_message(message):
+    """Ensure a message is a proper LangChain message object"""
+    if isinstance(message, (HumanMessage, AIMessage)):
+        return message
+    elif isinstance(message, dict):
+        content = message.get('content', str(message))
+        message_type = message.get('type', 'ai')
+        if message_type == 'human':
+            return HumanMessage(content=content)
+        else:
+            return AIMessage(content=content)
+    elif isinstance(message, str):
+        return AIMessage(content=message)
+    else:
+        return AIMessage(content=str(message))
+
+
+def clean_messages_list(messages):
+    """Clean and ensure all messages in list are proper LangChain message objects"""
+    cleaned_messages = []
+    for msg in messages:
+        cleaned_msg = ensure_langchain_message(msg)
+        cleaned_messages.append(cleaned_msg)
+    return cleaned_messages
+
+
+def initialize_conversation():
+    """Initialize a new conversation state"""
+    return {
+        'thread_id': str(uuid.uuid4()),
+        'messages': [],
+        'notes': [],
+        'query_brief': '',
+        'final_message': '',
+        'created_at': datetime.now(),
+        'last_updated': datetime.now()
+    }
+
+
+def process_message(user_message, history, conversation_state):
+    """
+    Process user message and return response
+    
+    Args:
+        user_message: The user's input message
+        history: Gradio chat history (list of [user_msg, bot_msg] pairs)
+        conversation_state: Dictionary containing conversation context
+    
+    Returns:
+        Tuple of (empty string, updated history, updated conversation state, status message)
+    """
+    try:
+        if not user_message or not user_message.strip():
+            return "", history, conversation_state, "Please enter a message"
+        
+        # Initialize conversation state if None
+        if conversation_state is None:
+            conversation_state = initialize_conversation()
+        
+        thread_id = conversation_state['thread_id']
+        
+        # Add user message to conversation
+        human_message = HumanMessage(content=user_message)
+        conversation_state['messages'].append(human_message)
+        conversation_state['last_updated'] = datetime.now()
+        
+        # Clean messages
+        cleaned_messages = clean_messages_list(conversation_state['messages'])
+        
+        # Prepare input for sparrow agent
+        sparrow_input = {
+            'messages': cleaned_messages,
+            'notes': conversation_state.get('notes', []),
+            'query_brief': conversation_state.get('query_brief', ''),
+            'final_message': conversation_state.get('final_message', '')
+        }
+        
+        logger.info(f"[{thread_id}] Processing message: {user_message[:100]}")
+        logger.info(f"[{thread_id}] Input messages count: {len(cleaned_messages)}")
+        
+        # Invoke the sparrow agent
+        result = sparrowAgent.invoke(sparrow_input)
+        
+        # Extract response message
+        response_message = ""
+        ai_message = None
+        
+        if result.get('final_message'):
+            response_message = result['final_message']
+            ai_message = AIMessage(content=response_message)
+        else:
+            result_messages = clean_messages_list(result.get('messages', []))
+            
+            # Find last user message index
+            last_user_index = -1
+            for i, msg in enumerate(result_messages):
+                if isinstance(msg, HumanMessage):
+                    last_user_index = i
+            
+            # Get first AI message after last user message
+            for i in range(last_user_index + 1, len(result_messages)):
+                msg = result_messages[i]
+                if isinstance(msg, AIMessage) and msg.content and msg.content.strip():
+                    response_message = msg.content
+                    ai_message = msg
+                    break
+        
+        if not response_message:
+            response_message = "I'm processing your request. Could you provide more details?"
+            ai_message = AIMessage(content=response_message)
+        
+        # Update conversation state
+        if result.get('messages'):
+            conversation_state['messages'] = clean_messages_list(result['messages'])
+        else:
+            conversation_state['messages'].append(ai_message)
+        
+        # Remove consecutive duplicates
+        cleaned_conversation_messages = []
+        prev_content = None
+        prev_type = None
+        
+        for msg in conversation_state['messages']:
+            current_content = msg.content if hasattr(msg, 'content') else str(msg)
+            current_type = type(msg).__name__
+            
+            if current_content != prev_content or current_type != prev_type:
+                cleaned_conversation_messages.append(msg)
+                prev_content = current_content
+                prev_type = current_type
+        
+        conversation_state['messages'] = cleaned_conversation_messages
+        conversation_state['notes'] = result.get('notes', conversation_state.get('notes', []))
+        conversation_state['query_brief'] = result.get('query_brief', conversation_state.get('query_brief', ''))
+        conversation_state['final_message'] = result.get('final_message', conversation_state.get('final_message', ''))
+        conversation_state['last_updated'] = datetime.now()
+        
+        # Update Gradio chat history
+        history.append([user_message, response_message])
+        
+        # Create status message
+        status_info = f"Thread: {thread_id[:8]}... | Messages: {len(conversation_state['messages'])}"
+        if result.get('execution_jobs'):
+            status_info += f" | Executed: {', '.join(result['execution_jobs'])}"
+        elif result.get('notes') and isinstance(result['notes'], list) and result['notes']:
+            status_info += f" | Note: {str(result['notes'][-1])[:50]}"
+        
+        logger.info(f"[{thread_id}] Response generated: {response_message[:100]}")
+        logger.info(f"[{thread_id}] Final messages count: {len(conversation_state['messages'])}")
+        
+        return "", history, conversation_state, status_info
+    
+    except Exception as e:
+        logger.error(f"Error processing message: {str(e)}", exc_info=True)
+        error_msg = f"An error occurred: {str(e)}"
+        history.append([user_message, error_msg])
+        return "", history, conversation_state, f"Error: {str(e)}"
+
+
+def clear_conversation():
+    """Clear conversation and start fresh"""
+    new_state = initialize_conversation()
+    logger.info(f"[{new_state['thread_id']}] New conversation started")
+    return [], new_state, f"New conversation started (ID: {new_state['thread_id'][:8]}...)"
+
+
+def get_conversation_info(conversation_state):
+    """Get current conversation information"""
+    if conversation_state is None:
+        return "No active conversation"
+    
+    info_lines = [
+        f"**Thread ID:** {conversation_state['thread_id']}",
+        f"**Messages:** {len(conversation_state.get('messages', []))}",
+        f"**Notes:** {len(conversation_state.get('notes', []))}",
+        f"**Has Query Brief:** {bool(conversation_state.get('query_brief'))}",
+        f"**Has Final Message:** {bool(conversation_state.get('final_message'))}",
+        f"**Created:** {conversation_state.get('created_at', 'N/A')}",
+        f"**Last Updated:** {conversation_state.get('last_updated', 'N/A')}"
+    ]
+    
+    return "\n\n".join(info_lines)
+
+
+# Create Gradio interface
+with gr.Blocks(title="Sparrow Agent Chat", theme=gr.themes.Soft()) as demo:
+    gr.Markdown("# 🦜 Sparrow Agent Chat")
+    gr.Markdown("Interact with the Sparrow AI Agent. Ask questions and get intelligent responses!")
+    
+    # State to store conversation context
+    conversation_state = gr.State(initialize_conversation())
+    
+    with gr.Row():
+        with gr.Column(scale=4):
+            chatbot = gr.Chatbot(
+                label="Conversation",
+                height=500,
+                show_copy_button=True
+            )
+            
+            with gr.Row():
+                msg = gr.Textbox(
+                    label="Your Message",
+                    placeholder="Type your message here...",
+                    lines=2,
+                    scale=4
+                )
+                submit_btn = gr.Button("Send", variant="primary", scale=1)
+            
+            with gr.Row():
+                clear_btn = gr.Button("New Conversation", variant="secondary")
+                status_box = gr.Textbox(
+                    label="Status",
+                    interactive=False,
+                    lines=1
+                )
+        
+        with gr.Column(scale=1):
+            gr.Markdown("### Debug Info")
+            info_btn = gr.Button("Show Conversation Info")
+            info_display = gr.Markdown("Click button to show info")
+    
+    # Event handlers
+    submit_btn.click(
+        fn=process_message,
+        inputs=[msg, chatbot, conversation_state],
+        outputs=[msg, chatbot, conversation_state, status_box]
+    )
+    
+    msg.submit(
+        fn=process_message,
+        inputs=[msg, chatbot, conversation_state],
+        outputs=[msg, chatbot, conversation_state, status_box]
+    )
+    
+    clear_btn.click(
+        fn=clear_conversation,
+        inputs=[],
+        outputs=[chatbot, conversation_state, status_box]
+    )
+    
+    info_btn.click(
+        fn=get_conversation_info,
+        inputs=[conversation_state],
+        outputs=[info_display]
+    )
+    
+    # Initialize status on load
+    demo.load(
+        fn=lambda state: f"Ready | Thread: {state['thread_id'][:8]}...",
+        inputs=[conversation_state],
+        outputs=[status_box]
+    )
+
+
+# Launch the app
+if __name__ == "__main__":
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=int(os.environ.get('PORT', 7860)),
+        share=False
+    )

langgraph.json

@@ -0,0 +1,10 @@
+{
+    "dependencies": ["."],
+    "graphs": {
+        "query_graph": "./src/graphs/queryGraph.py:graph",
+        "execution_agent_graph": "./src/graphs/actionGraph.py:graph",
+        "master_graph": "./src/graphs/masterGraph.py:master_graph",
+        "finalSparrowAgent": "./src/graphs/finalAgentGraph.py:sparrowAgent"
+    },
+    "env": "./env"
+}

main.py

@@ -0,0 +1,6 @@
+def main():
+    print("Hello from sparrow-agent!")
+
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -0,0 +1,22 @@
+[project]
+name = "sparrow-agent"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.11"
+dependencies = [
+    "fastapi>=0.116.1",
+    "flask>=3.1.2",
+    "gradio>=5.49.1",
+    "langchain>=0.3.27",
+    "langchain-community>=0.3.27",
+    "langchain-core>=0.3.74",
+    "langchain-groq>=0.3.7",
+    "langgraph>=0.6.6",
+    "langgraph-cli[inmem]>=0.3.8",
+    "streamlit>=1.48.1",
+    "tavily-python>=0.7.11",
+    "uuid>=1.30",
+    "uvicorn>=0.35.0",
+    "watchdog>=6.0.0",
+]

requirements.txt

@@ -0,0 +1,17 @@
+langchain
+langgraph
+langchain_community
+langchain_core
+langchain_groq
+fastapi
+uvicorn
+watchdog
+langgraph-cli[inmem]
+streamlit
+tavily-python
+flask
+python-dotenv
+requests
+pymongo
+uuid
+gradio

src/graphs/actionGraph.py

@@ -0,0 +1,55 @@
+from langgraph.graph import StateGraph, START, END
+from src.states.masterState import ExecutorState, ExecutorOutputState
+
+from src.nodes.actionNode import ExecutorNode
+from src.llms.groqllm import GroqLLM
+from src.utils.prompts import execution_agent_prompt, compress_execution_human_message, compress_execution_system_prompt
+
+from src.utils.utils import think_tool, track_package, estimated_time_analysis
+
+tools = [think_tool, track_package, estimated_time_analysis]
+
+
+class ExecutorGraphBuilder:
+    def __init__(self, llm ):
+        self.llm = llm 
+        self.graph = StateGraph(ExecutorState, output=ExecutorOutputState)
+        self.tools = tools
+        self.execution_agent_prompt = execution_agent_prompt
+        self.compress_execution_system_prompt = compress_execution_system_prompt
+        self.compress_execution_human_message = compress_execution_human_message
+    
+    def build_executor_graph(self):
+        """Build a graph to build the executor"""
+        self.executor_node_obj = ExecutorNode(
+            self.llm, 
+        )
+
+        self.graph.add_node("llm_call", self.executor_node_obj.llm_call)
+        self.graph.add_node("tool_node", self.executor_node_obj.tool_node)
+        self.graph.add_node("compress_execution", self.executor_node_obj.compress_execution)
+
+        # Flow
+        self.graph.add_edge(START, "llm_call")
+        self.graph.add_conditional_edges(
+            "llm_call",
+            self.executor_node_obj.guard_llm,
+            {
+                "tool_node": "tool_node",
+                "compress_execution": "compress_execution",
+            },
+        )
+        self.graph.add_edge("tool_node", "llm_call")
+        self.graph.add_edge("compress_execution", END)
+
+        return self.graph
+    
+    def setup_graph(self):
+        return self.graph.compile()
+
+
+llm=GroqLLM().get_llm()
+
+## Creating the graph
+graph_builder=ExecutorGraphBuilder(llm)
+graph=graph_builder.build_executor_graph().compile()

src/graphs/finalAgentGraph.py

@@ -0,0 +1,173 @@
+# Updated Sparrow Agent with proper routing
+import asyncio
+import logging
+from src.graphs.masterGraph import master_graph
+from src.llms.groqllm import GroqLLM
+from src.states.queryState import SparrowAgentState, SparrowInputState
+from langgraph.graph import StateGraph, START, END
+from src.states.masterState import MasterState
+from langgraph.checkpoint.memory import MemorySaver
+from src.nodes.queryNode import QueryNode
+from langchain_core.messages import HumanMessage
+
+logger = logging.getLogger(__name__)
+
+llm = GroqLLM().get_llm()
+queryNode = QueryNode(llm)
+
+def convert_sparrow_to_master(state: SparrowAgentState) -> dict:
+    """Convert SparrowAgentState to master graph input format"""
+    return {
+        "query_brief": state.get("query_brief", ""),
+        "execution_jobs": [],
+        "completed_jobs": [],
+        "worker_outputs": [],
+        "final_output": ''
+    }
+
+def update_sparrow_from_master(sparrow_state: SparrowAgentState, master_state: dict) -> SparrowAgentState:
+    """Update sparrow state with master results"""
+    # Add the final result as a message and update notes
+    from langchain_core.messages import AIMessage
+    
+    final_output = master_state.get("final_output", "")
+    if final_output:
+        sparrow_state["messages"] = sparrow_state.get("messages", []) + [AIMessage(content=final_output)]
+        sparrow_state["final_message"] = final_output
+        
+    # Add execution details to notes
+    execution_jobs = master_state.get("execution_jobs", [])
+    completed_jobs = master_state.get("completed_jobs", [])
+    
+    if execution_jobs:
+        sparrow_state["notes"] = sparrow_state.get("notes", []) + [f"Execution jobs: {', '.join(execution_jobs)}"]
+    
+    if completed_jobs:
+        sparrow_state["notes"] = sparrow_state.get("notes", []) + [f"Completed: {', '.join(completed_jobs)}"]
+    
+    return sparrow_state
+
+def route_after_clarification(state: SparrowAgentState) -> str:
+    """Route based on clarification status from queryNode response"""
+    
+    # Check if clarification_complete flag is set (most reliable)
+    clarification_complete = state.get("clarification_complete", False)
+    needs_clarification = state.get("needs_clarification", True)
+    
+    if clarification_complete or not needs_clarification:
+        print("Clarification complete, proceeding to query brief")
+        return "write_query_brief"
+    
+    # Check messages for clarification status as fallback
+    messages = state.get("messages", [])
+    if not messages:
+        return "need_clarification"
+    
+    # Prevent infinite clarification loops
+    if len(messages) > 10:
+        print("Too many clarification rounds, proceeding to query brief")
+        return "write_query_brief"
+    
+    # Default: needs more clarification
+    print("More clarification needed")
+    return "need_clarification"
+
+def route_after_query_brief(state: SparrowAgentState) -> str:
+    """Route after query brief creation"""
+    
+    # Check if query brief exists and is adequate
+    query_brief = state.get("query_brief", "")
+    
+    if query_brief and len(query_brief.strip()) > 20:  # Reasonable length check
+        print(f"Query brief created: {query_brief[:100]}...")
+        return "master_subgraph"
+    else:
+        # Check how many times we've tried
+        messages = state.get("messages", [])
+        if len(messages) > 15:  
+            print("Too many attempts, ending conversation")
+            return "__end__"
+        
+        print("Query brief insufficient or missing, going back to clarification")
+        state["notes"] = state.get("notes", []) + ["Query brief creation failed, requesting more clarification"]
+        return "clarify_with_user"
+
+def need_clarification(state: SparrowAgentState) -> SparrowAgentState:
+    """Handle case where clarification is needed"""
+    from langchain_core.messages import AIMessage
+    
+    print("Additional clarification needed.")
+    
+    # Add a message indicating we need more information
+    clarification_msg = AIMessage(
+        content="I need a bit more information to help you effectively. Could you provide more details about your request?"
+    )
+    
+    state["messages"] = state.get("messages", []) + [clarification_msg]
+    state["notes"] = state.get("notes", []) + ["Requested additional clarification from user"]
+    
+    return state
+
+def run_master_subgraph(state: SparrowAgentState) -> SparrowAgentState:
+    """Run the master subgraph - using sync version to avoid async issues with Send"""
+    try:
+        print("Running master subgraph...")
+        master_input = convert_sparrow_to_master(state)
+        
+        # Use invoke instead of ainvoke to avoid issues with Send
+        master_result = master_graph.invoke(master_input)
+        
+        return update_sparrow_from_master(state, master_result)
+        
+    except Exception as e:
+        logger.error(f"Master subgraph failed: {e}")
+        return {**state, "error": str(e)}
+
+def route_after_need_clarification(state: SparrowAgentState) -> str:
+    """Route after need_clarification node - always end to wait for user input"""
+    return "__end__"
+
+# Build the graph
+sparrowAgentBuilder = StateGraph(SparrowAgentState, input_schema=SparrowInputState)
+
+sparrowAgentBuilder.add_node("clarify_with_user", queryNode.clarify_with_user)
+sparrowAgentBuilder.add_node("need_clarification", need_clarification)
+sparrowAgentBuilder.add_node("write_query_brief", queryNode.write_query_brief)
+sparrowAgentBuilder.add_node("master_subgraph", run_master_subgraph)
+
+# Edges
+sparrowAgentBuilder.add_edge(START, "clarify_with_user")
+
+sparrowAgentBuilder.add_conditional_edges(
+    "clarify_with_user",
+    route_after_clarification,
+    {
+        "need_clarification": "need_clarification",
+        "write_query_brief": "write_query_brief",
+        "__end__": END
+    }
+)
+
+# Improved clarification flow
+sparrowAgentBuilder.add_conditional_edges(
+    "need_clarification",
+    route_after_need_clarification,
+    {
+        "clarify_with_user": "clarify_with_user",
+        "__end__": END
+    }
+)
+
+sparrowAgentBuilder.add_conditional_edges(
+    "write_query_brief",
+    route_after_query_brief,
+    {
+        "clarify_with_user": "clarify_with_user",
+        "master_subgraph": "master_subgraph",
+        "__end__": END
+    }
+)
+
+sparrowAgentBuilder.add_edge("master_subgraph", END)
+
+sparrowAgent = sparrowAgentBuilder.compile()

src/graphs/masterGraph.py

@@ -0,0 +1,34 @@
+from langgraph.graph import StateGraph, START, END
+from src.nodes.masterNode import MasterOrchestrator
+from src.states.masterState import MasterState
+from src.llms.groqllm import GroqLLM
+
+
+class MasterBuilder:
+    def __init__(self, llm, ):
+        self.llm = llm
+
+    def build_master_graph(self):
+        master_obj = MasterOrchestrator(self.llm)
+        master_graph = StateGraph(MasterState)
+        
+        # Add nodes
+        master_graph.add_node("orchestrator", master_obj.orchestrator)
+        master_graph.add_node("worker_executor", master_obj.worker_executor)
+        master_graph.add_node("synthesizer", master_obj.synthesizer)
+        
+        # Add edges
+        master_graph.add_edge(START, "orchestrator")
+        master_graph.add_conditional_edges("orchestrator", master_obj.assign_workers, ["worker_executor"])
+        master_graph.add_edge("worker_executor", "synthesizer")
+        master_graph.add_edge("synthesizer", END)
+        
+        return master_graph.compile()
+
+
+
+# Building the graph
+llm = GroqLLM().get_llm()
+graph_builder = MasterBuilder(llm)
+master_graph = graph_builder.build_master_graph()
+print("Graph created successfully")

src/graphs/queryGraph.py

@@ -0,0 +1,35 @@
+from langgraph.graph import StateGraph, START, END 
+from src.states.queryState import SparrowAgentState, SparrowInputState
+
+from src.nodes.queryNode import QueryNode
+from src.llms.groqllm import GroqLLM
+
+
+
+class QueryGraphBuilder:
+    def __init__(self, llm):
+        self.llm = llm
+        self.graph = StateGraph(SparrowAgentState, input_schema=SparrowInputState)
+    
+    def build_query_graph(self):
+        """
+        Build a graph for customer query inquiry
+
+        """
+        self.query_node_obj= QueryNode(self.llm)
+        print(self.llm)
+
+        self.graph.add_node("clarify_with_user", self.query_node_obj.clarify_with_user)
+        self.graph.add_node("write_query_brief", self.query_node_obj.write_query_brief)
+
+        self.graph.add_edge(START, "clarify_with_user")
+        self.graph.add_edge("clarify_with_user", "write_query_brief")
+        self.graph.add_edge("write_query_brief", END)
+
+        return self.graph
+    
+llm = GroqLLM().get_llm()
+
+graph_builder=QueryGraphBuilder(llm)
+
+graph=graph_builder.build_query_graph().compile()

src/llms/groqllm.py

@@ -0,0 +1,37 @@
+from langchain_groq import ChatGroq
+import os 
+from dotenv import load_dotenv
+
+class GroqLLM:
+    def __init__(self):
+        load_dotenv()
+
+    def get_llm(self):
+        try:
+            # Don't print API key - security risk
+            # print(os.getenv("GROQ_API_KEY"))
+            os.environ["GROQ_API_KEY"] = self.groq_api_key = os.getenv("GROQ_API_KEY")
+            # Using llama models which have better JSON support
+            llm=ChatGroq(
+                api_key=self.groq_api_key, 
+                model="llama-3.3-70b-versatile",  # Better JSON support
+                streaming=False,
+                temperature=0.1  # Lower temperature for more consistent structured output
+            )
+            return llm
+        except Exception as e:
+            raise ValueError(f"Error occurred with exception: {e}")
+        
+    def get_moon(self):
+        try:
+            # Don't print API key - security risk
+            # print(os.getenv("GROQ_API_KEY"))
+            os.environ["GROQ_API_KEY"] = self.groq_api_key = os.getenv("GROQ_API_KEY")
+            llm=ChatGroq(
+                api_key=self.groq_api_key, 
+                model="moonshotai/kimi-k2-instruct", 
+                streaming=False
+            )
+            return llm
+        except Exception as e:
+            raise ValueError(f"Error occurred with exception: {e}")

src/nodes/actionNode.py

@@ -0,0 +1,214 @@
+from pydantic import BaseModel, Field
+from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage, filter_messages
+from src.utils.prompts import execution_agent_prompt, compress_execution_system_prompt, compress_execution_human_message
+from src.utils.utils import think_tool, track_package, estimated_time_analysis, get_today_str
+import logging
+
+tools = [think_tool, track_package, estimated_time_analysis]
+tools_by_name = {tool.name: tool for tool in tools}
+
+class ExecutorNode:
+    """
+    Executor node for handling tasks:
+    1. LLM reasoning
+    2. Tool invocation
+    3. Final compression
+    """
+
+    def __init__(self, llm):
+        self.llm = llm
+        self.tools = tools
+        self.tools_by_name = {tool.name: tool for tool in tools}
+        self.model_with_tools = llm.bind_tools(tools)
+        self.MAX_ITERATIONS = 6  # Increased to allow more tool calls (including think_tool)
+        self.execution_agent_prompt_template = execution_agent_prompt
+        self.compress_execution_system_prompt_template = compress_execution_system_prompt
+        self.compress_execution_human_message = compress_execution_human_message
+        
+        # Debug tool binding
+        print(f"Available tools: {list(self.tools_by_name.keys())}")
+
+    def llm_call(self, state: dict) -> dict:
+        """Calls the LLM with the executor message history and returns updated state."""
+        try:
+            # Ensure we have the execution job in the messages
+            execution_job = state.get("execution_job", "")
+            existing_messages = state.get("executor_messages", [])
+            print("EXECUTOR MESSAGES MESSAGES", existing_messages)
+            print("EXECUTION JOB", execution_job)
+
+                        # If no existing messages, add the execution job as initial human message
+            if not existing_messages and execution_job:
+                existing_messages = [HumanMessage(content=execution_job)]
+            
+            # Format the prompt with current date
+            formatted_prompt = self.execution_agent_prompt_template.format(date=get_today_str())
+            messages = [SystemMessage(content=formatted_prompt)] + existing_messages
+            
+            print(f"Calling LLM with {len(messages)} messages")
+            print(f"Last message: {messages[-1] if messages else 'No messages'}")
+            
+            response = self.model_with_tools.invoke(messages)
+            
+            print(f"LLM Response type: {type(response)}")
+            print(f"LLM Response content: {response.content[:100] if response.content else 'No content'}...")
+            print(f"Tool calls in response: {getattr(response, 'tool_calls', 'No tool_calls attribute')}")
+
+            return {
+                **state,
+                "executor_messages": existing_messages + [response]
+            }
+            
+        except Exception as e:
+            return {
+                **state,
+                "error": str(e),
+                "executor_messages": state.get("executor_messages", [])
+            }
+
+    def tool_node(self, state: dict) -> dict:
+        """Executes any tools requested by the LLM and appends ToolMessages."""
+        try:
+            executor_messages = state.get("executor_messages", [])
+            if not executor_messages:
+                print("No executor messages found")
+                return state
+                
+            last_message = executor_messages[-1]
+            print(f"Last message type: {type(last_message)}")
+            print(f"Last message attributes: {dir(last_message)}")
+            
+            # Get tool calls
+            tool_calls = getattr(last_message, "tool_calls", [])
+            print(f"Found {len(tool_calls)} tool calls: {tool_calls}")
+
+            if not tool_calls:
+                print("No tool calls found in last message")
+                return state
+
+            tool_outputs, new_data = [], []
+            
+            for call in tool_calls:
+                print(f"Processing tool call: {call}")
+                
+                tool_name = call.get("name")
+                args = call.get("args", {})
+                tool_id = call.get("id")
+                
+                print(f"Tool: {tool_name}, Args: {args}, ID: {tool_id}")
+                
+                if tool_name in self.tools_by_name:
+                    try:
+                        print(f"Invoking tool {tool_name} with args {args}")
+                        result = self.tools_by_name[tool_name].invoke(args)
+                        print(f"Tool {tool_name} result: {result}")
+                        
+                        tool_message = ToolMessage(
+                            content=str(result), 
+                            name=tool_name, 
+                            tool_call_id=tool_id
+                        )
+                        tool_outputs.append(tool_message)
+                        new_data.append(str(result))
+                        
+                    except Exception as e:
+                        error_msg = f"Tool {tool_name} failed: {e}"
+                        print(f"Tool error: {error_msg}")
+                        tool_outputs.append(
+                            ToolMessage(
+                                content=error_msg, 
+                                name=tool_name, 
+                                tool_call_id=tool_id
+                            )
+                        )
+                        new_data.append(error_msg)
+                else:
+                    error_msg = f"Tool {tool_name} not found. Available: {list(self.tools_by_name.keys())}"
+                    print(error_msg)
+                    tool_outputs.append(
+                        ToolMessage(
+                            content=error_msg, 
+                            name=tool_name, 
+                            tool_call_id=tool_id
+                        )
+                    )
+
+            print(f"Returning {len(tool_outputs)} tool outputs")
+            
+            return {
+                **state,
+                "executor_messages": executor_messages + tool_outputs,
+                "executor_data": state.get("executor_data", []) + new_data
+            }
+            
+        except Exception as e:
+            return {
+                **state,
+                "error": f"Tool execution failed: {str(e)}"
+            }
+
+    def compress_execution(self, state: dict) -> dict:
+        """Summarizes the execution and returns final structured output."""
+        try:
+            execution_job = state.get("execution_job", "Complete the assigned task")
+            executor_messages = state.get("executor_messages", [])
+            
+            # Format the system prompt with current date
+            formatted_system_prompt = self.compress_execution_system_prompt_template.format(date=get_today_str())
+            
+            messages = [
+                SystemMessage(content=formatted_system_prompt),
+                *executor_messages,
+                HumanMessage(content=self.compress_execution_human_message.format(
+                    shipment_request=execution_job
+                ))
+            ]
+
+            response = self.llm.invoke(messages)
+
+            executor_data = [
+                str(m.content) for m in executor_messages 
+                if hasattr(m, 'content') and m.content
+            ]
+
+            return {
+                "output": str(response.content),
+                "executor_data": executor_data,
+                "executor_messages": executor_messages
+            }
+            
+        except Exception as e:
+            return {
+                "output": f"Execution completed with errors: {str(e)}",
+                "executor_data": state.get("executor_data", []),
+                "executor_messages": state.get("executor_messages", [])
+            }
+
+    def route_after_llm(self, state: dict) -> str:
+        """Route: decide whether to call a tool or finalize."""
+        try:
+            executor_messages = state.get("executor_messages", [])
+            if not executor_messages:
+                return "compress_execution"
+                
+            last_msg = executor_messages[-1]
+            has_tool_calls = bool(getattr(last_msg, "tool_calls", None))
+            
+            print(f"Routing decision - Has tool calls: {has_tool_calls}")
+            
+            return "tool_node" if has_tool_calls else "compress_execution"
+        except Exception as e:
+            return "compress_execution"
+
+    def guard_llm(self, state: dict) -> str:
+        """Prevent infinite loops by limiting iterations."""
+        iteration_count = state.get("iteration_count", 0) + 1
+        state["iteration_count"] = iteration_count
+        
+        print(f"Iteration count: {iteration_count}/{self.MAX_ITERATIONS}")
+        
+        if iteration_count > self.MAX_ITERATIONS:
+            print("Max iterations reached, finalizing...")
+            return "compress_execution"
+            
+        return self.route_after_llm(state)

src/nodes/masterNode.py

@@ -0,0 +1,104 @@
+from langchain_core.messages import SystemMessage, HumanMessage
+from src.llms.groqllm import GroqLLM
+from src.states.masterState import MasterState, ExecutorState
+from src.nodes.actionNode import ExecutorNode
+import asyncio
+from typing import List, Dict
+from src.utils.prompts import master_agent_prompt
+from src.utils.utils import get_today_str
+from src.states.masterState import PlannerOutput
+from langgraph.constants import Send
+from src.graphs.actionGraph import graph
+
+class MasterOrchestrator:
+    def __init__(self, llm):
+        self.llm = llm
+        self.master_planner = llm.with_structured_output(PlannerOutput)
+        self.compiled_worker_graph = graph
+        self.master_agent_prompt_template = master_agent_prompt
+
+    def orchestrator(self, state: MasterState):
+        """Generate a plan by breaking down the query into execution jobs"""
+        
+        system_prompt = """You are a master task planner. Given a query, break it down into specific, actionable execution jobs.
+        
+        Each job should be:
+        1. Clear and specific
+        2. Actionable by a specialized worker
+        3. Independent or clearly sequenced
+        4. Focused on a single objective
+        
+        Return a list of execution jobs as strings."""
+        
+        planner_result = self.master_planner.invoke([
+            SystemMessage(content=system_prompt),
+            HumanMessage(content=f"Here is the query brief: {state['query_brief']}")
+        ])
+
+        print("Execution Jobs Generated:", planner_result.executor_jobs)
+        return {"execution_jobs": planner_result.executor_jobs}
+
+    def worker_executor(self, worker_input: dict):
+        """Execute a single job using the worker graph"""
+        
+        job_description = worker_input["execution_job"]
+        
+        # Prepare the initial state for the worker
+        # Pass the full job description as the execution_job - the worker will use available tools
+        worker_state = {
+            "executor_messages": [HumanMessage(content=job_description)],
+            "execution_job": job_description,  # Pass the full job description
+            "executor_data": []
+        }
+        
+        print(f"Executing job: {job_description}")
+        
+        # Execute the worker graph
+        try:
+            result = self.compiled_worker_graph.invoke(worker_state)
+            
+            # Return the completed job info
+            return {
+                "completed_jobs": [f"Job: {job_description} - Status: Completed"],
+                "worker_outputs": [result]
+            }
+        except Exception as e:
+            error_result = {
+                "output": f"Error executing job: {str(e)}",
+                "executor_data": [f"Error: {str(e)}"],
+                "executor_messages": []
+            }
+            return {
+                "completed_jobs": [f"Job: {job_description} - Status: Failed - {str(e)}"],
+                "worker_outputs": [error_result]
+            }
+
+    def assign_workers(self, state: MasterState):
+        """Assign a worker to each execution job using Send"""
+        return [
+            Send("worker_executor", {"execution_job": job}) 
+            for job in state["execution_jobs"]
+        ]
+
+    def synthesizer(self, state: MasterState):
+        """Combine all completed jobs into a final output"""
+        
+        # Create a synthesis prompt
+        synthesis_prompt = f"""
+        Original Query: {state['query_brief']}
+        
+        Completed Jobs Summary:
+        {chr(10).join([f"- {job}" for job in state['completed_jobs']])}
+        
+        Detailed Worker Outputs:
+        {chr(10).join([f"Output {i+1}: {output.get('output', 'No output')}" for i, output in enumerate(state['worker_outputs'])])}
+        
+        Please synthesize all the work into a comprehensive final response that addresses the original query.
+        """
+        
+        synthesis_result = self.llm.invoke([
+            SystemMessage(content="You are a synthesis expert. Combine the worker outputs into a coherent final response."),
+            HumanMessage(content=synthesis_prompt)
+        ])
+        
+        return {"final_output": synthesis_result.content}

src/nodes/queryNode.py

@@ -0,0 +1,151 @@
+
+
+
+
+from datetime import datetime
+from typing_extensions import Literal
+from src.llms.groqllm import GroqLLM
+from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, get_buffer_string
+from src.utils.prompts import clarification_with_user_instructions, transform_messages_into_customer_query_brief_prompt
+from src.states.queryState import SparrowAgentState, ClarifyWithUser, CustomerQuestion
+from src.utils.utils import get_today_str
+
+class QueryNode:
+    def __init__(self, llm):
+        self.llm = llm
+        
+    def clarify_with_user(self, state: SparrowAgentState) -> SparrowAgentState:
+        """
+        Determine if the user's request contains sufficient information to proceed.
+        Returns updated state with clarification status.
+        """
+        try:
+            # Use structured output with method="json_mode" for better compatibility
+            structured_output_model = self.llm.with_structured_output(
+                ClarifyWithUser,
+                method="json_mode",
+                include_raw=False
+            )
+            
+            response = structured_output_model.invoke([
+                SystemMessage(
+                    content="You are a helpful assistant that responds in JSON format. Route the input to yes or no based on the need of clarification of the query."
+                ),
+                HumanMessage(
+                    content=clarification_with_user_instructions.format(
+                        messages=get_buffer_string(messages=state.get("messages", [])),
+                        date=get_today_str()
+                    )
+                )
+            ])
+            
+            print("CLARIFICATION RESPONSE:", response)
+            
+            # Update state based on response
+            updated_state = {**state}
+            
+            if response.need_clarification == 'yes':
+                updated_state.update({
+                    "messages": state.get("messages", []) + [AIMessage(content=response.question)],
+                    "clarification_complete": False,
+                    "needs_clarification": True
+                })
+            else:
+                updated_state.update({
+                    "messages": state.get("messages", []) + [AIMessage(content=response.verification)],
+                    "clarification_complete": True,
+                    "needs_clarification": False
+                })
+            
+            return updated_state
+            
+        except Exception as e:
+            print(f"Error in clarify_with_user: {e}")
+            print(f"Error type: {type(e).__name__}")
+            import traceback
+            traceback.print_exc()
+            
+            # Fallback: Ask for clarification if there's an error
+            return {
+                **state,
+                "messages": state.get("messages", []) + [
+                    AIMessage(content="I'd be happy to help! Could you please provide more details about what you need? For example, if you want to track a package, please share the tracking number.")
+                ],
+                "clarification_complete": False,
+                "needs_clarification": True,
+                "error": str(e)
+            }
+
+    def write_query_brief(self, state: SparrowAgentState) -> SparrowAgentState:
+        """
+        Transform the conversation history into a comprehensive customer query brief.
+        """
+        try:
+            # Use structured output with json_mode for better compatibility
+            structured_output_model = self.llm.with_structured_output(
+                CustomerQuestion,
+                method="json_mode",
+                include_raw=False
+            )
+            
+            messages = state.get("messages", [])
+            print("STATE MESSAGES:", messages)
+            
+            if not messages:
+                print("ERROR: No messages in state")
+                return {
+                    **state,
+                    "query_brief": "",
+                    "error": "No messages available for query brief creation"
+                }
+            
+            prompt = transform_messages_into_customer_query_brief_prompt.format(
+                messages=get_buffer_string(messages),
+                date=get_today_str()
+            )
+            print("PROMPT:", prompt[:200], "...")  # Print first 200 chars only
+            
+            # Get structured response
+            response = structured_output_model.invoke([
+                SystemMessage(content="You are a helpful assistant that responds in JSON format."),
+                HumanMessage(content=prompt)
+            ])
+            print("STRUCTURED RESPONSE:", response)
+            
+            if response is None:
+                print("ERROR: Structured response is None")
+                return {
+                    **state,
+                    "query_brief": "",
+                    "error": "Failed to generate structured response"
+                }
+            
+            return {
+                **state,
+                "query_brief": response.query_brief,
+                "master_messages": [HumanMessage(content=response.query_brief)],
+                "query_brief_complete": True
+            }
+            
+        except Exception as e:
+            print(f"Error in write_query_brief: {e}")
+            print(f"Error type: {type(e).__name__}")
+            import traceback
+            traceback.print_exc()
+            
+            # Fallback: Create a simple query brief from the messages
+            messages = state.get("messages", [])
+            if messages:
+                # Extract the last user message as the query brief
+                user_messages = [msg.content for msg in messages if hasattr(msg, 'type') and msg.type == 'human']
+                fallback_brief = user_messages[-1] if user_messages else "Help with parcel query"
+            else:
+                fallback_brief = "Help with parcel query"
+            
+            return {
+                **state,
+                "query_brief": fallback_brief,
+                "master_messages": [HumanMessage(content=fallback_brief)],
+                "query_brief_complete": True,
+                "error": str(e)
+            }

src/states/actionState.py

@@ -0,0 +1,39 @@
+
+import operator 
+from typing_extensions import TypedDict, Annotated, List, Sequence
+from pydantic import BaseModel, Field
+from langchain_core.messages import BaseMessage
+from langgraph.graph.message import add_messages
+from typing_extensions import Literal
+
+
+class ExecutorState(TypedDict):
+    """
+    State for the executor agent containing message history and research metadata.
+
+    This state tracks the executors conversation, iteration count for limiting
+    tool calls, the executor topic being
+    """
+    executor_messages: Annotated[Sequence[BaseMessage], add_messages]
+    execution_job: str
+    executor_data: List[str]
+
+class ExecutorOutputState(TypedDict):
+    """
+    Output state for the executor agent containing final executor results.
+
+    This represents the final output of the execution process with executor_data,
+    executor_messages and output from the execution process.
+
+    """
+    output: str  
+    executor_data: List[str]
+    executor_messages: Annotated[Sequence[BaseMessage], add_messages]
+
+
+# Structured output schema 
+class CustomerQuestion(BaseModel):
+    """Schema for customer query brief generation"""
+    query_brief: str = Field(description="A customer question that will be used to guide the execution")
+
+

src/states/masterState.py

@@ -0,0 +1,37 @@
+from typing_extensions import TypedDict, Annotated, Sequence, List
+from langchain_core.messages import BaseMessage
+from langgraph.graph.message import add_messages
+from langchain_core.tools import tool 
+from pydantic import BaseModel, Field
+import operator 
+
+class ExecutorState(TypedDict):
+    """
+    State for the executor agent containing message history and research metadata.
+    """
+    executor_messages: Annotated[Sequence[BaseMessage], add_messages]
+    execution_job: str
+    executor_data: List[str]
+
+class ExecutorOutputState(TypedDict):
+    """
+    Output state for the executor agent containing final executor results.
+    """
+    output: str
+    executor_data: List[str]
+    executor_messages: Annotated[Sequence[BaseMessage], add_messages]
+
+
+class PlannerOutput(BaseModel):
+    """Simplified output for the planner that only returns execution jobs"""
+    executor_jobs: List[str] = Field(description="List of execution jobs to be completed")
+
+class MasterState(TypedDict):
+    """Master orchestrator state"""
+    query_brief: str
+    execution_jobs: List[str]
+    completed_jobs: Annotated[List[str], operator.add]
+    worker_outputs: Annotated[List[ExecutorOutputState], operator.add]
+    final_output: str
+
+

src/states/queryState.py

@@ -0,0 +1,67 @@
+import operator 
+from typing_extensions import Optional, Annotated, List, Sequence, Literal, Union
+
+from langchain_core.messages import BaseMessage
+from langgraph.graph import MessagesState
+from langgraph.graph.message import add_messages
+from pydantic import BaseModel, Field, field_validator 
+
+
+class SparrowInputState(MessagesState):
+    """Input state for the full agent - only contains from the user input."""
+    pass 
+
+class SparrowAgentState(MessagesState):
+    """
+    Main state for the full multi-agent Sparrow customer service system.
+
+    Extends MessagesState with additional fields for Sparrow customer service coordination.
+    Note: Some fields are duplicated across different state classes for proper
+    state management between subgraphs and main workflow.
+
+    """
+    query_brief: Optional[str]
+    master_messages: Annotated[Sequence[BaseMessage], add_messages]
+    notes: Annotated[list[str], operator.add] = []
+    final_message: str
+    clarification_complete: Optional[bool]
+    needs_clarification: Optional[bool]
+    query_brief_complete: Optional[bool]
+    execution_jobs: Optional[List[str]]
+    error: Optional[str]
+
+class ClarifyWithUser(BaseModel):
+    """Schema for user clarification decision and questions"""
+
+    need_clarification: Union[Literal["yes", "no"], bool] = Field(
+        description="Whether the user needs to be asked a clarifying question. Can be 'yes'/'no' or true/false"
+    )
+    question: str = Field(
+        description="A question to ask the user to clarify the need",
+        default=""
+    )
+    verification: str = Field(
+        description="Verify message that we will start research after the user has provided the necessary information",
+        default=""
+    )
+    
+    @field_validator('need_clarification', mode='before')
+    @classmethod
+    def convert_bool_to_string(cls, v):
+        """Convert boolean values to yes/no strings"""
+        if isinstance(v, bool):
+            return "yes" if v else "no"
+        if isinstance(v, str):
+            v_lower = v.lower()
+            if v_lower in ['true', '1', 'yes']:
+                return "yes"
+            elif v_lower in ['false', '0', 'no']:
+                return "no"
+        return v
+
+class CustomerQuestion(BaseModel):
+    """Schema for structured customer query brief """
+
+    query_brief: str = Field(
+        description="A customer question that will be used to guide the research."
+    )

src/utils/prompts.py

@@ -0,0 +1,343 @@
+clarification_with_user_instructions = """
+These are the messages that have been exchanged so far regarding the user's parcel request or tracking inquiry:
+<Messages>
+{messages}
+</Messages>
+
+Today's date is {date}.
+
+You are Sparrow, a friendly and helpful parcel operations assistant. Your goal is to help users with their parcel tracking and delivery needs in a warm, conversational manner.
+
+Assess whether you need to ask a clarifying question, or if the user has already provided enough information for you to proceed.
+IMPORTANT: If you can see in the messages history that you have already asked a clarifying question, you almost always do not need to ask another one. Only ask another question if ABSOLUTELY NECESSARY.
+
+If you need to ask a question, follow these guidelines:
+- Be friendly, warm, and conversational - imagine you're helping a friend
+- Use a casual, approachable tone (e.g., "I'd be happy to help!", "Let me check that for you!")
+- Show empathy and understanding (e.g., "I understand you're waiting for your package")
+- Keep it brief and to the point
+- Use emojis sparingly and appropriately to add warmth (📦, ✅, 🚚)
+- Make the user feel comfortable and valued
+- Use bullet points or numbered lists if appropriate for clarity
+- Do not ask for unnecessary information or information the user has already provided
+
+Respond in valid JSON format with these exact keys:
+"need_clarification": boolean,
+"question": "<question to ask the user to clarify their parcel request>",
+"verification": "<verification message that we will start processing the parcel request>"
+
+If you need to ask a clarifying question, return:
+"need_clarification": "yes",
+"question": "<your friendly clarifying question>",
+"verification": ""
+
+Example friendly questions:
+- "I'd be happy to help you track your package! Could you share the tracking number with me?"
+- "Great! Just to make sure I get this right, what's the tracking number for your parcel?"
+- "No problem! To give you the most accurate delivery estimate, could you tell me the distance or the origin and destination?"
+
+If you do not need to ask a clarifying question, return:
+"need_clarification": "no",
+"question": "",
+"verification": "<friendly acknowledgement message>"
+
+Example friendly verification messages:
+- "Perfect! I've got all the details I need. Let me check that for you right away! 📦"
+- "Great! I'll track down that package for you now. Just a moment!"
+- "Awesome! Let me look into this and get you those delivery details. ✅"
+- "Got it! I'll pull up the tracking information for you right now."
+
+For the verification message when no clarification is needed:
+- Start with a friendly acknowledgement ("Perfect!", "Great!", "Awesome!", "Got it!")
+- Briefly mention what you'll do ("Let me track that for you", "I'll look that up")
+- Add a friendly closing ("Just a moment!", "One sec!", "Right away!")
+- Keep it warm, concise, and reassuring
+- Optional: Use a relevant emoji (📦, 🚚, ✅)
+"""
+
+transform_messages_into_customer_query_brief_prompt = """
+You will be given a set of messages that have been exchanged so far between yourself and the user.  
+Your job is to translate these messages into a detailed and actionable parcel request brief that will be used to guide parcel processing, tracking, or consolidation.
+
+The messages that have been exchanged so far between yourself and the user are:
+<Messages>
+{messages}
+</Messages>
+
+Today's date is {date}.
+
+You will return a single, clear, and actionable brief that summarizes the user's parcel request.
+
+Guidelines:
+1. Maximize Specificity and Detail
+- Include all known shipment details provided by the user (e.g., parcel dimensions, weight, destination, delivery preferences, tracking numbers).
+- Explicitly list any key attributes or instructions mentioned by the user for consolidation, delivery timing, or handling.
+
+2. Handle Unstated Dimensions Carefully
+- If certain aspects are not specified but are critical for processing (e.g., preferred courier, insurance, or packaging requirements), note them as open considerations rather than making assumptions.
+- Example: Instead of assuming “fast delivery,” say “consider delivery options unless the user specifies a preference.”
+
+3. Avoid Unwarranted Assumptions
+- Never invent shipment details, preferences, or constraints that the user hasn’t stated.
+- If certain details are missing (e.g., shipment value, fragile contents), explicitly note this lack of specification.
+
+4. Distinguish Between Required Actions and Optional Preferences
+- Required actions: Steps or details necessary for successful processing or tracking of the shipment.
+- User preferences: Specific instructions provided by the user (must only include what the user stated).
+- Example: “Process and track parcels to the provided address, prioritizing delivery timing as specified by the user.”
+
+5. Use the First Person
+- Phrase the brief from the perspective of the user.
+
+6. Sources / References
+- If tracking numbers or courier platforms are mentioned, reference the official tracking pages or courier portals.
+- If specific shipment services or rules apply, prioritize official service guidelines over general logistics advice.
+"""
+
+
+
+## Defining the prompts 
+compress_execution_system_prompt = """You are Sparrow, a friendly and helpful parcel operations assistant who has gathered logistics information. Your job is to present the findings in a clear, friendly, and conversational way while preserving all relevant details. For context, today's date is {date}.
+
+<Tone Guidelines>
+- Be warm, friendly, and conversational
+- Use simple, clear language that anyone can understand
+- Show empathy and understanding
+- Be encouraging and positive
+- Use appropriate emojis to enhance clarity and warmth (📦, 🚚, ✅, 📍, ⏰)
+- Avoid technical jargon or overly formal language
+- Make the information easy to scan and understand
+</Tone Guidelines>
+
+<Task>
+You need to clean up information gathered from tool calls and web searches in the existing messages.
+All relevant parcel, shipment, tracking, and user information (e.g., status, ETA, user history, reports) should be repeated and rewritten in a cleaner, structured format.
+The purpose is to remove irrelevant or duplicate information (e.g., if multiple sources confirm the same ETA, state "Multiple sources confirm ETA is X").
+Only these fully comprehensive cleaned findings will be returned to the user, so it is crucial that no relevant information is lost from the raw messages.
+</Task>
+
+<Tool Call Filtering>
+**IMPORTANT**: When processing the research messages, focus only on substantive shipment, tracking, and delivery-related content:
+- **Include**: Results from `track_package`, `estimated_time_analysis`, and findings from carrier websites, courier portals, customs/government sites, and Sparrow docs.
+- **Exclude**: `think_tool` calls and responses — these are internal agent reflections for decision-making and should not be included in the final report.
+- **Focus on**: Actual information gathered from tools (e.g., parcel status, ETA predictions, generated reports) and official sources (e.g., transit times, delivery restrictions, service updates), not the agent's internal reasoning.
+</Tool Call Filtering>
+
+<Guidelines>
+1. Your output findings must be fully comprehensive and include ALL shipment, tracking, and delivery details from tool calls and web searches. Repeat key details verbatim.
+2. The cleaned logistics report can be as long as necessary to include ALL information gathered.
+3. Include inline citations for each source (carrier site, government/customs portal, Sparrow docs, or tool outputs).
+4. Add a "Sources" section at the end listing all sources (including tool outputs) with corresponding citations.
+5. Ensure every source and tool result used in gathering parcel/tracking/delivery information is preserved.
+6. Critical: Do not lose any source or tool output, even if it appears repetitive — future steps will handle merging/aggregation.
+7. For tool outputs, treat results from `track_package` and `estimated_time_analysis` as authoritative sources and cite them as "Sparrow Tool: [Tool Name]".
+</Guidelines>
+
+<Output Format>
+Present the information in a friendly, easy-to-read format:
+
+**Opening**: Start with a friendly acknowledgment
+- Example: "Here's what I found about your package! 📦"
+- Example: "Good news! I've got the details you need. ✅"
+
+**Main Information**: Present key details clearly
+- Use friendly headers and bullet points
+- Highlight important information (status, ETA, location)
+- Present details from `track_package` and `estimated_time_analysis` in a user-friendly way
+- Use emojis to make information easier to scan
+
+**Closing**: End with a helpful offer
+- Example: "Is there anything else you'd like to know about your delivery?"
+- Example: "Let me know if you need any other help! 😊"
+
+**Citations**: Include sources naturally in the text or at the end
+- Example: "According to our tracking system..."
+- Keep source citations brief and unobtrusive
+
+**CRITICAL - System Error Handling**:
+If tool results contain error messages about MongoDB, database connections, timeouts, or system failures:
+- DO NOT include technical error details in your response
+- Replace them with a friendly "system temporarily unavailable" message
+- Example: "I'm sorry, but our tracking system is temporarily unavailable. 🛠️ Please try again in a few minutes!"
+- Show empathy and provide clear next steps
+- Never expose internal system architecture or technical issues to users
+</Output Format>
+
+<Citation Rules>
+- Assign each unique source (URL or tool) a single citation number in your text.
+- End with ### Sources listing each source/tool with corresponding numbers.
+- **IMPORTANT**: Number sources sequentially without gaps (1,2,3,4...) in the final list, regardless of order.
+- Example format:
+  [1] Sparrow Tool: track_package
+  [2] DHL Tracking Portal: https://www.dhl.com/track
+  [3] Sparrow Tool: generate_report
+</Citation Rules>
+
+Critical Reminder: It is extremely important that any information relevant to the user's parcel request (status codes, ETAs, delivery limits, customs rules, courier policies, user history, generated reports) is preserved verbatim. Do not rewrite or paraphrase this information — keep it intact.
+
+"""
+
+compress_execution_human_message = """All above messages are about parcel/tracking research conducted by Sparrow's AI Operations Assistant for the following shipment request: 
+
+SHIPMENT REQUEST: {shipment_request}
+
+Your task is to clean up these logistics findings while preserving ALL information that is relevant to answering this specific shipment or tracking question. 
+
+CRITICAL REQUIREMENTS:
+- DO NOT summarize or paraphrase the information — preserve it verbatim
+- DO NOT lose any shipment details, tracking events, carrier names, numbers, or service constraints
+- DO NOT filter out information that seems relevant to the shipment request
+- Organize the information in a cleaner format but keep all the substance
+- Include ALL sources and citations found during research
+- Remember this research was conducted to answer the specific shipment/tracking request above
+
+The cleaned findings will be used for final report generation, so comprehensiveness is critical."""
+
+"""Prompt templates for the deep research system.
+
+This module contains all prompt templates used across the research workflow components,
+including user clarification, research brief generation, and report synthesis.
+"""
+
+
+
+execution_agent_prompt = """You are Sparrow, a friendly and helpful parcel operations assistant. Today's date is {date}.
+
+<Personality>
+You are warm, conversational, and genuinely helpful. You communicate like a knowledgeable friend who's excited to help users with their parcels. Use:
+- Friendly, casual language ("Let me check that for you!", "Here's what I found!")
+- Empathy and understanding ("I know waiting for packages can be frustrating")
+- Clear, straightforward explanations without jargon
+- Positive, reassuring tone
+- Appropriate emojis to add warmth (📦, 🚚, ✅, 📍)
+</Personality>
+
+<Task>
+Your job is to gather and verify information needed to process, track, or consolidate parcels. Goals include: interpreting tracking events and ETAs, checking carrier service availability/alerts, comparing delivery options or confirming packaging/size limits, and retrieving user-specific shipment details.
+</Task>
+
+<Available Tools>
+1. **think_tool(reflection: str)**: Summarize findings, note gaps, and plan next steps. Must always be called after any other tool call.
+2. **track_package(tracking_number: str)**: Tracks parcels using a tracking number.
+3. **estimated_time_analysis(distance_km: float, courier_experience_yrs: float, vehicle_type: str, weather: str, time_of_day: str, traffic_level: str)**: Estimates delivery time using ML model based on delivery parameters.
+   - distance_km: Distance in kilometers (required)
+   - courier_experience_yrs: Experience in years (default: 2.0)
+   - vehicle_type: 'Scooter', 'Pickup Truck', or 'Motorcycle' (default: 'Scooter')
+   - weather: 'Sunny', 'Rainy', 'Foggy', 'Snowy', or 'Windy' (default: 'Sunny')
+   - time_of_day: 'Morning', 'Afternoon', 'Evening', or 'Night' (default: 'Morning')
+   - traffic_level: 'Low', 'Medium', or 'High' (default: 'Medium')
+
+**CRITICAL RULES:**
+- Only call a tool if it is absolutely required to resolve the user’s request.
+- Always provide all required and correct arguments when calling a tool.
+- Never call a tool with empty, null, or placeholder arguments.
+- Never answer a question directly with text if a relevant tool exists and required arguments are available.
+- Always call `think_tool` immediately after any non-think_tool call, with a `reflection` explaining:
+    - What information was obtained
+    - What is missing
+    - Whether another tool call is justified
+</Available Tools>
+
+<Instructions>
+1. **Understand the request** – Determine the exact outcome needed (parcel status, ETA, delivery estimates, etc.).
+2. **Decide tool usage** – Only use a tool if required; otherwise, explain what information is missing to proceed.
+3. **Call tools properly** – Provide the correct arguments, do not assume or fabricate values. For `estimated_time_analysis`, the minimum required argument is `distance_km`.
+4. **Think after each tool** – Use `think_tool` to reflect on results before any further tool calls or final answers.
+5. **Stop when resolved** – Once sufficient data is available, provide a concise, actionable operational response.
+6. **Handle missing details** – If critical information is missing (tracking number, distance, etc.), explicitly state it and do not call a tool.
+</Instructions>
+
+<Hard Limits>
+- Simple requests: maximum 2–3 tool calls (including `think_tool`).
+- Complex requests: maximum 5 tool calls.
+- Stop immediately if tool calls return duplicate or unnecessary information, or the limit is reached without full resolution.
+</Hard Limits>
+
+<Final Output Guidance>
+- Be friendly and conversational in your responses
+- Use clear, simple language that anyone can understand
+- Include sources in a natural way (e.g., "According to our tracking system...")
+- Provide helpful, actionable information
+- Show empathy if there are delays or issues ("I understand this might be frustrating...")
+- Be encouraging and positive when sharing good news ("Great news!" "Your package is on its way!")
+- If unable to complete, politely explain what information you need ("To help you better, I'll need...")
+- End with an offer to help further ("Is there anything else I can help you with?")
+- Use appropriate emojis to enhance friendliness (📦 for packages, 🚚 for delivery, ✅ for success, 📍 for location)
+- Never generate free-text answers instead of calling a tool when the tool is required and arguments are available.
+
+**IMPORTANT - Handling System Errors:**
+If the `track_package` tool returns error messages indicating:
+- "MongoDB is not configured"
+- "Failed to connect to MongoDB"
+- "MongoDB operation failed"
+- "Unable to connect"
+- "Connection refused"
+- "Timeout"
+- Any database/system errors
+
+You MUST respond with a friendly system unavailable message like:
+- "I'm really sorry, but our tracking system seems to be temporarily unavailable right now. 😔 Could you please try again in a few minutes? If the issue persists, feel free to reach out to our support team!"
+- "Oops! It looks like our system is taking a little break. 🛠️ Please try checking your package again in a few moments. Sorry for the inconvenience!"
+- "I apologize, but I'm having trouble connecting to our tracking database at the moment. ⚠️ This is usually temporary. Please try again shortly, or contact our support team if you need immediate assistance!"
+
+DO NOT:
+- Share technical error details with the user
+- Mention MongoDB, database, or internal system names
+- Make the user feel like it's their fault
+- Leave them without any guidance
+
+DO:
+- Apologize sincerely
+- Explain the system is temporarily unavailable
+- Ask them to try again later
+- Offer alternative support options
+- Use appropriate emojis (😔, 🛠️, ⚠️) to soften the message
+"""
+
+master_agent_prompt = """
+You are a master agent for a logistics AI system, responsible for task delegation and coordination. Your job is to analyze logistics queries, break them into subtasks, and assign them to executor agents for parallel execution. Today's date is {date}.
+
+<Task>
+Analyze the logistics query and split it into the minimum number of necessary subtasks for efficient parallel execution. Return subtasks as a comma-separated string. Call the "ExecuteLogisticsTask" tool for each subtask to delegate to executor agents. When satisfied with the results, call the "LogisticsComplete" tool to indicate completion.
+</Task>
+
+<Available Tools>
+1. **think_tool(reflection: str)**: Summarize findings, note gaps, and plan next steps. Must always be called after any other tool call.
+2. **track_package(tracking_number: str)**: Tracks parcels using a tracking number.
+3. **estimated_time_analysis(distance_km: float, ...)**: Estimates delivery time using ML model based on distance and delivery parameters.
+
+**CRITICAL**: Use think_tool before ExecuteLogisticsTask to plan subtasks and after each task to evaluate results. Assign up to {max_concurrent_logistics_units} parallel subtasks per iteration for efficiency.
+</Available Tools>
+
+<Instructions>
+Act as a logistics manager with limited resources. Follow these steps:
+1. **Analyze the query** - Identify specific logistics needs (e.g., routing, inventory, scheduling).
+2. **Break into subtasks** - Decompose query into clear, independent subtasks for parallel execution. Return as comma-separated string.
+3. **Delegate subtasks** - Use ExecuteLogisticsTask for each subtask.
+4. **Assess progress** - After each ExecuteLogisticsTask, use think_tool to evaluate results and decide next steps.
+</Instructions>
+
+<Hard Limits>
+- **Bias towards minimal subtasks** - Use single executor unless query clearly benefits from parallelization.
+- **Stop when sufficient** - Call LogisticsComplete when query is resolved adequately.
+</Hard Limits>
+
+<Show Your Thinking>
+Before ExecuteLogisticsTask:
+- Use think_tool to plan: Can the query be split into independent subtasks (e.g., check inventory, optimize route, schedule delivery)? List as comma-separated string.
+
+After ExecuteLogisticsTask:
+- Use think_tool to analyze: What was achieved? What's missing? Is the query resolved? Should more subtasks be delegated, or is LogisticsComplete appropriate?
+</Show Your Thinking>
+
+<Scaling Rules>
+- **Simple queries** (e.g., check single item inventory): Use 1 executor.
+  - Example: "Check stock for item X in warehouse Y" → 1 subtask.
+- **Complex logistics queries**: Assign one executor per distinct subtask.
+  - Example: "Plan delivery from warehouse A to cities B, C, D" → 3 subtasks (one per city).
+- **Subtask design**: Ensure subtasks are clear, non-overlapping, and standalone. Provide complete instructions for each ExecuteLogisticsTask call.
+- **Note**: A separate agent will compile the final logistics plan; focus on gathering comprehensive data.
+</Scaling Rules>
+
+"""
+

src/utils/utils.py

@@ -0,0 +1,364 @@
+from pathlib import Path 
+from datetime import datetime
+from typing_extensions import Annotated, List, Literal
+import os
+import logging
+import requests
+from typing import Optional, Dict, Any
+from pymongo import MongoClient
+from pymongo.errors import ConnectionFailure, OperationFailure
+
+from langchain_core.messages import HumanMessage
+from langchain_core.tools import tool, InjectedToolArg
+
+from src.llms.groqllm import GroqLLM
+
+# Configure logging
+logger = logging.getLogger(__name__)
+
+# Configuration for MongoDB
+MONGODB_URI = os.getenv("MONGODB_URI", None)
+MONGODB_DATABASE = os.getenv("MONGODB_DATABASE", "parcel_tracking")
+MONGODB_COLLECTION = os.getenv("MONGODB_COLLECTION", "parcels")
+MONGODB_TIMEOUT = int(os.getenv("MONGODB_TIMEOUT", "5000"))  # milliseconds
+
+# Configuration for ETA API
+ETA_API_BASE_URL = os.getenv("ETA_API_BASE_URL", "http://localhost:8000")
+ETA_API_TIMEOUT = int(os.getenv("ETA_API_TIMEOUT", "10"))  # seconds
+
+# MongoDB client singleton
+_mongo_client = None
+
+def get_mongo_client():
+    """Get or create MongoDB client singleton"""
+    global _mongo_client
+    
+    if _mongo_client is None:
+        if not MONGODB_URI:
+            raise ValueError("MONGODB_URI environment variable is not set")
+        
+        try:
+            _mongo_client = MongoClient(
+                MONGODB_URI,
+                serverSelectionTimeoutMS=MONGODB_TIMEOUT,
+                connectTimeoutMS=MONGODB_TIMEOUT
+            )
+            # Test connection
+            _mongo_client.admin.command('ping')
+            logger.info("MongoDB connection established successfully")
+        except Exception as e:
+            logger.error(f"Failed to connect to MongoDB: {e}")
+            raise
+    
+    return _mongo_client
+
+def get_today_str() -> str:
+    """Get current data in a human-readable format."""
+    return datetime.now().strftime("%a %b %d, %Y")
+
+groq = GroqLLM()
+
+@tool 
+def think_tool(reflection:str) -> str:
+    """
+    Tool for strategic reflection on execution progress and decision-making.
+
+    Use this tool after each search to analyze results and plan next steps systematically.
+    This creates a deliberate pause in customer query execution workflow for quality decision-making.
+
+    When to use:
+    - After receiving search results: What key information did I find?
+    - Before deciding next steps: Do I have enough to answer comprehensively?
+    - When assessing execution gaps: What specific execution am I still missing?
+    - Before concluding execution: Can I provide a complete answer now?
+
+    Reflection should address:
+    1. Analysis of current findings - What concrete information have I gathered?
+    2. Gap assessment - What crucial execution or information is still missing?
+    3. Quality evaluation - Do I have sufficient evidence/examples for a good answer?
+    4. Strategic decision - Should I continue execution or provide my output?
+
+    Args: 
+        reflection: Your detailed reflection on the execution progress, findings, gaps, and next steps
+
+    Returns:
+        Confirmation that reflection was recorded for decision-making
+    """
+    return f"Reflection recorded: {reflection}"
+
+
+@tool(description="Track parcel based on tracking number")
+def track_package(tracking_number: str) -> str:
+    """
+    Tool for tracking customer packages/parcels from MongoDB database.
+
+    This tool retrieves real-time parcel tracking information from the MongoDB database.
+    It fetches details such as current status, location, delivery ETA, sender/recipient info,
+    and tracking history.
+
+    Args:
+        tracking_number(str): The unique tracking number of the parcel
+    
+    Returns:
+        A string describing the parcel status, location, history, and other relevant details
+    """
+    logger.info(f"Tracking parcel: {tracking_number}")
+    
+    try:
+        # Check if MongoDB is configured
+        if not MONGODB_URI:
+            error_msg = "MongoDB is not configured. Please set MONGODB_URI environment variable."
+            logger.error(error_msg)
+            return error_msg
+        
+        # Get MongoDB client and collection
+        client = get_mongo_client()
+        db = client[MONGODB_DATABASE]
+        collection = db[MONGODB_COLLECTION]
+        
+        # Query for the tracking number
+        logger.debug(f"Querying MongoDB for tracking_number: {tracking_number}")
+        parcel = collection.find_one({"tracking_number": tracking_number})
+        
+        if not parcel:
+            # Try case-insensitive search
+            parcel = collection.find_one(
+                {"tracking_number": {"$regex": f"^{tracking_number}$", "$options": "i"}}
+            )
+        
+        if not parcel:
+            logger.warning(f"Tracking number not found: {tracking_number}")
+            return f"Tracking number '{tracking_number}' not found in the system. Please verify the tracking number and try again."
+        
+        # Format the response with all available information
+        response_parts = []
+        response_parts.append(f"📦 Parcel Tracking Information for {tracking_number}")
+        response_parts.append("=" * 50)
+        
+        # Basic Information
+        if parcel.get("status"):
+            response_parts.append(f"\n🔹 Status: {parcel['status']}")
+        
+        if parcel.get("current_location"):
+            response_parts.append(f"🔹 Current Location: {parcel['current_location']}")
+        
+        # Delivery Information
+        if parcel.get("estimated_delivery"):
+            response_parts.append(f"🔹 Estimated Delivery: {parcel['estimated_delivery']}")
+        
+        if parcel.get("actual_delivery_date"):
+            response_parts.append(f"🔹 Delivered On: {parcel['actual_delivery_date']}")
+        
+        # Sender and Recipient Information
+        if parcel.get("sender"):
+            sender = parcel["sender"]
+            if isinstance(sender, dict):
+                sender_info = f"{sender.get('name', 'N/A')}"
+                if sender.get('address'):
+                    sender_info += f" ({sender['address']})"
+                response_parts.append(f"\n🔹 Sender: {sender_info}")
+            else:
+                response_parts.append(f"\n🔹 Sender: {sender}")
+        
+        if parcel.get("recipient"):
+            recipient = parcel["recipient"]
+            if isinstance(recipient, dict):
+                recipient_info = f"{recipient.get('name', 'N/A')}"
+                if recipient.get('address'):
+                    recipient_info += f" ({recipient['address']})"
+                response_parts.append(f"🔹 Recipient: {recipient_info}")
+            else:
+                response_parts.append(f"🔹 Recipient: {recipient}")
+        
+        # Package Details
+        if parcel.get("weight"):
+            response_parts.append(f"\n🔹 Weight: {parcel['weight']}")
+        
+        if parcel.get("dimensions"):
+            response_parts.append(f"🔹 Dimensions: {parcel['dimensions']}")
+        
+        if parcel.get("description"):
+            response_parts.append(f"🔹 Description: {parcel['description']}")
+        
+        # Courier Information
+        if parcel.get("courier_name"):
+            response_parts.append(f"\n🔹 Courier: {parcel['courier_name']}")
+        
+        if parcel.get("vehicle_type"):
+            response_parts.append(f"🔹 Vehicle Type: {parcel['vehicle_type']}")
+        
+        # Tracking History
+        if parcel.get("tracking_history"):
+            history = parcel["tracking_history"]
+            if isinstance(history, list) and len(history) > 0:
+                response_parts.append(f"\n📍 Tracking History:")
+                for idx, event in enumerate(history[-5:], 1):  # Show last 5 events
+                    if isinstance(event, dict):
+                        timestamp = event.get('timestamp', 'N/A')
+                        location = event.get('location', 'N/A')
+                        status = event.get('status', 'N/A')
+                        response_parts.append(f"  {idx}. [{timestamp}] {location} - {status}")
+                    else:
+                        response_parts.append(f"  {idx}. {event}")
+                
+                if len(history) > 5:
+                    response_parts.append(f"  ... and {len(history) - 5} more events")
+        
+        # Additional Notes
+        if parcel.get("notes"):
+            response_parts.append(f"\n📝 Notes: {parcel['notes']}")
+        
+        # Special Instructions
+        if parcel.get("special_instructions"):
+            response_parts.append(f"⚠️ Special Instructions: {parcel['special_instructions']}")
+        
+        # Shipping Method
+        if parcel.get("shipping_method"):
+            response_parts.append(f"\n🔹 Shipping Method: {parcel['shipping_method']}")
+        
+        # Cost Information
+        if parcel.get("shipping_cost"):
+            response_parts.append(f"🔹 Shipping Cost: {parcel['shipping_cost']}")
+        
+        response_text = "\n".join(response_parts)
+        logger.info(f"Successfully retrieved tracking info for: {tracking_number}")
+        return response_text
+        
+    except ConnectionFailure as e:
+        error_msg = f"Failed to connect to MongoDB: {str(e)}. Please check your connection."
+        logger.error(error_msg)
+        return error_msg
+        
+    except OperationFailure as e:
+        error_msg = f"MongoDB operation failed: {str(e)}. Please check your permissions."
+        logger.error(error_msg)
+        return error_msg
+        
+    except Exception as e:
+        error_msg = f"Error tracking parcel '{tracking_number}': {str(e)}"
+        logger.error(error_msg, exc_info=True)
+        return error_msg
+
+@tool(description="Estimate delivery time for a parcel based on delivery parameters.")
+def estimated_time_analysis(
+    distance_km: float,
+    courier_experience_yrs: float = 2.0,
+    vehicle_type: str = "Scooter",
+    weather: str = "Sunny",
+    time_of_day: str = "Morning",
+    traffic_level: str = "Medium"
+) -> str:
+    """
+    Estimate delivery time for a parcel based on various delivery parameters.
+    
+    This tool makes an API call to the ETA prediction service which uses a trained ML model
+    to predict delivery time based on distance, courier experience, vehicle type, weather,
+    time of day, and traffic conditions.
+
+    Args:
+        distance_km: Distance in kilometers (must be positive, max 1000km)
+        courier_experience_yrs: Courier experience in years (0-50, default: 2.0)
+        vehicle_type: Type of vehicle - one of: 'Scooter', 'Pickup Truck', 'Motorcycle' (default: 'Scooter')
+        weather: Weather condition - one of: 'Sunny', 'Rainy', 'Foggy', 'Snowy', 'Windy' (default: 'Sunny')
+        time_of_day: Time of day - one of: 'Morning', 'Afternoon', 'Evening', 'Night' (default: 'Morning')
+        traffic_level: Traffic level - one of: 'Low', 'Medium', 'High' (default: 'Medium')
+
+    Returns:
+        A string describing the estimated delivery time in minutes with the input parameters used.
+        
+    Example:
+        estimated_time_analysis(distance_km=15.5, vehicle_type="Motorcycle", traffic_level="High")
+    """
+    logger.info(f"Estimating delivery time: distance={distance_km}km, vehicle={vehicle_type}, traffic={traffic_level}")
+    
+    try:
+        # Prepare the request payload matching the API schema
+        payload = {
+            "Distance_km": distance_km,
+            "Courier_Experience_yrs": courier_experience_yrs,
+            "Vehicle_Type": vehicle_type,
+            "Weather": weather,
+            "Time_of_Day": time_of_day,
+            "Traffic_Level": traffic_level
+        }
+        
+        # Make API call to ETA prediction service
+        api_url = f"{ETA_API_BASE_URL}/predict"
+        logger.debug(f"Calling ETA API: {api_url} with payload: {payload}")
+        
+        response = requests.post(
+            api_url,
+            json=payload,
+            timeout=ETA_API_TIMEOUT
+        )
+        
+        # Check if request was successful
+        response.raise_for_status()
+        
+        # Parse response
+        result = response.json()
+        predicted_time = result.get("predicted_delivery_time")
+        
+        if predicted_time is None:
+            raise ValueError("No prediction returned from API")
+        
+        # Format response
+        hours = int(predicted_time // 60)
+        minutes = int(predicted_time % 60)
+        
+        time_str = ""
+        if hours > 0:
+            time_str += f"{hours} hour{'s' if hours > 1 else ''}"
+        if minutes > 0:
+            if hours > 0:
+                time_str += " and "
+            time_str += f"{minutes} minute{'s' if minutes != 1 else ''}"
+        
+        response_text = f"""Estimated Delivery Time Analysis:
+- Predicted Time: {predicted_time:.1f} minutes ({time_str})
+- Distance: {distance_km} km
+- Vehicle Type: {vehicle_type}
+- Courier Experience: {courier_experience_yrs} years
+- Weather: {weather}
+- Time of Day: {time_of_day}
+- Traffic Level: {traffic_level}
+
+This prediction is based on a trained machine learning model considering all the above factors."""
+        
+        logger.info(f"ETA prediction successful: {predicted_time:.1f} minutes")
+        return response_text
+        
+    except requests.exceptions.ConnectionError:
+        error_msg = f"Unable to connect to ETA prediction service at {ETA_API_BASE_URL}. Please ensure the service is running."
+        logger.error(error_msg)
+        return error_msg
+        
+    except requests.exceptions.Timeout:
+        error_msg = f"ETA prediction service timed out after {ETA_API_TIMEOUT} seconds."
+        logger.error(error_msg)
+        return error_msg
+        
+    except requests.exceptions.HTTPError as e:
+        error_msg = f"ETA prediction service returned an error: {e.response.status_code} - {e.response.text}"
+        logger.error(error_msg)
+        return error_msg
+        
+    except Exception as e:
+        error_msg = f"Error during ETA estimation: {str(e)}"
+        logger.error(error_msg, exc_info=True)
+        return error_msg
+
+@tool
+def conduct_execution(execution_jobs: str) -> str:
+    """
+    Tool for delegating an execution task to a specialized sub-agent.
+    """
+    return f"Delegated execution job: {execution_jobs}"
+
+
+@tool
+def execution_complete() -> str:
+    """
+    Tool for indicating the execution process is complete.
+    """
+    return "Execution complete."

templates/index.html

@@ -0,0 +1,372 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>Sparrow Agent Chat</title>
+    <style>
+        * {
+            margin: 0;
+            padding: 0;
+            box-sizing: border-box;
+        }
+
+        body {
+            font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
+            background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
+            min-height: 100vh;
+            display: flex;
+            justify-content: center;
+            align-items: center;
+        }
+
+        .chat-container {
+            background: white;
+            border-radius: 20px;
+            box-shadow: 0 20px 40px rgba(0,0,0,0.1);
+            width: 90%;
+            max-width: 800px;
+            height: 80vh;
+            display: flex;
+            flex-direction: column;
+            overflow: hidden;
+        }
+
+        .chat-header {
+            background: linear-gradient(135deg, #4f46e5 0%, #7c3aed 100%);
+            color: white;
+            padding: 20px;
+            text-align: center;
+            position: relative;
+        }
+
+        .chat-header h1 {
+            font-size: 24px;
+            font-weight: 600;
+        }
+
+        .chat-header .subtitle {
+            opacity: 0.9;
+            font-size: 14px;
+            margin-top: 5px;
+        }
+
+        .chat-messages {
+            flex: 1;
+            overflow-y: auto;
+            padding: 20px;
+            background: #f8fafc;
+        }
+
+        .message {
+            margin-bottom: 20px;
+            display: flex;
+            animation: fadeInUp 0.3s ease;
+        }
+
+        @keyframes fadeInUp {
+            from {
+                opacity: 0;
+                transform: translateY(10px);
+            }
+            to {
+                opacity: 1;
+                transform: translateY(0);
+            }
+        }
+
+        .message.user {
+            justify-content: flex-end;
+        }
+
+        .message-content {
+            max-width: 70%;
+            padding: 15px 20px;
+            border-radius: 18px;
+            position: relative;
+        }
+
+        .message.user .message-content {
+            background: linear-gradient(135deg, #4f46e5 0%, #7c3aed 100%);
+            color: white;
+            border-bottom-right-radius: 5px;
+        }
+
+        .message.ai .message-content {
+            background: white;
+            border: 1px solid #e2e8f0;
+            border-bottom-left-radius: 5px;
+            box-shadow: 0 2px 8px rgba(0,0,0,0.05);
+        }
+
+        .message-time {
+            font-size: 11px;
+            opacity: 0.6;
+            margin-top: 8px;
+        }
+
+        .chat-input-container {
+            padding: 20px;
+            background: white;
+            border-top: 1px solid #e2e8f0;
+        }
+
+        .chat-input-form {
+            display: flex;
+            gap: 12px;
+            align-items: flex-end;
+        }
+
+        .chat-input {
+            flex: 1;
+            border: 2px solid #e2e8f0;
+            border-radius: 12px;
+            padding: 15px 20px;
+            font-size: 16px;
+            resize: none;
+            min-height: 50px;
+            max-height: 120px;
+            transition: all 0.2s ease;
+        }
+
+        .chat-input:focus {
+            outline: none;
+            border-color: #4f46e5;
+            box-shadow: 0 0 0 3px rgba(79, 70, 229, 0.1);
+        }
+
+        .send-button {
+            background: linear-gradient(135deg, #4f46e5 0%, #7c3aed 100%);
+            color: white;
+            border: none;
+            border-radius: 12px;
+            padding: 15px 25px;
+            font-size: 16px;
+            font-weight: 600;
+            cursor: pointer;
+            transition: all 0.2s ease;
+            min-width: 80px;
+        }
+
+        .send-button:hover:not(:disabled) {
+            transform: translateY(-2px);
+            box-shadow: 0 8px 25px rgba(79, 70, 229, 0.3);
+        }
+
+        .send-button:disabled {
+            opacity: 0.6;
+            cursor: not-allowed;
+        }
+
+        .loading {
+            display: flex;
+            align-items: center;
+            gap: 10px;
+        }
+
+        .loading-dots {
+            display: flex;
+            gap: 4px;
+        }
+
+        .loading-dot {
+            width: 8px;
+            height: 8px;
+            border-radius: 50%;
+            background: #4f46e5;
+            animation: loadingPulse 1.4s infinite ease-in-out;
+        }
+
+        .loading-dot:nth-child(1) { animation-delay: -0.32s; }
+        .loading-dot:nth-child(2) { animation-delay: -0.16s; }
+
+        @keyframes loadingPulse {
+            0%, 80%, 100% {
+                transform: scale(0.6);
+                opacity: 0.5;
+            }
+            40% {
+                transform: scale(1);
+                opacity: 1;
+            }
+        }
+
+        .error-message {
+            background: #fee2e2;
+            border: 1px solid #fecaca;
+            color: #dc2626;
+            padding: 15px;
+            border-radius: 12px;
+            margin: 10px 0;
+        }
+
+        .status-info {
+            font-size: 12px;
+            color: #64748b;
+            margin-top: 5px;
+            font-style: italic;
+        }
+
+        /* Mobile responsive */
+        @media (max-width: 768px) {
+            .chat-container {
+                width: 95%;
+                height: 90vh;
+                border-radius: 15px;
+            }
+
+            .message-content {
+                max-width: 85%;
+            }
+
+            .chat-input-container {
+                padding: 15px;
+            }
+        }
+    </style>
+</head>
+<body>
+    <div class="chat-container">
+        <div class="chat-header">
+            <h1>Sparrow Logistics Agent</h1>
+            <div class="subtitle">Your Intelligent Assistant</div>
+        </div>
+        
+        <div class="chat-messages" id="messages">
+            <div class="message ai">
+                <div class="message-content">
+                    <div>Hello! I'm Sparrow, your intelligent assistant. I can help you with various tasks and questions. What would you like to know or do today?</div>
+                    <div class="message-time">Just now</div>
+                </div>
+            </div>
+        </div>
+        
+        <div class="chat-input-container">
+            <form class="chat-input-form" id="chatForm">
+                <textarea 
+                    class="chat-input" 
+                    id="messageInput" 
+                    placeholder="Type your message here..." 
+                    rows="1"
+                    required
+                ></textarea>
+                <button type="submit" class="send-button" id="sendButton">
+                    Send
+                </button>
+            </form>
+        </div>
+    </div>
+
+    <script>
+        const messagesContainer = document.getElementById('messages');
+        const chatForm = document.getElementById('chatForm');
+        const messageInput = document.getElementById('messageInput');
+        const sendButton = document.getElementById('sendButton');
+
+        // Auto-resize textarea
+        messageInput.addEventListener('input', function() {
+            this.style.height = 'auto';
+            this.style.height = this.scrollHeight + 'px';
+        });
+
+        // Handle Enter key (Shift+Enter for new line)
+        messageInput.addEventListener('keydown', function(e) {
+            if (e.key === 'Enter' && !e.shiftKey) {
+                e.preventDefault();
+                chatForm.dispatchEvent(new Event('submit'));
+            }
+        });
+
+        function addMessage(content, isUser = false, showTime = true) {
+            const messageDiv = document.createElement('div');
+            messageDiv.className = `message ${isUser ? 'user' : 'ai'}`;
+            
+            const currentTime = new Date().toLocaleTimeString([], {hour: '2-digit', minute:'2-digit'});
+            
+            messageDiv.innerHTML = `
+                <div class="message-content">
+                    <div>${content}</div>
+                    ${showTime ? `<div class="message-time">${currentTime}</div>` : ''}
+                </div>
+            `;
+            
+            messagesContainer.appendChild(messageDiv);
+            messagesContainer.scrollTop = messagesContainer.scrollHeight;
+            
+            return messageDiv;
+        }
+
+        function showLoading() {
+            return addMessage(`
+                <div class="loading">
+                    <span>Sparrow is thinking</span>
+                    <div class="loading-dots">
+                        <div class="loading-dot"></div>
+                        <div class="loading-dot"></div>
+                        <div class="loading-dot"></div>
+                    </div>
+                </div>
+            `, false, false);
+        }
+
+        function setButtonState(loading) {
+            sendButton.disabled = loading;
+            sendButton.textContent = loading ? 'Sending...' : 'Send';
+            messageInput.disabled = loading;
+        }
+
+        chatForm.addEventListener('submit', async function(e) {
+            e.preventDefault();
+            
+            const message = messageInput.value.trim();
+            if (!message) return;
+            
+            // Add user message
+            addMessage(message, true);
+            messageInput.value = '';
+            messageInput.style.height = 'auto';
+            
+            // Show loading
+            setButtonState(true);
+            const loadingMessage = showLoading();
+            
+            try {
+                const response = await fetch('/chat', {
+                    method: 'POST',
+                    headers: {
+                        'Content-Type': 'application/json',
+                    },
+                    body: JSON.stringify({ message: message })
+                });
+                
+                const data = await response.json();
+                
+                // Remove loading message
+                loadingMessage.remove();
+                
+                if (data.success) {
+                    // Add AI response
+                    let responseContent = data.response;
+                    if (data.status) {
+                        responseContent += `<div class="status-info">${data.status}</div>`;
+                    }
+                    addMessage(responseContent);
+                } else {
+                    addMessage(`<div class="error-message">Sorry, I encountered an error: ${data.error}</div>`);
+                }
+                
+            } catch (error) {
+                loadingMessage.remove();
+                addMessage(`<div class="error-message">Sorry, I couldn't connect to the server. Please try again.</div>`);
+                console.error('Error:', error);
+            } finally {
+                setButtonState(false);
+                messageInput.focus();
+            }
+        });
+
+        // Focus input on load
+        messageInput.focus();
+    </script>
+</body>
+</html>