Aria Agentic AI Orchestrator Workflow Engine

🤖 Aria - Advanced AI Agent Chatbot

> A sophisticated multi-utility AI chatbot powered by LangGraph, featuring advanced RAG, intelligent tool calling, MCP integration, and agentic workflow orchestration.

Table of Contents

• [Overview](#overview)
• [Architecture](#architecture)
• [Key Features](#key-features)
• [Core Concepts](#core-concepts)
• [Tech Stack](#tech-stack)
• [Project Structure](#project-structure)
• [Installation](#installation)
• [Usage](#usage)
• [Advanced Features](#advanced-features)
• [LangSmith Observability](#langsmith-observability)
• [Contributing](#contributing)

---

Overview

Aria is an intelligent conversational agent that combines the power of:

• LangGraph: For orchestrating complex AI workflows
• LangChain: For tool management and LLM interactions
• Google Gemini 2.5: State-of-the-art language understanding
• Agentic RAG: Intelligent document retrieval with reasoning
• MCP Protocol: Extensible tool integration via Model Context Protocol
• SQLite: Persistent multi-threaded conversation storage

This chatbot goes beyond simple retrieval—it understands when to use which tool, combines information from multiple sources, and maintains context across dozens of independent conversations.

---

Architecture

System Design

┌─────────────────────────────────────────────────────────┐
│                    Streamlit Frontend                    │
│  (Chat UI, PDF Management, Thread Handling)              │
└────────────────┬────────────────────────────────────────┘
                 │
┌────────────────▼────────────────────────────────────────┐
│              LangGraph State Machine                      │
│  ┌──────────────────────────────────────────────────┐   │
│  │ START → chat_node → tools_condition             │   │
│  │           ↑                    ↓                 │   │
│  │           └─── tool_node ←────┘                 │   │
│  └──────────────────────────────────────────────────┘   │
└────────────────┬────────────────────────────────────────┘
                 │
    ┌────────────┼────────────┐
    │            │            │
┌───▼──┐  ┌─────▼──┐  ┌──────▼──────┐
│ LLM  │  │ Tools  │  │ Storage     │
│      │  │        │  │ (SQLite)    │
└──────┘  └────────┘  └─────────────┘

Request Flow

1. User Input → Streamlit frontend
2. Thread Routing → SQLite retrieves conversation history
3. LLM Processing → Gemini determines which tools to use
4. Tool Execution → Parallel/sequential tool invocation
5. Response Assembly → Stream response back to user
6. Storage → Persist conversation to database

---

Key Features

🧠 1. Agentic RAG (Retrieval-Augmented Generation)

Traditional RAG:

Query → Retrieve → Generate

Aria's Agentic RAG:

Query → Analyze → Decide Retrieval Strategy → Retrieve → Reason → Generate

What makes it "Agentic":

• Intelligent Routing: LLM decides whether RAG is needed or if other tools are better
• Multi-step Reasoning: Can combine document insights with web search or calculations
• Adaptive Chunking: 1000-character chunks with 200-char overlap for semantic cohesion
• Semantic Search: Uses Google's embeddings (models/embedding-001) for similarity matching

How it works:

1. User uploads a PDF → Text extraction & splitting
2. Documents are embedded using Google's embedding model
3. FAISS index created for fast similarity search
4. When queried, LLM receives top-4 relevant chunks
5. LLM decides: use RAG info + search + calculate + track expenses?

---

🔄 2. Tool Calling System

7 Integrated Tools:

| Tool | Purpose | Provider | Type | |------|---------|----------|------| | DuckDuckGo Search | Real-time web information | DuckDuckGo | External API | | Stock Prices | Financial data retrieval | Alpha Vantage | External API | | Calculator | Math operations | Native | Built-in | | RAG Tool | Document Q&A | FAISS + Embeddings | Internal | | Add Expense | Track spending | MCP Server | Remote Protocol | | List Expenses | View transaction history | MCP Server | Remote Protocol | | Summarize | Generate expense summaries | MCP Server | Remote Protocol |

Tool Calling Flow:

LLM Decision:
├─ If ("search" in intent) → DuckDuckGo_Search()
├─ If ("stock" in intent) → Get_Stock_Price()
├─ If ("math" in intent) → Calculator()
├─ If ("document" in intent) → RAG_Tool()
├─ If ("expense" in intent) → MCP_Expense_Tools()
└─ If ("none" in intent) → Direct_Response()

Key Technologies:

• LangChain StructuredTool: Type-safe tool definitions
• Tool Binding: LLM knows function signatures, parameters, return types
• Conditional Edges: LangGraph routes based on LLM output
• Async Wrapping: MCP async tools converted to sync for compatibility

---

🔌 3. MCP (Model Context Protocol) Integration

What is MCP? MCP is a standardized protocol for connecting AI models with external tools and data sources. It's like an "API standard" for AI assistants.

Aria's MCP Setup:

Aria Chatbot ←→ MCP Client ←→ Expense Tracker Server
              (stdio transport)

Local vs Remote:

• ❌ Remote Setup Failed: 401 Unauthorized from remote endpoint
• ✅ Local Setup Success: Python subprocess stdio transport

Architecture:

# Local MCP Expense Tracker
MCP Service:
├─ add_expense(amount, date, category)
├─ list_expenses(filters)
└─ summarize(period)

Transport: Stdio (stdin/stdout pipes)
Async Handling: StructuredTool wrapper for sync execution
Threading: Separate event loop for async operations

Benefits:

• Extensibility: Add any MCP server without modifying core code
• Isolation: Tools run in separate processes
• Standardization: Same API whether local or remote

---

� 6. LangSmith Integration & Observability

What is LangSmith?

LangSmith is a unified platform for debugging, testing, and monitoring AI applications. It provides complete visibility into your LLM calls, tool usage, latency, and failures.

Aria's LangSmith Integration:

Every chat interaction is automatically traced and logged to LangSmith for real-time monitoring and debugging.

What Gets Traced:

Chat Turn Trace:
├─ User Input
│  ├─ Message content
│  ├─ Thread ID
│  └─ Message count
├─ Processing
│  ├─ LLM invocation (Gemini 2.5)
│  ├─ Tool decisions & calls
│  └─ Tool execution results
└─ Output
   ├─ AI response
   ├─ Response length
   ├─ Tools used
   └─ Execution timing

Key Metrics Captured:

| Metric | Purpose | Use Case | |--------|---------|----------| | Response Time | Track LLM latency | Identify slow responses | | Tools Used | Monitor tool routing | Understand agent behavior | | Token Count | Measure input/output size | Cost analysis | | Error Tracking | Log failures | Debug issues | | Thread ID | Link to conversation | Trace user sessions | | Timestamp | Record exact time | Performance trending |

LangSmith Dashboard Views:

1. Traces Tab: Complete execution flow for each chat turn

• Visual hierarchy of LLM calls, tool calls, and responses
• Input/output for each node
• Timing information

1. Metrics Tab: Aggregated performance data

• Average response time per conversation
• Tool usage frequency
• Error rates and types

1. Debugging: Deep-dive into specific traces

• See exact prompts sent to LLM
• Tool execution details
• Error messages and stack traces

Setup:

# Add to .env file
LANGCHAIN_TRACING_V2=true
LANGCHAIN_PROJECT=Advance LangGraph project
LANGCHAIN_API_KEY=
LANGCHAIN_ENDPOINT=https://api.smith.langchain.com

Get your API key from: https://smith.langchain.com/settings/org/api-keys

How It Works:

# Automatic tracing happens in chatbot_stream_with_tracing()
@traceable(name="Chat Turn", tags=["chat", "stream"])
def traced_stream():
    # All code here is automatically traced
    for chunk in chatbot.stream(input_dict, config=config):
        yield chunk

Every interaction flows through LangSmith's tracer, creating a complete audit trail of:

• What the user asked
• How the agent reasoned about tools
• Which tools were called and why
• What data was retrieved/computed
• How the final response was generated

Use Cases:

1. Real-time Monitoring: Watch trace execution as users chat
2. Performance Analysis: Identify bottlenecks (slow tools, token usage)
3. Debugging: Replay failed traces to understand what went wrong
4. Cost Tracking: Monitor LLM token usage and API calls
5. Quality Assurance: Audit agent behavior patterns
6. User Behavior: Understand which features users interact with

Example Scenario:

User: "Based on my uploaded document, find instances where costs exceeded $10,000"

LangSmith Trace Shows:

• LLM received: user query + system prompt + document chunks (RAG)
• LLM decision: "Need to use RAG tool to search document"
• Tool executed: ragtool("costs exceeded 10000", threadid=...)
• Tool returned: 3 matching sections from document
• LLM regenerated: Final response with filtered results
• Trace metadata: responsetime=2.3s, toolsused=["ragtool"], tokencount=1240

Features:

✅ Automatic tracing - No code changes needed, works out of the box ✅ Tool tracking - See which tools called and their execution time ✅ Error capture - Failures are automatically logged with context ✅ Session continuity - Thread IDs link traces to conversations ✅ No data leakage - API calls are secure, private to your org

Dashboard Sample:

Project: Advance LangGraph project
├─ Today
│  ├─ Chat Turn (2:45 PM) - Response: "The document shows..." | Tools: rag_tool | Time: 2.3s
│  ├─ Chat Turn (2:41 PM) - Response: "AAPL is trading at..." | Tools: stock_price | Time: 1.2s
│  ├─ Chat Turn (2:38 PM) - Response: "47 * 32 = 1504" | Tools: calculator | Time: 0.5s
│  └─ Chat Turn (2:35 PM) - Response: "Here's what I found..." | Tools: duckduckgo_search | Time: 3.8s
└─ Last 7 days
   └─ (Historical data...)

For detailed configuration and troubleshooting, see [LANGSMITHFIX.md](LANGSMITHFIX.md).

---

�💬 4. Thread Management (Multi-Conversation Support)

What are Threads?

Threads enable independent conversation contexts. Each thread has:

• Unique UUID identifier
• Isolated conversation history
• Separate document index (FAISS)
• Dedicated metadata storage

Per-Thread Storage:

Database (SQLite):
├─ Thread A (UUID: 22258138...)
│  ├─ Message 1: "Hi, analyze my budget"
│  ├─ Message 2: "Create expense tracking"
│  └─ Metadata: {docs_indexed: 1, chunks: 245}
│
├─ Thread B (UUID: 8f9a1d2c...)
│  ├─ Message 1: "Search latest AI news"
│  └─ Message 2: "What about LangGraph?"
│
└─ Thread C (UUID: 7c4e5b9a...)
   ├─ Message 1: "Calculate 500 * 2.5"
   └─ ...

Implementation:

• SQLite Checkpointer: langgraph_checkpoint_sqlite persists state
• Thread ID Config: Passed to every invocation via configurable["thread_id"]
• Per-Thread Retrievers: _THREAD_RETRIEVERS dict maintains separate FAISS indexes
• Conversation Isolation: Messages in Thread A don't affect Thread B

Thread Operations:

# Create new thread
thread_id = generate_thread_id()

# Load existing thread
state = chatbot.get_state(config={"configurable": {"thread_id": thread_id}})

# Continue conversation
for msg in chatbot.stream({"messages": [HumanMessage(...)]}, 
                          config={"configurable": {"thread_id": thread_id}}):
    process(msg)

---

📚 5. Database Architecture

SQLite with LangGraph Checkpointer

Schema (Auto-managed by langgraph-checkpoint-sqlite):

threads:
├─ thread_id (primary key)
├─ checkpoint_data (serialized state)
└─ timestamp

Type structure (internal):
├─ messages: List[Union[HumanMessage, AIMessage, ToolMessage]]
└─ metadata: Dict[str, Any]

Thread-Local Storage:

# Memory storage (session-based)
_THREAD_RETRIEVERS: Dict[str, Any]  # FAISS indices per thread
_THREAD_METADATA: Dict[str, dict]   # Document info per thread

# Persistent storage (database)
SQLiteSaver checkpointer               # Message history

Persistence Flow:

1. User sends message
2. LLM generates response + tool calls
3. Tools execute
4. State updated: messages + tool outputs
5. SQLiteSaver checkpoints to database
6. Next request loads from checkpoint

Why SQLite?

• ✅ Simple, no external DB needed
• ✅ Thread-safe (aiosqlite for async)
• ✅ Persistent across restarts
• ✅ Perfect for single-machine deployment
• ✅ Easy migration to PostgreSQL later

---

Core Concepts

Concept 1: StateGraph (LangGraph)

What it is: A directed acyclic graph (DAG) that defines AI workflow execution.

Aria's Graph:

graph = StateGraph(ChatState)
graph.add_node("chat_node", chat_node)           # LLM invocation
graph.add_node("tools", tool_node)               # Tool execution
graph.add_edge(START, "chat_node")               # Entry point
graph.add_conditional_edges(                     # Smart routing
    "chat_node", 
    tools_condition,  # Decides: call tools or finish?
    {
        "continue": "tools",
        END: END
    }
)
graph.add_edge("tools", "chat_node")            # Loop back

Execution Trace:

START
  ↓
chat_node → [LLM decides: need tools?]
  ↓
tools_condition → {
    No tools → END (return response)
    Tools needed → tools
}
  ↓
tools → Execute all tools in parallel
  ↓
chat_node → [LLM processes tool outputs]
  ↓
(loop until no more tools needed)

Concept 2: add_messages Reducer

Problem: How do we merge tool outputs back into the message history?

Solution: Reducer function

messages: Annotated[list[BaseMessage], add_messages]

What it does:

Old messages:    [User, Assistant, ToolMessage]
New messages:    [ToolMessage, ToolMessage]
Result:          [User, Assistant, ToolMessage, ToolMessage, ...]

(Automatically merges instead of replacing)

Concept 3: Streaming

Why Stream?

• ✅ Real-time token-by-token display
• ✅ Better UX (doesn't feel frozen)
• ✅ Faster apparent response time
• ✅ Can cancel long responses

Aria's Streaming:

for message_chunk, _ in chatbot.stream(
    {"messages": [HumanMessage(content=user_input)]},
    config=CONFIG,
    stream_mode="messages"
):
    if isinstance(message_chunk, AIMessage):
        yield message_chunk.content  # Stream to Streamlit

---

Tech Stack

Core Frameworks

• LangGraph 1.1.6: AI workflow orchestration
• LangChain 1.2.15: LLM abstractions & tools
• Streamlit 1.48: Web UI framework
• SQLite + aiosqlite: Persistent storage

AI/ML Components

• Google Generative AI: Gemini 2.5 Flash Lite LLM + embedding model
• FAISS: Vector similarity search (Facebook AI Similarity Search)
• Sentence Transformers: Embeddings pipeline

Observability & Monitoring

• LangSmith: AI application debugging, testing, and monitoring
• LangSmith Traces: Real-time execution tracking
• LangSmith Metrics: Performance and cost analytics

Data Processing

• PyPDF: PDF text extraction
• LangChain Text Splitters: Semantic chunking (1000 chars, 200 overlap)
• HuggingFace Hub: Model downloading

External Integrations

• DuckDuckGo Search (ddgs 9.13.0): Web search
• Alpha Vantage API: Stock price data
• MCP Protocol (fastmcp 3.2.0): Standard tool interface
• Model Context Protocol (mcp 1.27.0): Protocol support

Development Tools

• Python 3.12.2: Language version
• Virtual Environment (venv): Dependency isolation

---

Project Structure

Aria/
├── frontend.py                 # Streamlit UI (300+ lines)
│   ├── Sidebar: Thread management, PDF upload
│   ├── Main: Chat interface, features showcase
│   └── CSS: Custom styling
│
├── langgraph_backend.py        # Core orchestration (700+ lines)
│

…

## Source & license

This open-source MCP server is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [AbhaySingh71](https://github.com/AbhaySingh71)
- **Source:** [AbhaySingh71/Aria-Agentic-AI-Orchestrator-Workflow-Engine](https://github.com/AbhaySingh71/Aria-Agentic-AI-Orchestrator-Workflow-Engine)
- **License:** MIT

Install and usage instructions live in the source repository linked above.