AI Flow — Setup, Architecture, and Operations Guide

This project is an agentic RAG application with:

• A Streamlit chat interface
• A LangGraph agent and router
• Tavily and SerpAPI web search tools
• Composio agent toolkits (Jira, Linear, Gmail, Google Calendar, Notion, GitHub, Typeform, Apollo, Todoist)
• LlamaCloud/LlamaParse document parsing
• Pinecone semantic retrieval
• BM25 lexical retrieval
• Neo4j knowledge-graph storage and visualization
• SQLite conversation history
• NeMo Guardrails input validation and pre-check greetings bypass
• Human approval before saving a turn to long-term graph/vector memory

---

🤖 The Three Agents & Integration Details

This project contains three independent agentic workflows with specific design purposes, interfaces, and integration components:

1. RAG Agent

• Entrypoint File: src/agents/rag_agent.py
• UI Interface: ui/app.py (Runs on port 8501)
• Core Tech Stack: LangGraph (routing & node state), Pinecone (dense vector search), BM25 (lexical search), Neo4j (Knowledge Graph constraints & Cypher execution), SQLite (conversation storage), spacy (NER & relationship extraction).
• 🔌 Integrated Tools:
  • Tavily & SerpAPI for real-time web searches.
• 🛡️ NVIDIA NeMo Guardrails:
  • YES, used ONLY in this agent.
  • Pre-checks incoming queries against semantic intent rules in disallowed.co and config.yml.
  • Blocks disallowed topics (cooking, coding, system overrides, violence) instantly before executing LangGraph nodes.
  • Combines with a local fast-path greeting filter in src/utils/guardrails.py to save LLM / API token costs on simple hellos/greetings.
• 📄 LlamaCloud (LlamaParse):
  • YES, used ONLY in this agent & ingestion pipeline.
  • Powers high-fidelity document parsing for uploaded files (PDF, DOCX, etc.) both in the chat UI upload and in the background ingestion pipeline (src/pipelines/ingestion.py).
  • Implements SHA-256 checksum caching (data/parsed/<sha256>.md) to avoid duplicate API calls.

2. Composio Agent

• Entrypoint File: src/agents/composio_agent.py
• UI Interface: ui/composio-agent.py (Runs on port 8502)
• Core Tech Stack: LangGraph ReAct loop (create_react_agent), OpenAI GPT-4o-mini, Composio client.
• 🔌 Integrated Tools: Composio Toolkits (GitHub, Jira, Linear, Slack, Gmail, Google Calendar, Notion, Typeform, Apollo, Todoist, Reddit, LinkedIn, Google Meet, Google Docs) linked to the user's sidebar OAuth sessions.
• 🛡️ NVIDIA NeMo Guardrails: NO, not used in this agent.
• 📄 LlamaCloud (LlamaParse): NO, not used in this agent.

3. Workflow Agent

• Entrypoint File: src/workflows/agent.py
• UI Interface: src/workflows/app.py (Runs on port 8503)
• Core Tech Stack: LangGraph custom StateGraph (Agent node + Verifier node), OpenAI, Composio client.
• 🔌 Integrated Tools: Composio Meta-Tools (COMPOSIO_SEARCH_TOOLS & COMPOSIO_GET_TOOL_SCHEMAS) to dynamically query action parameters and build execution schemas.
• Architecture: Decouples AI Workflow Design (searching for action schemas using OpenAI and building the workflow structure) from Workflow Execution (pure, deterministic python calls via the Composio SDK, requiring zero AI calls).
• 🛡️ NVIDIA NeMo Guardrails: NO, not used in this agent.
• 📄 LlamaCloud (LlamaParse): NO, not used in this agent.

---

1. Requirements

• Python >=3.12,<3.14
• Poetry
• API credentials configured in .env
• Pinecone and Neo4j instances
• Windows Users: Visual Studio C++ Build Tools (specifically the Desktop development with C++ workload) is required to compile NeMo Guardrails dependencies (e.g., hnswlib).

To install dependencies and lock environment:

poetry lock
poetry install

---

2. Environment Configuration

Create a .env file in the repository root:

OPENAI_API_KEY=your_openai_api_key
TAVILY_API_KEY=your_tavily_api_key
SERPAPI_API_KEY=your_serpapi_api_key
LLAMA_CLOUD_API_KEY=your_llamacloud_api_key
COMPOSIO_API_KEY=your_composio_api_key

PINECONE_API_KEY=your_pinecone_key
PINECONE_INDEX_NAME=agentic-system

NEO4J_URI=bolt://localhost:7687
NEO4J_USERNAME=neo4j
NEO4J_PASSWORD=your_password
NEO4J_DATABASE=neo4j

---

3. Running the Applications

3.1 RAG Agent (Main UI)

Runs on port 8501 by default.

poetry run streamlit run ui/app.py

Open: http://localhost:8501

3.2 Composio Agent UI

Runs on port 8502.

poetry run streamlit run ui/composio-agent.py --server.port 8502

Open: http://localhost:8502

3.3 Workflow Agent UI

Runs on port 8503 (or any free port).

poetry run streamlit run src/workflows/app.py --server.port 8503

Open: http://localhost:8503

---

4. Project Structure

ai_flow/
├── dags/
│   └── document_ingestion_dag.py
├── data/
│   ├── uploads/                 # Safe content-addressed uploads; gitignored
│   ├── parsed/                  # Cached LlamaCloud markdown; gitignored
│   ├── conversations.db         # Chat history; gitignored
│   └── bm25_index.pkl           # Local lexical index; gitignored
├── src/
│   ├── agents/
│   │   ├── rag_agent.py         # Main RAG agent logic
│   │   ├── composio_agent.py    # Composio integrations agent
│   │   └── state.py             # LangGraph state configurations
│   ├── guardrails/
│   │   ├── config.yml           # NeMo Guardrails configuration
│   │   └── disallowed.co        # Colang disallowed intents and flows
│   ├── pipelines/
│   │   └── ingestion.py         # Background document ingestion pipeline
│   ├── tools/
│   │   ├── conversation_store.py
│   │   └── search.py
│   ├── utils/
│   │   ├── entity_extractor.py
│   │   ├── graph_store.py
│   │   ├── guardrails.py        # Pre-check greetings & NeMo Guardrails helper functions
│   │   ├── hybrid_search.py
│   │   ├── keyword_search.py
│   │   ├── parser.py            # LlamaParse execution helper
│   │   ├── uploads.py           # Safe uploads service
│   │   └── vector_store.py
│   └── workflows/
│       ├── __init__.py          # Package marker
│       ├── agent.py             # Custom workflow designer agent
│       ├── app.py               # Streamlit UI for workflow agent
│       └── schema.py            # Pydantic schemas for workflows
├── tests/
│   ├── test_agents.py
│   ├── test_knowledge_graph_extraction.py
│   ├── test_rag_utilities.py
│   ├── test_uploads_and_history.py
│   └── test_vector_store_index.py
├── ui/
│   ├── app.py                   # RAG Agent Streamlit interface
│   ├── composio-agent.py        # Composio Agent Streamlit interface
│   └── pages/
│       └── relationships.py     # Neo4j relationship graph visualization
└── pyproject.toml

---

5. Implementation & Integration Details

5.1 NeMo Input Guardrails

We implemented NVIDIA NeMo Guardrails (v0.22.0) to check and sanitise incoming user queries before they reach the main LangGraph agent.

Installation & Prerequisites

NeMo Guardrails is specified in pyproject.toml and is automatically installed when you run poetry install. However, on Windows, installing it has specific compilation requirements:

1. C++ Build Tools Requirement: NeMo Guardrails installs underlying packages like hnswlib (a C++ vector database binding) and others that compile native C++ extensions during installation.

   • Error Symptoms: Without C++ compilers, installation will fail with errors such as error: Microsoft Visual C++ 14.0 or greater is required. or compilation errors for hnswlib.
   • Solution: You must install the Build Tools for Visual Studio (Visual Studio 2022 Build Tools or newer).
     • Download from Visual Studio Build Tools.
     • During the installation, select the Desktop development with C++ workload.
     • Ensure the MSVC v143 - VS 2022 C++ x64/x86 build tools and Windows 11 SDK (or Windows 10 SDK) components are checked.

2. Python Version: NeMo Guardrails has strict dependencies and requires Python >=3.12,<3.14.

3. Active Loops Setup: It requires nest_asyncio to run within Streamlit or other environments with active asyncio loops. This is automatically handled dynamically in src/utils/guardrails.py.

How It Works

NeMo Guardrails coordinates validation flows using a combination of configuration files, semantic embedding search, and Colang script logic:

1. Configuration (src/guardrails/config.yml):
   • Defines the engines used: gpt-4o-mini for the LLM-based rails and text-embedding-3-small for generating sentence embeddings.
   • Restricts Dialog flows by setting embeddings_only: true with a similarity_threshold of 0.82.
   • Specifies a fallback intent (unhandled_user_intent) if the query does not map to any defined intent.
2. Colang Intent Mapping (src/guardrails/disallowed.co):
   • Maps standard user query variations to specific disallowed intents:
     • user ask cooking recipe: Intercepts cooking/food instructions.
     • user ask developer mode: Intercepts system prompts, developer simulations, system overrides, and jailbreak attempts.
     • user ask python code: Intercepts general coding requests.
     • user ask violence: Intercepts safety-critical queries.
   • Defines a standard refusal response:
     • bot refuse request: "I cannot help you with that request. I am designed to assist only with technical documentation and tech-related web searches, not for cooking, coding, developer simulation, or harmful queries."
   • Defines dialog flows connecting each disallowed intent to the refusal action (e.g., if a user asks a cooking recipe, the bot replies with the refusal response).
3. Semantic Matching: When a query is checked, NeMo Guardrails calculates its embedding and checks the similarity against the canonical examples in the .co file. If the similarity meets or exceeds 0.82, it triggers the associated block flow.

How It Is Used in the Codebase

• Pre-Check Hook: Before the query initiates the LangGraph compiler and states in src/agents/rag_agent.py, run_agent() triggers pre_check_query().
• Fast-Path Greeting Filter: Before making any API requests, src/utils/guardrails.py performs a regex check against GREETINGS_MAP. If a greeting is matched, the predefined reply is returned instantly. This prevents latency and avoids any LLM / OpenAI token costs.
• Guardrails Invocation: If it's a standard user query, check_input_guardrails(query) is called:
  • It temporarily injects NEMO_API_KEY (configured in src/utils/guardrails.py) as the OPENAI_API_KEY environment variable.
  • It retrieves the cached LLMRails instance from get_rails_instance().
  • It calls rails.generate() with the user query.
  • It inspects the returned string. If the response contains any of the known refusal markers (such as "I cannot help you with that request" or "I am designed to assist only with technical documentation"), the helper marks the query as blocked and returns the refusal message.
  • The refusal message is saved to the SQLite conversation table using record_conversation_turn(), and the execution terminates immediately, short-circuiting the LangGraph router.
• Fail-Safe Fallback: If an exception or connection failure occurs during the NeMo check, the exception is caught, logged, and the query is allowed through to prevent disrupting the user flow.

---

5.2 LlamaCloud / LlamaParse Ingestion

We use LlamaParse to parse documents (PDFs, Docx, images, etc.) into high-quality structured markdown before chunking and embedding.

• Deterministic Checksum Cache: Before sending a file to LlamaCloud, we compute its SHA-256 checksum and save the parsed markdown in data/parsed/<sha256>.md. If the file is uploaded again or processed in a background job, we reuse the cached markdown instead of making another paid API request to LlamaCloud.
• Chunking: The markdown is chunked using the LlamaIndex SentenceSplitter.
• Vector Ingestion: Chunks are embedded using OpenAI embeddings and stored in Pinecone.
• Knowledge Graph Ingestion: Chunks are processed via spacy to extract entities/relationships and merged into Neo4j using Cypher.
• Lexical Index Ingestion: Chunks are indexed locally using a rank-bm25 index cached in data/bm25_index.pkl.

---

6. Durable Workflows with Apache Airflow

Airflow is not the chat server and should not run each user question. Use Airflow for durable, observable background workflows such as scheduled document ingestion, nightly re-indexing, rebuilding BM25, and knowledge-graph maintenance.

Practising Airflow Locally

Apache Airflow does not natively support Windows. Use WSL2 or Docker Desktop.

WSL2 stand-alone execution:

Inside Ubuntu/WSL:

export AIRFLOW_HOME=~/airflow
airflow standalone

Open: http://localhost:8080

Link this project's dags folder into $AIRFLOW_HOME/dags.

Useful learning commands:

airflow version
airflow dags list
airflow dags list-import-errors
airflow dags trigger document_ingestion_dag

---

7. Tests & Verification

Run the test suite using pytest:

poetry run pytest tests -q