feat: implement AutoStream conversational AI sales agent with LangGraph

- Implements a stateful agent workflow graph using LangGraph
- Sets up an LLM-based intent classifier with structured outputs
- Implements a local FAISS-based RAG pipeline
- Includes a step-by-step lead qualification workflow and a mock backend tool execution
- Provides a CLI interface in main.py
- Creates a comprehensive testing suite mocking LLMs and Embeddings via pytest
- Adds thorough documentation on system architecture and integration capabilities

Co-authored-by: archc0der <119496494+archc0der@users.noreply.github.com>

.gitignore

@@ -0,0 +1,3 @@
+__pycache__/
+*.pyc
+.env

README.md

@@ -0,0 +1,73 @@
+# AutoStream Conversational AI Agent
+
+## Project Overview
+This project is a production-quality Conversational AI Agent built for **AutoStream**, a fictional SaaS company. It handles customer inquiries, answers product questions using a Knowledge Base (RAG), and detects high-intent users to seamlessly collect lead information and execute backend lead capture functions.
+
+## System Architecture
+The system is designed as an agentic workflow using **LangGraph**, replacing traditional linear chatbots with a stateful, branching graph architecture.
+1. **User Input & State Management**: User messages and conversational context are persisted in a shared `AgentState` that tracks details like intent, history, and collected lead fields.
+2. **Intent Classification**: Using `gpt-4o-mini` with structured output, the agent categorizes messages (e.g., GREETING, PRICING_QUERY, HIGH_INTENT_LEAD).
+3. **Routing**: A conditional edge acts as a router, directing the conversation to specialized nodes based on intent.
+4. **Knowledge Retrieval**: Product and pricing questions are routed to a RAG pipeline that retrieves context from a FAISS vector store.
+5. **Lead Qualification**: High-intent users are routed to a multi-turn lead collection workflow. The agent selectively asks for missing fields (Name, Email, Creator Platform).
+6. **Tool Execution**: Once all fields are collected, the agent safely executes a simulated backend lead-capture tool.
+
+## Running Locally
+
+### Prerequisites
+- Python 3.9+
+- An OpenAI API Key
+
+### Setup
+1. Clone this repository.
+2. Install dependencies:
+   ```bash
+   pip install -r requirements.txt
+   ```
+3. Set your OpenAI API key in the environment or create a `.env` file at the root of the project:
+   ```env
+   OPENAI_API_KEY=your_openai_api_key_here
+   ```
+
+### Running the CLI Agent
+To interact with the conversational agent:
+```bash
+python main.py
+```
+
+### Running the Tests
+The project features a full automated testing suite that runs completely without API keys, as all LLM and embedding calls are securely mocked.
+```bash
+pytest
+```
+
+## RAG Pipeline (Retrieval-Augmented Generation)
+When the user asks a product or pricing question, the agent utilizes a RAG pipeline:
+1. The `data/knowledge_base.md` is loaded and chunked using a `RecursiveCharacterTextSplitter`.
+2. Chunks are embedded using `OpenAIEmbeddings` and indexed into a local `FAISS` vector database.
+3. The retriever fetches the top `k` relevant chunks for the user's query and injects them into the RAG generation prompt.
+4. The LLM generates a well-grounded response strictly based on the retrieved context.
+
+## Lead Capture Workflow
+For users expressing a desire to purchase or sign up, the intent classifier triggers `HIGH_INTENT_LEAD`.
+The workflow then shifts to `process_lead`. The system relies on structured extraction to glean fields (Name, Email, Creator Platform) from incoming text. It incrementally prompts the user over several turns until all required fields are collected, effectively pausing the LangGraph execution between inputs.
+
+## State Management
+A `TypedDict` named `AgentState` tracks the overarching conversation context. This prevents duplicate questions and provides memory. State variables include `conversation_history` (up to 6 turns), the currently `detected_intent`, `retrieved_documents`, and incremental lead variables (`user_name`, `user_email`, `creator_platform`). The state flows deterministically through each node, creating predictable transitions.
+
+## Tool Execution Safety
+The mock backend tool (`mock_lead_capture`) is heavily guarded. It executes solely in the `execute_tool` node, which only runs if the router confirms `lead_ready` is `True`. Furthermore, the node performs a strict validation to ensure `user_name`, `user_email`, and `creator_platform` are all non-null before triggering the function, ensuring no premature or incomplete lead data is dispatched.
+
+## WhatsApp Integration
+This agent can easily be deployed on WhatsApp using webhooks and Twilio:
+1. **Twilio API**: Set up a Twilio WhatsApp Business API sandbox or account.
+2. **Webhook Endpoint**: Create an HTTP endpoint (e.g., via FastAPI or Flask) to receive incoming webhook payloads containing the user's WhatsApp message.
+3. **Agent Backend**: The webhook extracts the message text and user identifier (phone number) and invokes the LangGraph agent.
+4. **Session Management**: A database (like Redis) can key the `AgentState` to the user's phone number, maintaining continuity and conversational memory across incoming webhooks.
+5. **Response Dispatch**: After the graph runs, the final `response` string is dispatched back to the user via a POST request to Twilio's Message API.
+
+## Testing Architecture
+A rigorous suite of tests sits in the `tests/` directory:
+1. **Mocking**: All AI inference (LLMs and Embeddings) is aggressively mocked using `pytest-mock` and standard injection.
+2. **Deterministic Reliability**: By returning controlled mock objects, tests validate the graph structure, logic, state changes, routing, and tool safety independently of live API behavior and latencies.
+3. **End-to-End Simulation**: `test_agent_e2e.py` walks through a multi-turn conversation step-by-step, mimicking user turns and validating correct downstream transitions from Greeting -> RAG -> Lead Capture -> Tool Execution.

agent/graph.py

@@ -0,0 +1,68 @@
+from langgraph.graph import StateGraph, START, END
+from agent.state import AgentState
+from agent.nodes import (
+    detect_intent,
+    handle_greeting,
+    handle_unknown,
+    retrieve_knowledge,
+    generate_rag_response,
+    process_lead,
+    execute_tool
+)
+from agent.router import route_intent, route_after_lead
+
+def build_graph():
+    # Initialize the graph with the typed state
+    workflow = StateGraph(AgentState)
+
+    # Add nodes
+    workflow.add_node("detect_intent", detect_intent)
+    workflow.add_node("handle_greeting", handle_greeting)
+    workflow.add_node("handle_unknown", handle_unknown)
+    workflow.add_node("retrieve_knowledge", retrieve_knowledge)
+    workflow.add_node("generate_rag_response", generate_rag_response)
+    workflow.add_node("process_lead", process_lead)
+    workflow.add_node("execute_tool", execute_tool)
+
+    # Define edges
+    # Start -> detect_intent
+    workflow.add_edge(START, "detect_intent")
+
+    # detect_intent -> conditional routing based on intent
+    workflow.add_conditional_edges(
+        "detect_intent",
+        route_intent,
+        {
+            "handle_greeting": "handle_greeting",
+            "retrieve_knowledge": "retrieve_knowledge",
+            "process_lead": "process_lead",
+            "handle_unknown": "handle_unknown"
+        }
+    )
+
+    # retrieve_knowledge -> generate_rag_response
+    workflow.add_edge("retrieve_knowledge", "generate_rag_response")
+
+    # process_lead -> conditional routing (execute_tool or end)
+    workflow.add_conditional_edges(
+        "process_lead",
+        route_after_lead,
+        {
+            "execute_tool": "execute_tool",
+            "__end__": END
+        }
+    )
+
+    # Define terminal edges
+    workflow.add_edge("handle_greeting", END)
+    workflow.add_edge("handle_unknown", END)
+    workflow.add_edge("generate_rag_response", END)
+    workflow.add_edge("execute_tool", END)
+
+    # Compile the graph
+    app = workflow.compile()
+
+    return app
+
+# Expose a compiled instance
+app = build_graph()

agent/nodes.py

@@ -0,0 +1,112 @@
+from typing import Optional
+from pydantic import BaseModel, Field
+from langchain_openai import ChatOpenAI
+from langchain_core.prompts import ChatPromptTemplate
+from agent.state import AgentState
+from rag.retriever import retrieve_documents
+from tools.lead_capture import mock_lead_capture
+
+def get_llm():
+    return ChatOpenAI(model="gpt-4o-mini", temperature=0)
+
+class IntentResponse(BaseModel):
+    intent: str = Field(description="The intent of the user. Must be one of: GREETING, PRODUCT_QUERY, PRICING_QUERY, HIGH_INTENT_LEAD, UNKNOWN")
+    confidence: float = Field(description="Confidence score between 0 and 1")
+
+class LeadExtractionResponse(BaseModel):
+    user_name: Optional[str] = Field(default=None, description="The name of the user if provided")
+    user_email: Optional[str] = Field(default=None, description="The email address of the user if provided")
+    creator_platform: Optional[str] = Field(default=None, description="The creator platform (e.g., YouTube, Instagram) if provided")
+
+def detect_intent(state: AgentState) -> AgentState:
+    llm = get_llm()
+    prompt = ChatPromptTemplate.from_messages([
+        ("system", "You are an intent classification assistant for AutoStream. Analyze the user's message and determine the intent. Categories: GREETING, PRODUCT_QUERY, PRICING_QUERY, HIGH_INTENT_LEAD, UNKNOWN. A 'HIGH_INTENT_LEAD' is when a user explicitly expresses interest in signing up, buying, or trying out a plan."),
+        ("user", "{message}")
+    ])
+
+    chain = prompt | llm.with_structured_output(IntentResponse)
+
+    history_str = "\n".join([f"{msg['role']}: {msg['content']}" for msg in state.get("conversation_history", [])[-3:]])
+    context_message = f"Recent history:\n{history_str}\n\nCurrent message:\n{state['current_message']}"
+
+    response = chain.invoke({"message": context_message})
+
+    return {"detected_intent": response.intent}
+
+def handle_greeting(state: AgentState) -> AgentState:
+    return {"response": "Hello! I'm the AutoStream assistant. I can answer questions about our features and pricing. How can I help you today?"}
+
+def handle_unknown(state: AgentState) -> AgentState:
+    return {"response": "I'm not quite sure how to help with that. Could you clarify your question about AutoStream?"}
+
+def retrieve_knowledge(state: AgentState) -> AgentState:
+    docs = retrieve_documents(state["current_message"])
+    return {"retrieved_documents": docs}
+
+def generate_rag_response(state: AgentState) -> AgentState:
+    llm = get_llm()
+    prompt = ChatPromptTemplate.from_messages([
+        ("system", "You are a helpful sales assistant for AutoStream. Answer the user's question based strictly on the following retrieved knowledge:\n\n{context}\n\nIf the answer is not in the context, say you don't know."),
+        ("user", "{message}")
+    ])
+
+    context = "\n\n".join(state.get("retrieved_documents", []))
+    chain = prompt | llm
+
+    response = chain.invoke({
+        "context": context,
+        "message": state["current_message"]
+    })
+
+    return {"response": response.content}
+
+def process_lead(state: AgentState) -> AgentState:
+    llm = get_llm()
+
+    extract_prompt = ChatPromptTemplate.from_messages([
+        ("system", "Extract the user's name, email, and creator platform (e.g. YouTube, TikTok, Instagram) from the message if present. Return null for fields not found."),
+        ("user", "{message}")
+    ])
+    extract_chain = extract_prompt | llm.with_structured_output(LeadExtractionResponse)
+
+    history_str = "\n".join([f"{msg['role']}: {msg['content']}" for msg in state.get("conversation_history", [])[-3:]])
+    context_message = f"Recent history:\n{history_str}\n\nCurrent message:\n{state['current_message']}"
+
+    extracted = extract_chain.invoke({"message": context_message})
+
+    updates = {}
+    if extracted.user_name and not state.get("user_name"):
+        updates["user_name"] = extracted.user_name
+    if extracted.user_email and not state.get("user_email"):
+        updates["user_email"] = extracted.user_email
+    if extracted.creator_platform and not state.get("creator_platform"):
+        updates["creator_platform"] = extracted.creator_platform
+
+    current_name = updates.get("user_name", state.get("user_name"))
+    current_email = updates.get("user_email", state.get("user_email"))
+    current_platform = updates.get("creator_platform", state.get("creator_platform"))
+
+    if not current_name:
+        updates["response"] = "Great! I can help with that. Could I have your name?"
+        return updates
+    elif not current_email:
+        updates["response"] = f"Thanks {current_name}! What is your email address?"
+        return updates
+    elif not current_platform:
+        updates["response"] = "Got it. And what creator platform do you primarily use (e.g., YouTube, TikTok)?"
+        return updates
+    else:
+        updates["lead_ready"] = True
+        return updates
+
+def execute_tool(state: AgentState) -> AgentState:
+    if state.get("lead_ready") and state.get("user_name") and state.get("user_email") and state.get("creator_platform"):
+        mock_lead_capture(
+            state["user_name"],
+            state["user_email"],
+            state["creator_platform"]
+        )
+        return {"response": f"Thanks {state['user_name']}! I've successfully collected your information for your {state['creator_platform']} channel. Our team will reach out to {state['user_email']} shortly."}
+    else:
+        return {"response": "Error: Tried to execute lead capture tool without all required fields."}

agent/router.py

@@ -0,0 +1,41 @@
+from agent.state import AgentState
+
+def route_intent(state: AgentState) -> str:
+    """
+    Router node that directs the workflow based on the detected intent.
+    It returns the name of the next node to execute.
+    """
+    # If we are already in the middle of lead collection, we should stay in that flow
+    # This is slightly simplified; we'll route to process_lead if we detected HIGH_INTENT_LEAD
+    # or if we are already missing lead fields but have HIGH_INTENT_LEAD in previous turns.
+    # To keep it simple, if intent is HIGH_INTENT_LEAD, we go to lead workflow.
+    # If we are expecting lead info, the intent classifier might classify as UNKNOWN or something else
+    # We can handle this by checking if there's an ongoing lead collection in state.
+
+    intent = state.get("detected_intent")
+
+    # Check if we were already in lead collection
+    has_partial_lead = (
+        state.get("user_name") is not None or
+        state.get("user_email") is not None or
+        state.get("creator_platform") is not None
+    ) and not state.get("lead_ready")
+
+    if intent == "HIGH_INTENT_LEAD" or has_partial_lead:
+        return "process_lead"
+    elif intent in ["PRODUCT_QUERY", "PRICING_QUERY"]:
+        return "retrieve_knowledge"
+    elif intent == "GREETING":
+        return "handle_greeting"
+    else:
+        return "handle_unknown"
+
+def route_after_lead(state: AgentState) -> str:
+    """
+    Router node after process_lead to decide whether to execute the tool or stop.
+    """
+    if state.get("lead_ready"):
+        return "execute_tool"
+    else:
+        # We need more info, so we just end the graph execution here to wait for user input
+        return "__end__"

agent/state.py

@@ -0,0 +1,20 @@
+from typing import TypedDict, List, Optional, Dict, Any
+
+class AgentState(TypedDict):
+    """
+    Shared state object used by the agent graph.
+    """
+    conversation_history: List[Dict[str, str]]  # list of {"role": "user"/"assistant", "content": "..."}
+    current_message: str
+    detected_intent: Optional[str]
+    retrieved_documents: List[str]
+
+    # Lead collection fields
+    user_name: Optional[str]
+    user_email: Optional[str]
+    creator_platform: Optional[str]
+
+    lead_ready: bool
+
+    # Final response to the user
+    response: str

data/knowledge_base.md

@@ -0,0 +1,16 @@
+# AutoStream Pricing & Features
+
+## Basic Plan
+* $29/month
+* 10 videos per month
+* 720p resolution
+
+## Pro Plan
+* $79/month
+* Unlimited videos
+* 4K resolution
+* AI captions included
+
+# Company Policies
+* No refunds after 7 days
+* 24/7 support available only on Pro plan

main.py

@@ -0,0 +1,73 @@
+import os
+from dotenv import load_dotenv
+from agent.graph import app
+from agent.state import AgentState
+
+def print_header(title):
+    print(f"\n{'='*50}\n{title}\n{'='*50}")
+
+def main():
+    # Load environment variables
+    load_dotenv()
+
+    if not os.environ.get("OPENAI_API_KEY"):
+        print("Warning: OPENAI_API_KEY is not set. The agent will not be able to call the LLM.")
+        print("Please set it in your environment or create a .env file.")
+
+    print_header("AutoStream AI Sales Assistant")
+    print("Type 'quit' or 'exit' to end the conversation.\n")
+
+    # Initialize state
+    state = AgentState(
+        conversation_history=[],
+        current_message="",
+        detected_intent=None,
+        retrieved_documents=[],
+        user_name=None,
+        user_email=None,
+        creator_platform=None,
+        lead_ready=False,
+        response=""
+    )
+
+    while True:
+        try:
+            user_input = input("\nYou: ")
+            if user_input.lower() in ['quit', 'exit']:
+                break
+
+            # Update state with new message
+            state["current_message"] = user_input
+
+            # Run the agent graph
+            print("\n[Agent is thinking...]")
+
+            # Run the graph
+            result_state = app.invoke(state)
+
+            # Update our persistent state with the new state from the graph
+            state = result_state
+
+            # Add to conversation history
+            state["conversation_history"].append({"role": "user", "content": user_input})
+            state["conversation_history"].append({"role": "assistant", "content": state["response"]})
+
+            # Keep history to max 6 turns
+            if len(state["conversation_history"]) > 12:  # 6 turns (user+assistant)
+                state["conversation_history"] = state["conversation_history"][-12:]
+
+            # Display results
+            print(f"[Detected Intent]: {state.get('detected_intent', 'UNKNOWN')}")
+
+            if state.get("retrieved_documents") and state.get("detected_intent") in ["PRODUCT_QUERY", "PRICING_QUERY"]:
+                print(f"[RAG Retrieval]: Found {len(state['retrieved_documents'])} relevant knowledge chunks.")
+
+            print(f"\nAgent: {state['response']}")
+
+        except KeyboardInterrupt:
+            break
+        except Exception as e:
+            print(f"\nAn error occurred: {e}")
+
+if __name__ == "__main__":
+    main()

rag/embeddings.py

@@ -0,0 +1,7 @@
+from langchain_openai import OpenAIEmbeddings
+
+def get_embeddings():
+    """
+    Returns the embedding model used for the RAG pipeline.
+    """
+    return OpenAIEmbeddings()

rag/retriever.py

@@ -0,0 +1,11 @@
+from rag.vectorstore import get_vectorstore
+
+def retrieve_documents(query: str, k: int = 3):
+    """
+    Retrieves the top k relevant documents for the given query.
+    """
+    vectorstore = get_vectorstore()
+    retriever = vectorstore.as_retriever(search_kwargs={"k": k})
+    docs = retriever.invoke(query)
+
+    return [doc.page_content for doc in docs]

rag/vectorstore.py

@@ -0,0 +1,36 @@
+import os
+from langchain_community.document_loaders import TextLoader
+from langchain_text_splitters import RecursiveCharacterTextSplitter
+from langchain_community.vectorstores import FAISS
+from rag.embeddings import get_embeddings
+
+def build_vectorstore(filepath: str = "data/knowledge_base.md"):
+    """
+    Loads the knowledge base, splits it, and builds a FAISS vector store.
+    """
+    if not os.path.exists(filepath):
+        raise FileNotFoundError(f"Knowledge base not found at {filepath}")
+
+    loader = TextLoader(filepath)
+    docs = loader.load()
+
+    text_splitter = RecursiveCharacterTextSplitter(
+        chunk_size=100,
+        chunk_overlap=20,
+        separators=["\n\n", "\n", " ", ""]
+    )
+    splits = text_splitter.split_documents(docs)
+
+    embeddings = get_embeddings()
+    vectorstore = FAISS.from_documents(splits, embeddings)
+
+    return vectorstore
+
+# Cache the vector store globally so we don't rebuild it on every request
+_vectorstore = None
+
+def get_vectorstore(filepath: str = "data/knowledge_base.md"):
+    global _vectorstore
+    if _vectorstore is None:
+        _vectorstore = build_vectorstore(filepath)
+    return _vectorstore

requirements.txt

@@ -0,0 +1,10 @@
+langchain
+langgraph
+langchain-openai
+langchain-community
+langchain-text-splitters
+faiss-cpu
+python-dotenv
+pydantic
+pytest
+pytest-mock

tests/test_agent_e2e.py

@@ -0,0 +1,105 @@
+import pytest
+from agent.graph import app
+from agent.state import AgentState
+from agent.nodes import IntentResponse, LeadExtractionResponse
+from langchain_core.runnables import RunnableLambda
+
+def simulate_conversation(messages, mock_llm_setup_func):
+    """
+    Helper utility that simulates a multi-turn conversation.
+    Feeds messages sequentially through the agent graph and returns the final state.
+    """
+    state = AgentState(
+        conversation_history=[],
+        current_message="",
+        detected_intent=None,
+        retrieved_documents=[],
+        user_name=None,
+        user_email=None,
+        creator_platform=None,
+        lead_ready=False,
+        response=""
+    )
+
+    for idx, msg in enumerate(messages):
+        state["current_message"] = msg
+        mock_llm_setup_func(idx) # setup mocks for this turn
+        state = app.invoke(state)
+
+        # update history manually
+        state["conversation_history"].append({"role": "user", "content": state["current_message"]})
+        state["conversation_history"].append({"role": "assistant", "content": state["response"]})
+
+    return state
+
+def test_agent_e2e(mocker):
+    # E2E Test USING graph.invoke
+    # We patch the `get_llm` inside `agent.nodes` to return a mock LLM.
+    mock_llm = mocker.MagicMock()
+    mocker.patch('agent.nodes.get_llm', return_value=mock_llm)
+
+    # Mock RAG retrieval
+    mocker.patch('agent.nodes.retrieve_documents', return_value=["We have Basic and Pro plans for $29 and $79."])
+
+    mock_tool = mocker.patch('agent.nodes.mock_lead_capture')
+
+    messages = [
+        "Hi",
+        "Tell me about pricing",
+        "I want the Pro plan for my YouTube channel",
+        "My name is Alex",
+        "alex@email.com"
+    ]
+
+    def setup_mocks_for_turn(idx):
+        if idx == 0:
+            # Turn 1: Greeting
+            mock_chain = RunnableLambda(lambda x: IntentResponse(intent="GREETING", confidence=0.99))
+            mock_llm.with_structured_output.return_value = mock_chain
+        elif idx == 1:
+            # Turn 2: Pricing
+            mock_chain = RunnableLambda(lambda x: IntentResponse(intent="PRICING_QUERY", confidence=0.99))
+            mock_llm.with_structured_output.return_value = mock_chain
+
+            # The regular invoke for generate_rag_response returns AIMessage-like object
+            class FakeResponse:
+                content = "We have Basic and Pro plans."
+            mock_llm.invoke.return_value = FakeResponse()
+
+        elif idx == 2:
+            # Turn 3: High intent lead
+            # The router uses intent. The process_lead uses with_structured_output.
+            # Since both use with_structured_output in the same turn, we need a side_effect.
+            def mock_structured_output(schema):
+                if schema.__name__ == "IntentResponse":
+                    return RunnableLambda(lambda x: IntentResponse(intent="HIGH_INTENT_LEAD", confidence=0.99))
+                else:
+                    return RunnableLambda(lambda x: LeadExtractionResponse(user_name=None, user_email=None, creator_platform="YouTube"))
+            mock_llm.with_structured_output.side_effect = mock_structured_output
+
+        elif idx == 3:
+            # Turn 4: Provide name
+            def mock_structured_output(schema):
+                if schema.__name__ == "IntentResponse":
+                    return RunnableLambda(lambda x: IntentResponse(intent="HIGH_INTENT_LEAD", confidence=0.99))
+                else:
+                    return RunnableLambda(lambda x: LeadExtractionResponse(user_name="Alex", user_email=None, creator_platform=None))
+            mock_llm.with_structured_output.side_effect = mock_structured_output
+
+        elif idx == 4:
+            # Turn 5: Provide email
+            def mock_structured_output(schema):
+                if schema.__name__ == "IntentResponse":
+                    return RunnableLambda(lambda x: IntentResponse(intent="HIGH_INTENT_LEAD", confidence=0.99))
+                else:
+                    return RunnableLambda(lambda x: LeadExtractionResponse(user_name=None, user_email="alex@email.com", creator_platform=None))
+            mock_llm.with_structured_output.side_effect = mock_structured_output
+
+    final_state = simulate_conversation(messages, setup_mocks_for_turn)
+
+    assert final_state.get("user_name") == "Alex"
+    assert final_state.get("user_email") == "alex@email.com"
+    assert final_state.get("creator_platform") == "YouTube"
+    assert final_state.get("lead_ready") is True
+
+    mock_tool.assert_called_once_with("Alex", "alex@email.com", "YouTube")

tests/test_intent_classifier.py

@@ -0,0 +1,68 @@
+import pytest
+from agent.nodes import detect_intent
+from agent.state import AgentState
+from agent.nodes import IntentResponse
+from langchain_core.runnables import RunnableLambda
+
+def test_intent_classifier_greeting(mocker):
+    state = AgentState(
+        conversation_history=[],
+        current_message="Hi there",
+        detected_intent=None,
+        retrieved_documents=[],
+        user_name=None,
+        user_email=None,
+        creator_platform=None,
+        lead_ready=False,
+        response=""
+    )
+
+    mock_llm = mocker.MagicMock()
+    mock_chain = RunnableLambda(lambda x: IntentResponse(intent="GREETING", confidence=0.99))
+    mock_llm.with_structured_output.return_value = mock_chain
+    mocker.patch('agent.nodes.get_llm', return_value=mock_llm)
+
+    result = detect_intent(state)
+    assert result["detected_intent"] == "GREETING"
+
+def test_intent_classifier_pricing(mocker):
+    state = AgentState(
+        conversation_history=[],
+        current_message="What are your pricing plans?",
+        detected_intent=None,
+        retrieved_documents=[],
+        user_name=None,
+        user_email=None,
+        creator_platform=None,
+        lead_ready=False,
+        response=""
+    )
+
+    mock_llm = mocker.MagicMock()
+    mock_chain = RunnableLambda(lambda x: IntentResponse(intent="PRICING_QUERY", confidence=0.95))
+    mock_llm.with_structured_output.return_value = mock_chain
+    mocker.patch('agent.nodes.get_llm', return_value=mock_llm)
+
+    result = detect_intent(state)
+    assert result["detected_intent"] == "PRICING_QUERY"
+
+def test_intent_classifier_high_intent(mocker):
+    state = AgentState(
+        conversation_history=[],
+        current_message="I want to sign up for Pro plan",
+        detected_intent=None,
+        retrieved_documents=[],
+        user_name=None,
+        user_email=None,
+        creator_platform=None,
+        lead_ready=False,
+        response=""
+    )
+
+    mock_llm = mocker.MagicMock()
+    mock_chain = RunnableLambda(lambda x: IntentResponse(intent="HIGH_INTENT_LEAD", confidence=0.91))
+    mock_llm.with_structured_output.return_value = mock_chain
+    mocker.patch('agent.nodes.get_llm', return_value=mock_llm)
+
+    result = detect_intent(state)
+    assert result["detected_intent"] == "HIGH_INTENT_LEAD"

tests/test_lead_workflow.py

@@ -0,0 +1,56 @@
+import pytest
+from agent.nodes import process_lead, LeadExtractionResponse
+from agent.state import AgentState
+from langchain_core.runnables import RunnableLambda
+
+def test_lead_workflow_step_by_step(mocker):
+    # Step 1: User says they want the Pro plan for YouTube
+    state = AgentState(
+        conversation_history=[],
+        current_message="I want the Pro plan for my YouTube channel",
+        detected_intent="HIGH_INTENT_LEAD",
+        retrieved_documents=[],
+        user_name=None,
+        user_email=None,
+        creator_platform=None,
+        lead_ready=False,
+        response=""
+    )
+
+    mock_llm = mocker.MagicMock()
+    mock_chain_1 = RunnableLambda(lambda x: LeadExtractionResponse(user_name=None, user_email=None, creator_platform="YouTube"))
+    mock_llm.with_structured_output.return_value = mock_chain_1
+    mocker.patch('agent.nodes.get_llm', return_value=mock_llm)
+
+    result = process_lead(state)
+    assert result.get("user_name") is None
+    assert result.get("creator_platform") == "YouTube"
+    assert "name" in result["response"].lower()
+
+    # Simulate state update
+    state.update(result)
+    state["conversation_history"].append({"role": "user", "content": state["current_message"]})
+    state["conversation_history"].append({"role": "assistant", "content": state["response"]})
+
+    # Step 2: User provides name
+    state["current_message"] = "My name is Alex"
+    mock_chain_2 = RunnableLambda(lambda x: LeadExtractionResponse(user_name="Alex", user_email=None, creator_platform=None))
+    mock_llm.with_structured_output.return_value = mock_chain_2
+
+    result = process_lead(state)
+    assert result.get("user_name") == "Alex"
+    assert "email" in result["response"].lower()
+
+    # Simulate state update
+    state.update(result)
+    state["conversation_history"].append({"role": "user", "content": state["current_message"]})
+    state["conversation_history"].append({"role": "assistant", "content": state["response"]})
+
+    # Step 3: User provides email
+    state["current_message"] = "alex@email.com"
+    mock_chain_3 = RunnableLambda(lambda x: LeadExtractionResponse(user_name=None, user_email="alex@email.com", creator_platform=None))
+    mock_llm.with_structured_output.return_value = mock_chain_3
+
+    result = process_lead(state)
+    assert result.get("user_email") == "alex@email.com"
+    assert result.get("lead_ready") is True

tests/test_rag_pipeline.py

@@ -0,0 +1,51 @@
+import pytest
+import os
+from rag.vectorstore import build_vectorstore
+import rag.vectorstore
+from langchain_core.embeddings import Embeddings
+from typing import List
+
+os.environ["OPENAI_API_KEY"] = "dummy_key"
+
+class MockEmbedding(Embeddings):
+    def embed_documents(self, texts: List[str]) -> List[List[float]]:
+        # Just return a zero vector of size 1536 for each input text
+        return [[0.0] * 1536 for _ in texts]
+
+    def embed_query(self, text: str) -> List[float]:
+        return [0.0] * 1536
+
+def test_rag_pipeline_loads_and_retrieves(mocker, tmp_path):
+    # Test end-to-end vectorstore build and retrieval (testing doc loading and splitting)
+    kb_file = tmp_path / "knowledge_base.md"
+    kb_file.write_text("""
+# AutoStream Pricing & Features
+
+## Pro Plan
+* $79/month
+* Unlimited videos
+* 4K resolution
+* AI captions included
+    """)
+
+    # We must patch get_embeddings in vectorstore so it uses our mock that doesn't call OpenAI
+    mocker.patch('rag.vectorstore.get_embeddings', return_value=MockEmbedding())
+    # FAISS has an internal check for Embeddings class, so MockEmbedding must inherit from Embeddings
+
+    # Mock the actual FAISS from_documents internally to just create an empty FAISS store,
+    # OR we can let FAISS run with our mock embeddings. Let's let it run with mock embeddings.
+    vs = build_vectorstore(str(kb_file))
+    assert vs is not None
+
+    # Now patch the global get_vectorstore so our retriever uses this one
+    mocker.patch('rag.retriever.get_vectorstore', return_value=vs)
+    from rag.retriever import retrieve_documents
+
+    docs = retrieve_documents("What does the Pro plan cost?", k=1)
+
+    # Since all embeddings are 0, it will return the first document(s) it split.
+    # With chunk size 100, the first few lines should be retrieved.
+    assert len(docs) > 0
+    # The actual retrieval will return a chunk. The first chunk should have "Pro Plan" or "AutoStream Pricing".
+    # Just asserting it retrieved something from our mock file.
+    assert "AutoStream" in docs[0] or "Pro Plan" in docs[0] or "$79/month" in docs[0]

tests/test_tool_execution.py

@@ -0,0 +1,45 @@
+import pytest
+from agent.nodes import execute_tool
+from agent.state import AgentState
+
+def test_tool_execution_missing_fields(mocker):
+    mock_tool = mocker.patch('agent.nodes.mock_lead_capture')
+
+    state = AgentState(
+        conversation_history=[],
+        current_message="",
+        detected_intent="HIGH_INTENT_LEAD",
+        retrieved_documents=[],
+        user_name="Alex",
+        user_email="alex@email.com",
+        creator_platform=None, # Missing platform
+        lead_ready=True,
+        response=""
+    )
+
+    result = execute_tool(state)
+
+    # Tool should NOT be executed
+    mock_tool.assert_not_called()
+    assert "Error" in result["response"]
+
+def test_tool_execution_all_fields(mocker):
+    mock_tool = mocker.patch('agent.nodes.mock_lead_capture')
+
+    state = AgentState(
+        conversation_history=[],
+        current_message="",
+        detected_intent="HIGH_INTENT_LEAD",
+        retrieved_documents=[],
+        user_name="Alex",
+        user_email="alex@email.com",
+        creator_platform="YouTube",
+        lead_ready=True,
+        response=""
+    )
+
+    result = execute_tool(state)
+
+    # Tool should be executed exactly once
+    mock_tool.assert_called_once_with("Alex", "alex@email.com", "YouTube")
+    assert "Thanks Alex" in result["response"]

tools/lead_capture.py

@@ -0,0 +1,5 @@
+def mock_lead_capture(name: str, email: str, platform: str):
+    """
+    Mock backend function to capture lead information.
+    """
+    print(f"Lead captured successfully: {name}, {email}, {platform}")