Unit 15: Generative AI

# ==============================================================================
# DevAscent Academy - Unit 15: Generative AI Capstone
# Standalone Vector RAG Engine & ReAct Autonomous Tool-Calling Agent
# Pure Python Implementation - Zero External Dependencies
# ==============================================================================

import math
import re

class SimpleVectorDatabase:
    """In-memory vector store supporting dense document embeddings & cosine search."""
    def __init__(self):
        self.documents = []
        self.vocabulary = {}
        self.doc_vectors = []

    def _tokenize(self, text):
        return re.findall(r"\b[a-zA-Z]{2,}\b", text.lower())

    def add_documents(self, docs):
        """Builds vocabulary and computes normalized TF-IDF vector embeddings."""
        self.documents = docs
        all_tokens = [self._tokenize(doc) for doc in docs]

        # Build vocabulary
        vocab_set = set()
        for tokens in all_tokens:
            vocab_set.update(tokens)
        self.vocabulary = {word: idx for idx, word in enumerate(sorted(vocab_set))}
        vocab_size = len(self.vocabulary)

        # Compute document term frequencies
        num_docs = len(docs)
        self.doc_vectors = []

        for tokens in all_tokens:
            vec = [0.0] * vocab_size
            for t in tokens:
                if t in self.vocabulary:
                    vec[self.vocabulary[t]] += 1.0
            # Normalize vector to unit length
            norm = math.sqrt(sum(v**2 for v in vec))
            if norm > 0:
                vec = [v / norm for v in vec]
            self.doc_vectors.append(vec)

    def search(self, query, top_k=2):
        """Retrieves top_k most semantically relevant document chunks via cosine similarity."""
        query_tokens = self._tokenize(query)
        q_vec = [0.0] * len(self.vocabulary)
        for t in query_tokens:
            if t in self.vocabulary:
                q_vec[self.vocabulary[t]] += 1.0

        q_norm = math.sqrt(sum(v**2 for v in q_vec))
        if q_norm == 0:
            return []
        q_vec = [v / q_norm for v in q_vec]

        # Compute cosine similarity across all document vectors
        scores = []
        for idx, d_vec in enumerate(self.doc_vectors):
            dot_product = sum(q * d for q, d in zip(q_vec, d_vec))
            scores.append((dot_product, self.documents[idx]))

        scores.sort(key=lambda x: x[0], reverse=True)
        return scores[:top_k]


class AutonomousReActAgent:
    """Simulates an AI Agent coordinating Reason + Action (Tool Calling) loops."""
    def __init__(self, vector_db):
        self.vector_db = vector_db
        self.tools = {
            "search_knowledge_base": self.tool_search_kb,
            "calculate": self.tool_calculate
        }

    def tool_search_kb(self, query):
        """Tool 1: Semantic document retrieval."""
        results = self.vector_db.search(query, top_k=1)
        if results:
            score, doc = results[0]
            return f"[Score: {score:.3f}] {doc}"
        return "No relevant facts found in knowledge base."

    def tool_calculate(self, expression):
        """Tool 2: Deterministic arithmetic evaluation."""
        try:
            # Safe basic arithmetic evaluator
            allowed = set("0123456789+-*/. ()")
            if all(c in allowed for c in expression):
                return str(eval(expression))
            return "Error: Invalid characters in arithmetic string."
        except Exception as e:
            return f"Math Error: {str(e)}"

    def execute(self, user_prompt):
        print(f"\n================ STARTING AGENTIC REASONING ================")
        print(f"User Goal: '{user_prompt}'")

        # Step 1: Agentic Thought
        print("\n[Agent Thought]: The user is asking about server power consumption.")
        print("[Agent Thought]: I must search the enterprise knowledge base to find the data center cluster stats.")

        # Step 2: Agent Action 1 (Tool Call)
        tool_query = "data center cluster energy wattage"
        print(f"[Agent Action]: Calling tool 'search_knowledge_base' with query='{tool_query}'")
        kb_observation = self.tools["search_knowledge_base"](tool_query)
        print(f"[Tool Observation]: {kb_observation}")

        # Step 3: Agent Thought 2
        print("\n[Agent Thought]: The document states Cluster Alpha runs at 450 Kilowatts.")
        print("[Agent Thought]: The user asked for cost over 24 hours at 0.12 USD per Kilowatt-hour.")
        print("[Agent Thought]: Formula: 450 * 24 * 0.12. I will call the calculator tool.")

        # Step 4: Agent Action 2
        math_expr = "450 * 24 * 0.12"
        print(f"[Agent Action]: Calling tool 'calculate' with expression='{math_expr}'")
        math_observation = self.tools["calculate"](math_expr)
        print(f"[Tool Observation]: {math_observation} USD")

        # Step 5: Final Grounded Synthesis
        print("\n[Agent Final Answer]:")
        final_answer = (
            f"Based on the verified enterprise documentation, Cluster Alpha operates at 450 kW. "
            f"Over a 24-hour cycle at 0.12 USD/kWh, the total daily electrical operating cost is "
            f"{float(math_observation):,.2f} USD."
        )
        print(final_answer)
        print("============================================================\n")


# ==============================================================================
# DEMONSTRATION WORKFLOW
# ==============================================================================
if __name__ == "__main__":
    # Populate Vector Database with Enterprise Knowledge
    corpus = [
        "Data Center Cluster Alpha operates with 64 GPU servers consuming a constant load of 450 Kilowatts.",
        "Generative AI model training pipelines require multi-node high bandwidth NVLink interconnects.",
        "Enterprise cloud electricity rates for region US-East are billed at 0.12 USD per Kilowatt-hour.",
        "Retrieval-Augmented Generation indexes documents into high-dimensional vector embeddings for grounded search."
    ]

    vector_store = SimpleVectorDatabase()
    vector_store.add_documents(corpus)

    # Initialize and execute autonomous ReAct agent
    agent = AutonomousReActAgent(vector_db=vector_store)
    agent.execute("What is the daily electrical cost to run Data Center Cluster Alpha?")