Refactor research agent implementation and remove unused modules

- Removed the LangChainRAG model and related components to streamline the architecture.
- Introduced a new agent building mechanism using `build_agent` function.
- Updated the configuration to include a maximum search results parameter.
- Replaced the RAG prompt template with a more concise version.
- Implemented a new RSS feed tool for fetching articles from specified URLs.
- Consolidated search tools into a single function for easier management.
- Enhanced error handling and logging throughout the agent's processing functions.
- Updated dependencies in `pyproject.toml` to include necessary libraries for new features.

py-src/app.py

@@ -18,44 +18,12 @@ from qdrant_client import QdrantClient
 from qdrant_client.http.models import Distance, VectorParams
 from lets_talk.config import LLM_MODEL, LLM_TEMPERATURE
 import lets_talk.utils.blog as blog
-from lets_talk.models.rag import LangChainRAG
+from lets_talk.agent import build_agent,parse_output
 
-    
-# Load vector store using the utility function
-vector_store = blog.load_vector_store()
-
-# Create a retriever
-retriever = vector_store.as_retriever()
-
-# Set up ChatOpenAI with environment variables
-
-llm = ChatOpenAI(model=LLM_MODEL, temperature=LLM_TEMPERATURE)
-
-# Create RAG prompt template
-rag_prompt_template = """\
-You are a helpful assistant that answers questions based on the context provided. 
-Generate a concise answer to the question in markdown format and include a list of relevant links to the context.
-Use links from context to help user to navigate to to find more information.
-You have access to the following information:
-
-Context:
-{context}
-
-Question:
-{question}
-
-If context is unrelated to question, say "I don't know".
-"""
 
-rag_prompt = ChatPromptTemplate.from_template(rag_prompt_template)
-
-# Create chain
-retrieval_augmented_qa_chain = (
-    {"context": itemgetter("question") | retriever, "question": itemgetter("question")}
-    | RunnablePassthrough.assign(context=itemgetter("context"))
-    | {"response": rag_prompt | llm, "context": itemgetter("context")}
-)
+    
 
+tdg_agent = build_agent()
 
   
 @cl.on_chat_start
@@ -74,29 +42,56 @@ async def setup_chain():
     # Set a loading message
     msg = cl.Message(content="Let's talk about [TheDataGuy](https://thedataguy.pro)'s blog posts, how can I help you?", author="System")
     await msg.send()
-
-    #rag_chain = LangChainRAG(llm=llm, retriever=retriever)
     
     # Store the chain in user session
-    cl.user_session.set("chain", retrieval_augmented_qa_chain)
-    #cl.user_session.set("chain", rag_chain)
+    cl.user_session.set("agent", tdg_agent)
+    
     
 
     
 
 @cl.on_message
 async def on_message(message: cl.Message):
+    """
+    Handler for user messages. Processes the query through the research agent
+    and streams the response back to the user.
+    
+    Args:
+        message: The user's message
+    """
+    agent_executor = cl.user_session.get("agent")
+    
+    # Create Chainlit message for streaming
     msg = cl.Message(content="")
-
-    # Get chain from user session
-    chain = cl.user_session.get("chain")
     
-    # Call the chain with the user message
-    response = await chain.ainvoke({"question": message.content})
-    #response = await chain.arun_pipeline(message.content)
+    # Create a parent step for the research process
+    with cl.Step(name="TheDataGuy thinking", type="tool") as step:
+        # Run the agent executor with callbacks to stream the response
+        result = await agent_executor.ainvoke(
+            {"question": message.content},
+            config={
+                "callbacks": [cl.AsyncLangchainCallbackHandler()],
+                "configurable": {"session_id": message.id}  # Add session_id from message
+            }
+        )
+        
+        # Add steps from agent's intermediate steps
+        # for i, step_data in enumerate(result.get("intermediate_steps", [])):
+        #     step_name = f"Using: {step_data[0].tool}"
+        #     step_input = str(step_data[0].tool_input)
+        #     step_output = str(step_data[1])
+            
+        #     # Create individual steps as children of the main step
+        #     with cl.Step(name=step_name, type="tool") as substep:
+        #         await cl.Message(
+        #             content=f"**Input:** {step_input}\n\n**Output:** {step_output}",
+        #         ).send()
+    
+    # Get the final answer
+    final_answer = parse_output(result)
     
-     # Stream tokens from the final_answer
-    await msg.stream_token(response["response"].content)
+    # Stream tokens from the final_answer
+    await msg.stream_token(final_answer)
     await msg.send()
 
   

py-src/lets_talk/{models/agent.py → agent.py}

@@ -1,6 +1,4 @@
-"""
-LangGraph Agent implementation for the Research Agent.
-"""
+
 from typing import TypedDict, Annotated, Dict, Any, Literal, Union, cast, List, Optional
 from langchain_openai import ChatOpenAI
 from langchain_core.tools import Tool
@@ -9,8 +7,12 @@ from langchain_core.documents import Document
 from langgraph.graph.message import add_messages
 from langgraph.graph import StateGraph, END
 from langgraph.prebuilt import ToolNode
-from lets_talk.models.research_tools import RAGQueryInput
+from lets_talk.models import RAGQueryInput
 from lets_talk.config import LLM_MODEL, LLM_TEMPERATURE
+from lets_talk.tools import create_search_tools
+from datetime import datetime
+import lets_talk.rag as rag
+
 
 class ResearchAgentState(TypedDict):
     """
@@ -23,7 +25,68 @@ class ResearchAgentState(TypedDict):
     """
     messages: Annotated[list[BaseMessage], add_messages]
     context: str
-    documents: Optional[List[Document]]
+    
+
+rag_prompt_template = """\
+You are a helpful assistant that answers questions based on the context provided. 
+Generate a concise answer to the question in markdown format and include a list of relevant links to the context.
+Use links from context to help user to navigate to to find more information.
+
+You have access to the following information:
+
+Context:
+{context}
+
+If context is unrelated to question, say "I don't know".
+"""
+
+# Update the call_model function to include current datetime
+def call_model(model, state: Dict[str, Any]) -> Dict[str, list[BaseMessage]]:
+    """
+    Process the current state through the language model.
+    
+    Args:
+        model: Language model with tools bound
+        state: Current state containing messages and context
+        
+    Returns:
+        Updated state with model's response added to messages
+    """
+    try:
+        messages = state["messages"]
+        context = state.get("context", "")
+        
+        
+        # Get current datetime
+        current_datetime = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        # Insert system message with context before the latest user message
+        sys_prompt = rag_prompt_template.format(
+            context=context,
+        )
+        sys_prompt = f"Today is: {current_datetime}\n\n" + sys_prompt
+        print(sys_prompt)
+        
+        context_message = SystemMessage(content=sys_prompt)
+
+        # Find the position of the last user message
+        for i in range(len(messages)-1, -1, -1):
+            if isinstance(messages[i], HumanMessage):
+                # Insert context right after the last user message
+                enhanced_messages = messages[:i+1] + [context_message] + messages[i+1:]
+                break
+        else:
+            # No user message found, just append context
+            enhanced_messages = messages + [context_message]
+        
+        # Get response from the model
+        response = model.invoke(enhanced_messages)
+        return {"messages": [response]}
+    except Exception as e:
+        # Handle exceptions gracefully
+        error_msg = f"Error calling model: {str(e)}"
+        print(error_msg)  # Log the error
+        # Return a fallback response
+        return {"messages": [HumanMessage(content=error_msg)]}
 
 
 def call_model(model, state: Dict[str, Any]) -> Dict[str, list[BaseMessage]]:
@@ -44,8 +107,8 @@ def call_model(model, state: Dict[str, Any]) -> Dict[str, list[BaseMessage]]:
         # Add context from documents if available
         if context:
             # Insert system message with context before the latest user message
-            context_message = SystemMessage(content=f"Use the following information from uploaded documents to enhance your response if relevant:\n\n{context}")
-            
+            context_message = SystemMessage(content=rag_prompt_template.format(context=context))
+
             # Find the position of the last user message
             for i in range(len(messages)-1, -1, -1):
                 if isinstance(messages[i], HumanMessage):
@@ -87,17 +150,8 @@ def should_continue(state: Dict[str, Any]) -> Union[Literal["action"], Literal["
     return "end"
 
 
-def retrieve_from_documents(state: Dict[str, Any], retriever) -> Dict[str, str]:
-    """
-    Retrieve relevant context from uploaded documents based on the user query.
-    
-    Args:
-        state: Current state containing messages and optional documents
-        retriever: Document retriever to use
-        
-    Returns:
-        Updated state with context from document retrieval
-    """
+def retrieve_from_blog(state: Dict[str, Any]) -> Dict[str, str]:
+  
     # Get the last user message
     for message in reversed(state["messages"]):
         if isinstance(message, HumanMessage):
@@ -107,48 +161,26 @@ def retrieve_from_documents(state: Dict[str, Any], retriever) -> Dict[str, str]:
         # No user message found
         return {"context": ""}
     
-    # Skip if no documents are uploaded
-    if not retriever:
-        return {"context": ""}
-    
     try:
-        # Retrieve relevant documents
-        docs = retriever.invoke(query)
-        if not docs:
-            return {"context": ""}
-        
-        # Extract text from documents
-        context = "\n\n".join([f"Document excerpt: {doc.page_content}" for doc in docs])
+        #context = blog_search_tool(query)
+        response = rag.rag_chain.invoke({"question": query})
+
+        context = response["response"].content
+
         return {"context": context}
     except Exception as e:
         print(f"Error retrieving from documents: {str(e)}")
         return {"context": ""}
 
 
-def document_search_tool(retriever, query: str) -> str:
-    """
-    Tool function to search within uploaded documents.
-    
-    Args:
-        retriever: Document retriever to use
-        query: Search query string
-        
-    Returns:
-        Information retrieved from the documents
-    """
-    if not retriever:
-        return "No documents have been uploaded yet. Please upload a document first."
-    
-    docs = retriever.invoke(query)
+def blog_search_tool(query: str) -> str:
+    docs =  rag.retriever.invoke(query)
     if not docs:
-        return "No relevant information found in the uploaded documents."
+        return "No relevant documents found."
     
-    # Format the results
-    results = []
-    for i, doc in enumerate(docs):
-        results.append(f"[Document {i+1}] {doc.page_content}")
-    
-    return "\n\n".join(results)
+    context = "\n\n---".join([ f"link: {doc.metadata["url"] }\n\n{doc.page_content}" for doc in docs])
+    return  context
+
 
 
 def convert_inputs(input_object: Dict[str, str]) -> Dict[str, list[BaseMessage]]:
@@ -183,22 +215,10 @@ def parse_output(input_state: Dict[str, Any]) -> str:
         return "I encountered an error while processing your request."
 
 
-def build_agent_chain(tools, retriever) -> StateGraph:
-    """
-    Constructs and returns the research agent execution chain.
-    
-    The chain consists of:
-    1. A retrieval node that gets context from documents
-    2. An agent node that processes messages
-    3. A tool node that executes tools when called
-    
-    Args:
-        tools: List of tools for the agent
-        retriever: Optional retriever for document search
-        
-    Returns:
-        Compiled agent chain ready for execution
-    """
+def build_agent() -> StateGraph:
+       
+    tools = create_search_tools(5)
+
     # Create an instance of ChatOpenAI
     model = ChatOpenAI(model=LLM_MODEL, temperature=LLM_TEMPERATURE)
     model = model.bind_tools(tools)
@@ -206,9 +226,9 @@ def build_agent_chain(tools, retriever) -> StateGraph:
     # Create document search tool if retriever is provided
     
     doc_search_tool = Tool(
-        name="DocumentSearch",
-        description="Search within the user's uploaded document. Use this tool when you need information from the specific document that was uploaded.",
-        func=lambda query: document_search_tool(retriever, query),
+        name="TheDataGuy Blog Search",
+        description="Search within blog posts of thedataguy.pro. ALWAYS use this tool to retrieve the context.",
+        func=lambda query: blog_search_tool(query),
         args_schema=RAGQueryInput
     )
     
@@ -226,17 +246,15 @@ def build_agent_chain(tools, retriever) -> StateGraph:
     def call_model_node(state):
         return call_model(model, state)
 
-    # Add nodes
     
     # Define retrieval node factory with bound retriever
     def retrieve_node(state):
-        return retrieve_from_documents(state, retriever)
+        return retrieve_from_blog(state)
     
     uncompiled_graph.add_node("retrieve", retrieve_node)
     uncompiled_graph.set_entry_point("retrieve")
-    uncompiled_graph.add_edge("retrieve", "agent")
-        
     uncompiled_graph.add_node("agent", call_model_node)
+    uncompiled_graph.add_edge("retrieve", "agent")
     uncompiled_graph.add_node("action", tool_node)
     
     # Add an end node - this is required for the "end" state to be valid

py-src/lets_talk/config.py

@@ -12,11 +12,7 @@ QDRANT_COLLECTION = os.environ.get("QDRANT_COLLECTION", "thedataguy_documents")
 BLOG_BASE_URL = os.environ.get("BLOG_BASE_URL", "https://thedataguy.pro/blog/")
 LLM_MODEL = os.environ.get("LLM_MODEL", "gpt-4o-mini")
 LLM_TEMPERATURE = float(os.environ.get("TEMPERATURE", "0"))
+MAX_SEARCH_RESULTS = int(os.environ.get("MAX_SEARCH_RESULTS", "5"))
 
 
 
-SYSTEM_TEMPLATE = """
-You are a helpful assistant that answers questions based on the context provided. 
-Generate a concise answer to the question in markdown format and include a list of relevant links to the context.
-Use links from context to help user to navigate to to find more information. If context is unrelated to question, say "I don't know".
-"""

py-src/lets_talk/models.py

@@ -0,0 +1,29 @@
+from pydantic import BaseModel, Field
+from typing import List, Optional
+
+class ArxivQueryInput(BaseModel):
+    """Input for arXiv query."""
+    query: str = Field(..., description="The search query to find papers on arXiv")
+    max_results: int = Field(default=5, description="The maximum number of results to return")
+
+class RAGQueryInput(BaseModel):
+    """Input for RAG query."""
+    query: str = Field(..., description="The query to search in the uploaded document")
+
+class WebSearchInput(BaseModel):
+    """Input for web search."""
+    query: str = Field(..., description="The search query for web search")
+    max_results: int = Field(default=5, description="The maximum number of results to return")
+
+class DocumentAnalysisInput(BaseModel):
+    """Input for document analysis."""
+    query: str = Field(..., description="The specific question to analyze in the document")
+    include_citations: bool = Field(default=True, description="Whether to include citations in the response")
+
+class RSSFeedInput(BaseModel):
+    """Input for RSS feed tool."""
+    urls: List[str] = Field(..., description="List of RSS feed URLs to fetch articles from")
+    query: Optional[str] = Field(None, description="Optional query to filter articles by relevance")
+    max_results: int = Field(default=5, description="Maximum number of articles to return")
+    nlp: bool = Field(default=True, description="Whether to use NLP processing on articles (extracts keywords and summaries)")
+

py-src/lets_talk/models/rag.py

@@ -1,69 +0,0 @@
-"""
-RAG (Retrieval Augmented Generation) model implementation.
-"""
-from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
-from langchain_core.output_parsers import StrOutputParser
-from langchain_core.runnables import RunnablePassthrough
-
-from lets_talk import config
-
-# Create prompt template
-prompt = ChatPromptTemplate.from_messages([
-    ("system", config.SYSTEM_TEMPLATE),
-    MessagesPlaceholder(variable_name="chat_history"),
-    ("human", "{question}"),
-    ("human", "Context: {context}")
-])
-
-class LangChainRAG:
-    """
-    RAG implementation using LangChain components.
-    """
-    def __init__(self, retriever, llm):
-        """
-        Initialize the RAG model.
-        
-        Args:
-            retriever: Document retriever component
-            llm: Language model for generation
-        """
-        self.retriever = retriever
-        self.llm = llm
-        self.chain = self._create_chain()
-        
-    def _create_chain(self):
-        """
-        Create the RAG chain.
-        
-        Returns:
-            A runnable chain that processes user queries
-        """
-        # Define the RAG chain
-        rag_chain = (
-            {"context": self.retriever, "question": RunnablePassthrough(), "chat_history": lambda _: []}
-            | prompt
-            | self.llm
-            | StrOutputParser()
-        )
-        return rag_chain
-    
-    async def arun_pipeline(self, user_query: str):
-        """
-        Run the RAG pipeline with the user query.
-        
-        Args:
-            user_query: User's question
-            
-        Returns:
-            Dict containing the response generator and context
-        """
-        # Get relevant documents for context
-        docs = self.retriever.invoke(user_query)
-        context_list = [(doc.page_content, doc.metadata) for doc in docs]
-        
-        # Create async generator for streaming
-        async def generate_response():
-            async for chunk in self.chain.astream(user_query):
-                yield chunk
-        
-        return {"response": generate_response(), "context": context_list}

py-src/lets_talk/rag.py

@@ -0,0 +1,49 @@
+"""
+RAG (Retrieval Augmented Generation) model implementation.
+"""
+from operator import itemgetter
+from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
+from langchain_core.output_parsers import StrOutputParser
+from langchain_core.runnables import RunnablePassthrough
+from langchain.prompts import ChatPromptTemplate
+from langchain.schema.runnable import RunnablePassthrough
+from langchain_openai.chat_models import ChatOpenAI
+from langchain_qdrant import QdrantVectorStore
+from lets_talk import config
+from lets_talk.utils import blog
+import lets_talk.utils.blog as blog
+
+# Load vector store using the utility function
+vector_store:QdrantVectorStore = blog.load_vector_store()
+
+# Create a retriever
+retriever = vector_store.as_retriever()
+
+llm = ChatOpenAI(model=config.LLM_MODEL, temperature=config.LLM_TEMPERATURE)
+
+# Create RAG prompt template
+rag_prompt_template = """\
+You are a helpful assistant that answers questions based on the context provided. 
+Generate a concise answer to the question in markdown format and include a list of relevant links to the context.
+Use links from context to help user to navigate to to find more information.
+You have access to the following information:
+
+Context:
+{context}
+
+Question:
+{question}
+
+If context is unrelated to question, say "I don't know".
+"""
+
+rag_prompt = ChatPromptTemplate.from_template(rag_prompt_template)
+
+# Create chain
+rag_chain = (
+    {"context": itemgetter("question") | retriever, "question": itemgetter("question")}
+    | RunnablePassthrough.assign(context=itemgetter("context"))
+    | {"response": rag_prompt | llm, "context": itemgetter("context")}
+)
+
+

py-src/lets_talk/{models/research_tools.py → rss_tool.py}

@@ -5,37 +5,10 @@ This module implements input schemas and tools specifically for research purpose
 """
 from typing import List, Optional, Dict, Any
 from pydantic import BaseModel, Field
-
 from langchain_core.tools import Tool
 from langchain_core.documents import Document
 import feedparser
-import datetime
-
-class ArxivQueryInput(BaseModel):
-    """Input for arXiv query."""
-    query: str = Field(..., description="The search query to find papers on arXiv")
-    max_results: int = Field(default=5, description="The maximum number of results to return")
-
-class RAGQueryInput(BaseModel):
-    """Input for RAG query."""
-    query: str = Field(..., description="The query to search in the uploaded document")
-
-class WebSearchInput(BaseModel):
-    """Input for web search."""
-    query: str = Field(..., description="The search query for web search")
-    max_results: int = Field(default=5, description="The maximum number of results to return")
-
-class DocumentAnalysisInput(BaseModel):
-    """Input for document analysis."""
-    query: str = Field(..., description="The specific question to analyze in the document")
-    include_citations: bool = Field(default=True, description="Whether to include citations in the response")
-
-class RSSFeedInput(BaseModel):
-    """Input for RSS feed tool."""
-    urls: List[str] = Field(..., description="List of RSS feed URLs to fetch articles from")
-    query: Optional[str] = Field(None, description="Optional query to filter articles by relevance")
-    max_results: int = Field(default=5, description="Maximum number of articles to return")
-    nlp: bool = Field(default=True, description="Whether to use NLP processing on articles (extracts keywords and summaries)")
+from .models import RSSFeedInput
 
 
 def rss_feed_tool(urls: List[str], query: Optional[str] = None, max_results: int = 5, nlp: bool = True) -> str:

py-src/lets_talk/{models/search_tools.py → tools.py}

@@ -3,8 +3,10 @@ Search tools module containing different search implementations.
 """
 
 from langchain_community.tools.arxiv.tool import ArxivQueryRun
-from langchain_community.tools import DuckDuckGoSearchResults
+#from langchain_community.tools import DuckDuckGoSearchResults
 from langchain_core.tools import Tool
+from .rss_tool import rss_feed_tool
+
 
 def create_search_tools(max_results=5):
     """
@@ -16,13 +18,33 @@ def create_search_tools(max_results=5):
     Returns:
         List of search tools for the agent
     """
+
+
+    def create_rss_feed_tool() -> Tool:
+        """
+        Create and return an RSS feed tool.
+        
+        Returns:
+            Tool object for RSS feed functionality
+        """
+        def _rss_feed_tool_wrapper(*args, **kwargs):
+            
+            return rss_feed_tool(urls=['https://thedataguy.pro/rss.xml'])
+                
+        return Tool(
+            name="RSSFeedReader",
+            description="Fetch and read articles from TheDataGuy's RSS feeds. Use this tool when you need the latest blog posts, what's new or latest updates.",
+            func=_rss_feed_tool_wrapper
+        )
+
+
     # Initialize standard search tools
-    #tavily_tool = TavilySearchResults(max_results=max_results)
-    duckduckgo_tool = DuckDuckGoSearchResults(max_results=max_results)
-    arxiv_tool = ArxivQueryRun()
+    #duckduckgo_tool = DuckDuckGoSearchResults(max_results=max_results)
+    #arxiv_tool = ArxivQueryRun()
+    tdg_rss_tool = create_rss_feed_tool()
     
     return [
-        #tavily_tool,
-        duckduckgo_tool,
-        arxiv_tool,
+        tdg_rss_tool,
+        #duckduckgo_tool,
+        #arxiv_tool,
     ]

pyproject.toml

@@ -5,7 +5,9 @@ description = "Add your description here"
 readme = "README.md"
 requires-python = ">=3.13"
 dependencies = [
+    "arxiv>=2.2.0",
     "chainlit>=2.5.5",
+    "feedparser>=6.0.11",
     "ipykernel>=6.29.5",
     "ipython>=9.2.0",
     "langchain>=0.3.25",
@@ -15,6 +17,7 @@ dependencies = [
     "langchain-openai>=0.3.16",
     "langchain-qdrant>=0.2.0",
     "langchain-text-splitters>=0.3.8",
+    "langgraph>=0.4.3",
     "pandas>=2.2.3",
     "python-dotenv>=1.1.0",
     "qdrant-client>=1.14.2",