check for the following error occurance "Error running agent on task cabe07ed-9eca-40ea-8ead-410ef5e83f91: HumanMessage.init() missing 1 required positional argument: 'content'" at every possible area in the following code: agent.py #─── Basic imports ───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── import os import math import sqlite3 import fitz # PyMuPDF for PDF parsing import re from dotenv import load_dotenv from langgraph.graph import START, StateGraph, MessagesState from langchain_community.vectorstores import SupabaseVectorStore, FAISS from langgraph.prebuilt import tools_condition from langgraph.prebuilt import ToolNode from langgraph.constants import START from langgraph.graph import START from langchain_core.tools import tool from langchain.embeddings import HuggingFaceEmbeddings from langchain.vectorstores import Pinecone from langchain.tools.retriever import create_retriever_tool import pinecone from pinecone import Pinecone as PineconeClient, ServerlessSpec from pinecone import Index # the blocking‐call client constructor from pinecone import Pinecone as PineconeClient, ServerlessSpec from langchain.embeddings import HuggingFaceEmbeddings from langchain_community.vectorstores.pinecone import Pinecone as LC_Pinecone # ─── Langchain Frameworks ───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── from langchain.tools import Tool from langchain.chat_models import ChatOpenAI from langchain_groq import ChatGroq from langchain_mistralai import ChatMistralAI from langchain.agents import initialize_agent, AgentType from langchain.schema import Document from langchain.chains import RetrievalQA from langchain.embeddings import OpenAIEmbeddings from langchain_community.embeddings import HuggingFaceEmbeddings from langchain.vectorstores import FAISS from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.prompts import PromptTemplate from langchain_community.document_loaders import TextLoader, PyMuPDFLoader from langchain.document_loaders import WikipediaLoader, ArxivLoader from langchain_experimental.tools.python.tool import PythonREPLTool # ─── File paths ──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # Ensure this is at the very top # ─── SQL Agent ──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── from langchain_community.utilities import SQLDatabase from langchain_community.agent_toolkits import SQLDatabaseToolkit from langchain.chat_models import ChatOpenAI from langgraph.prebuilt import create_react_agent from langchain.agents import create_sql_agent # ─── Memory ─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── from langchain.agents import initialize_agent, AgentType from langchain.tools import Tool from typing import List, Callable from langchain.schema import BaseMemory, AIMessage, HumanMessage, SystemMessage from langchain.schema import HumanMessage, SystemMessage from langchain.llms.base import LLM from langchain.memory.chat_memory import BaseChatMemory from pydantic import PrivateAttr from langchain_core.messages import get_buffer_string # 1) Create your memory object from typing import List from langchain.memory import ConversationBufferMemory class AutoSummaryMemory(ConversationBufferMemory): _llm: LLM = PrivateAttr() _max_entries: int = PrivateAttr() _reduce_to: int = PrivateAttr() _summary_system_prompt: str = PrivateAttr() def __init__( self, llm: LLM, memory_key: str = "chat_history", return_messages: bool = True, max_entries: int = 20, reduce_to: int = 5, summary_system_prompt: str = ( "Summarize the following conversation so far in a concise paragraph. " "Keep important facts and questions." ) ): super().__init__(memory_key=memory_key, return_messages=return_messages) self._llm = llm # PrivateAttr self._max_entries = max_entries # PrivateAttr self._reduce_to = reduce_to # PrivateAttr self._summary_system_prompt = summary_system_prompt # PrivateAttr def add_memory(self, inputs: dict, outputs: dict) -> None: # Add the new turn as normal super().add_memory(inputs=inputs, outputs=outputs) # Check if memory length exceeded msgs = self.chat_memory.messages if len(msgs) >= self._max_entries: full_text = "\n".join([f"{m.type}: {m.content}" for m in msgs]) summary = self._llm.predict(f"{self._summary_system_prompt}\n\n{full_text}") recent = msgs[-self._reduce_to:] self.chat_memory.messages = [ SystemMessage(content="Conversation summary: " + summary), *recent ] # ─── Image Processing ──────────────────────────────────────────────────────────────────────────────────────────────────────────────────── from PIL import Image import pytesseract from transformers import pipeline from groq import Groq import requests from io import BytesIO from PIL import Image from transformers import pipeline, TrOCRProcessor, VisionEncoderDecoderModel from PIL import Image import requests from io import BytesIO import base64 from PIL import UnidentifiedImageError # ─── Browser var ───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── from typing import List, Dict import json from io import BytesIO from langchain.tools import tool # or langchain_core.tools from playwright.sync_api import sync_playwright from duckduckgo_search import DDGS import json import time import random import logging from functools import lru_cache, wraps import requests from playwright.sync_api import sync_playwright from bs4 import BeautifulSoup # Additional imports for new functionality import pandas as pd from PyPDF2 import PdfReader import docx import pytesseract from PIL import Image import speech_recognition as sr from pydub import AudioSegment from pytube import YouTube from newspaper import Article from langchain.document_loaders import ArxivLoader from langchain_community.document_loaders.youtube import YoutubeLoader, TranscriptFormat from playwright.sync_api import sync_playwright # Attempt to import Playwright for dynamic page rendering try: from playwright.sync_api import sync_playwright _playwright_available = True except ImportError: _playwright_available = False # Define forbidden keywords for basic NSFW filtering _forbidden = ["porn", "sex", "xxx", "nude", "erotic"] # ─── LLM Setup ─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # Load OpenAI API key from environment (required for LLM and embeddings) # API Keys from .env file os.environ.setdefault("OPENAI_API_KEY", "") # Set your own key or env var os.environ["GROQ_API_KEY"] = os.getenv("GROQ_API_KEY", "default_key_or_placeholder") os.environ["MISTRAL_API_KEY"] = os.getenv("MISTRAL_API_KEY", "default_key_or_placeholder") # Tavily API Key TAVILY_API_KEY = os.getenv("TAVILY_API_KEY", "default_key_or_placeholder") _forbidden = ["nsfw", "porn", "sex", "explicit"] _playwright_available = True # set False to disable Playwright # Globals for RAG system vector_store = None rag_chain = None DB_PATH = None # will be set when a .db is uploaded DOC_PATH = None # will be set when a document is uploaded IMG_PATH = None # will be set when an image is uploaded OTH_PATH = None # will be set when an other file is uploaded # ─── LLMS ──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── #llm = ChatOpenAI(model_name="gpt-3.5-turbo", streaming=True, temperature=0) llm = ChatGroq(model="deepseek-r1-distill-llama-70b", streaming=False, temperature=0) #llm = ChatMistralAI(model="mistral-large-latest", streaming=True, temperature=0) # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # ─────────────────────────────────────────────── Tool for multiply ────────────────────────────────────────────────────────────────────── # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── @tool def multiply(a: int, b: int) -> int: """Multiply two numbers. Args: a: first int b: second int """ return a * b # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # ─────────────────────────────────────────────── Tool for add ─────────────────────────────────────────────────────────────────────────── # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── @tool def add(a: int, b: int) -> int: """Add two numbers. Args: a: first int b: second int """ return a + b # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # ─────────────────────────────────────────────── Tool for subtract ────────────────────────────────────────────────────────────────────── # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── @tool def subtract(a: int, b: int) -> int: """Subtract two numbers. Args: a: first int b: second int """ return a - b # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # ─────────────────────────────────────────────── Tool for divide ────────────────────────────────────────────────────────────────────── # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── @tool def divide(a: int, b: int) -> int: """Divide two numbers. Args: a: first int b: second int """ if b == 0: raise ValueError("Cannot divide by zero.") return a / b # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # ─────────────────────────────────────────────── Tool for modulus ────────────────────────────────────────────────────────────────────── # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── @tool def modulus(a: int, b: int) -> int: """Get the modulus of two numbers. Args: a: first int b: second int """ return a % b # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # ─────────────────────────────────────────────── Tool for browsing ────────────────────────────────────────────────────────────────────── # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── def with_retry(max_attempts: int = 3, backoff_base: int = 2): """ Decorator for retrying a function with exponential backoff on exception. """ def decorator(fn): @wraps(fn) def wrapper(*args, **kwargs): for attempt in range(max_attempts): try: return fn(*args, **kwargs) except Exception as e: wait = backoff_base ** attempt + random.uniform(0, 1) logger.warning(f"{fn.__name__} failed (attempt {attempt+1}/{max_attempts}): {e}") if attempt < max_attempts - 1: time.sleep(wait) logger.error(f"{fn.__name__} failed after {max_attempts} attempts.") return [] return wrapper return decorator @with_retry() @lru_cache(maxsize=128) def tavily_search(query: str, top_k: int = 3) -> List[Dict]: """Call Tavily API and return a list of result dicts.""" if not TAVILY_API_KEY: logger.info("[Tavily] No API key set. Skipping Tavily search.") return [] url = "https://api.tavily.com/search" headers = { "Authorization": f"Bearer {TAVILY_API_KEY}", "Content-Type": "application/json", } payload = {"query": query, "num_results": top_k} resp = requests.post(url, headers=headers, json=payload, timeout=10) resp.raise_for_status() data = resp.json() results = [] for item in data.get("results", []): results.append({ "title": item.get("title", ""), "url": item.get("url", ""), "content": item.get("content", "")[:200], "source": "Tavily" }) return results @with_retry() @lru_cache(maxsize=128) def duckduckgo_search(query: str, top_k: int = 3) -> List[Dict]: """Query DuckDuckGo and return up to top_k raw SERP hits.""" results = [] with DDGS() as ddgs: for hit in ddgs.text(query, safesearch="On", max_results=top_k): results.append({ "title": hit.get("title", ""), "url": hit.get("href") or hit.get("url", ""), "content": hit.get("body", ""), "source": "DuckDuckGo" }) if len(results) >= top_k: break return results def hybrid_search(query: str, top_k: int = 3) -> List[Dict]: """Combine Tavily + DuckDuckGo search results with fallback.""" results = tavily_search(query, top_k) + duckduckgo_search(query, top_k) if not results: return [{ "title": "No results found", "url": "", "content": f"Could not find suitable web results for '{query}'.", "source": "None" }] return results def format_search_docs(search_docs: List[Dict]) -> Dict[str, str]: """ Turn a list of {source, page, content} dicts into one big string with … entries separated by `---`. """ formatted_search_docs = "\n\n---\n\n".join( [ f'\n' f'{doc.get("content", "")}\n' f'' for doc in search_docs ] ) return {"web_results": formatted_search_docs} @tool def web_search(query: str, top_k: int = 3) -> Dict[str, str]: """ Full hybrid search combining Tavily + DuckDuckGo + scraping. Returns results formatted as XML-like blocks. """ results = [] # Step 1: Scraping first few DDG URLs try: hits = duckduckgo_search(query, top_k) for hit in hits: url = hit.get("url") if not url: continue try: with sync_playwright() as pw: browser = pw.chromium.launch(headless=True) page = browser.new_page() page.goto(url, wait_until="domcontentloaded", timeout=15000) text = page.content() browser.close() soup = BeautifulSoup(text, "html.parser") content = " ".join(soup.get_text(separator=" ", strip=True).split()[:200]) if any(f in content.lower() for f in _forbidden): continue results.append({ "source": url, "page": "", "content": content }) except Exception as e: logger.warning(f"Scraping failed for {url}: {e}") except Exception as e: logger.warning(f"DuckDuckGo scraping phase failed: {e}") # Step 2: Hybrid fallback if len(results) < top_k: fallback = hybrid_search(query, top_k) # Normalize fallback items for doc in fallback: results.append({ "source": doc.get("url", doc.get("source", "")), "page": "", "content": doc.get("content", doc.get("snippet", "")) }) # Step 3: Format and return return format_search_docs(results[:top_k]) ''' def format_search_docs(search_docs: List[Dict]) -> Dict[str,str]: """ Turn a list of {source, page, content} dicts into one big string with … entries separated by `---`. """ formatted_search_docs = "\n\n---\n\n".join( [ # note: .get("page","") in case it's missing f'\n' f'{doc["content"]}\n' f'' for doc in search_docs ] ) return {"web_results": formatted_search_docs} def tavily_search(query: str, top_k: int = 3) -> List[Dict]: """Call Tavily API and return a list of result dicts.""" if not TAVILY_API_KEY: print("[Tavily] No API key set. Skipping Tavily search.") return [] url = "https://api.tavily.com/search" headers = { "Authorization": f"Bearer {TAVILY_API_KEY}", "Content-Type": "application/json", } payload = {"query": query, "num_results": top_k} try: resp = requests.post(url, headers=headers, json=payload, timeout=10) resp.raise_for_status() data = resp.json() results = [] for item in data.get("results", []): results.append({ "title": item.get("title", ""), "url": item.get("url", ""), "snippet": item.get("content", "")[:200], "source": "Tavily" }) return results except (requests.exceptions.RequestException, ValueError) as e: print(f"[Tavily] search failed: {e}") return [] def duckduckgo_search(query: str, top_k: int = 3) -> List[Dict]: """Query DuckDuckGo and return up to top_k raw SERP hits.""" try: results = [] with DDGS() as ddgs: for hit in ddgs.text(query, safesearch="On", max_results=top_k): results.append({ "title": hit.get("title", ""), "url": hit.get("href") or hit.get("url", ""), "snippet": hit.get("body", ""), "source": "DuckDuckGo" }) if len(results) >= top_k: break return results except Exception as e: print(f"[DuckDuckGo] search failed: {e}") return [] def hybrid_web_search(query: str, top_k: int = 3) -> str: """ Returns a JSON string with combined Tavily + DuckDuckGo results. Always returns non-empty JSON with at least a placeholder result. """ tavily = tavily_search(query, top_k) ddg = duckduckgo_search(query, top_k) combined = tavily + ddg if not combined: combined = [{ "title": "No results found", "url": "", "snippet": f"Could not find suitable web results for '{query}'.", "source": "None" }] output = {"query": query, "results": combined} return json.dumps(output, ensure_ascii=False, indent=2) @tool def web_search(query: str, top_k: int = 3) -> Dict[str, str]: """ Full hybrid search with Playwright/BeautifulSoup scraping + Tavily/DuckDuckGo. Returns results formatted as XML-like blocks. """ results: List[Dict] = [] # Step 1: DuckDuckGo + scraping try: with DDGS() as ddgs: hits = ddgs.text(query, safesearch="On", max_results=top_k) except Exception as e: print(f"[web_search] DuckDuckGo lookup failed: {e}") hits = [] for hit in hits: url = hit.get("href") or hit.get("url") if not url: continue try: with sync_playwright() as pw: browser = pw.chromium.launch(headless=True) page = browser.new_page() page.goto(url, wait_until="domcontentloaded", timeout=15000) html = page.content() browser.close() soup = BeautifulSoup(html, "html.parser") text = soup.get_text(separator=" ", strip=True) except Exception as e: print(f"[web_search] scraping failed for {url}: {e}") continue if any(f in text.lower() for f in _forbidden): continue excerpt = " ".join(text.split()[:200]) results.append({ "title": hit.get("title", ""), "url": url, "snippet": hit.get("body", ""), "content": excerpt }) # Step 2: Hybrid JSON fallback try: raw = hybrid_web_search(query, top_k) parsed = json.loads(raw) other = parsed.get("results", []) except Exception as e: print(f"[web_search] parsing hybrid results failed: {e}") other = [] combined = results + other if not combined: combined = [{ "title": "No results found", "url": "", "snippet": f"Could not find suitable content for '{query}'.", "source": "None" }] # Step 3: Format as blocks formatted = "\n\n---\n\n".join( f'\n' f'{doc["content"]}\n' f'' for doc in combined ) return {"web_results": formatted} ''' # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # ─────────────────────────────────────────────── Tool for calculation ───────────────────────────────────────────────────────────────────── # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── @tool def download_file(url: str, dest_path: str) -> str: """Download a file from URL to dest_path and return the local path.""" r = requests.get(url, stream=True) r.raise_for_status() with open(dest_path, 'wb') as f: for chunk in r.iter_content(8192): f.write(chunk) return dest_path @tool def process_excel_to_text(file_path: str) -> str: """Read an Excel file and return its contents as CSV-formatted text.""" df = pd.read_excel(file_path) return df.to_csv(index=False) @tool def read_text_from_pdf(file_path: str) -> str: """Extract and return all text from a PDF file.""" reader = PdfReader(file_path) return "\n".join([page.extract_text() or "" for page in reader.pages]) @tool def read_text_from_docx(file_path: str) -> str: """Extract and return all text from a DOCX file.""" doc = docx.Document(file_path) return "\n".join([para.text for para in doc.paragraphs]) @tool def transcribe_audio(file_path: str) -> str: """Transcribe a local audio file to text using Google Web Speech API.""" recognizer = sr.Recognizer() with sr.AudioFile(file_path) as src: audio = recognizer.record(src) return recognizer.recognize_google(audio) @tool def youtube_audio_processing(youtube_url: str) -> str: """ Download audio from a YouTube video, convert it to WAV, and return the transcription. """ yt = YouTube(youtube_url) audio_stream = yt.streams.filter(only_audio=True).first() out_file = audio_stream.download(output_path='.', filename='yt_audio') wav_path = 'yt_audio.wav' AudioSegment.from_file(out_file).export(wav_path, format='wav') return transcribe_audio(wav_path) @tool def extract_article_text(url: str) -> str: """Download and parse a web article, returning its text.""" art = Article(url) art.download() art.parse() return art.text # ──────────────── ArXiv & YouTube QA tools ──────────────── @tool def arvix_search(query: str) -> Dict[str, str]: """Search ArXiv and return up to 3 papers as JSON.""" papers = ArxivLoader(query=query, load_max_docs=3).load() results = [ { "source": doc.metadata["source"], "id": doc.metadata.get("id", ""), "summary": doc.page_content[:1000], } for doc in papers ] return {"arvix_results": json.dumps(results)} @tool def answer_youtube_video_question( youtube_url: str, question: str, chunk_size_seconds: int = 30 ) -> str: """ Load transcript from a YouTube video, index it, and answer a question. """ loader = YoutubeLoader.from_youtube_url( youtube_url, add_video_info=True, transcript_format=TranscriptFormat.CHUNKS, chunk_size_seconds=chunk_size_seconds, ) documents = loader.load() embeddings = HuggingFaceEmbeddings(model_name='sentence-transformers/all-mpnet-base-v2') vectorstore = FAISS.from_documents(documents, embeddings) llm = ChatGroq(model="deepseek-r1-distill-llama-70b", streaming=False) qa_chain = RetrievalQA.from_chain_type(llm=llm, retriever=vectorstore.as_retriever()) return qa_chain.run(question) # ──────────────── Python REPL tool ──────────────── python_repl = PythonREPLTool() ''' def calculate(expr: str) -> str: """ Evaluates a mathematical expression safely. Uses Python's numexpr for security and speed:contentReference[oaicite:21]{index=21}. """ try: # Allow math constants local_dict = {"pi": math.pi, "e": math.e} # Evaluate expression using numexpr for safety/performance import numexpr result = numexpr.evaluate(expr, local_dict=local_dict) return str(result.item()) except Exception as e: return f"Error calculating expression: {e}" # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # ─────────────────────────────────────────────── Tool for Date and time ─────────────────────────────────────────────────────────────────── # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── def get_current_date(_: str = "") -> str: """ Returns the current date and time. Ignoring input. """ from datetime import datetime return datetime.now().strftime("%Y-%m-%d %H:%M:%S") # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # ─────────────────────────────────────────────── Tool for Wikipedia ─────────────────────────────────────────────────────────────────────── # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── ''' @tool def wiki_search(query: str) -> str: """Search Wikipedia for a query and return up to top_k results as a JSON string.""" # load up to top_k pages pages = WikipediaLoader(query=query, load_max_docs=3).load() results: List[Dict] = [] for doc in pages: results.append({ "source": doc.metadata["source"], "page": doc.metadata.get("page", ""), "content": doc.page_content[:1000], # truncate if you like }) return {"wiki_results": format_search_docs(results)} # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # ───────────────────────────────────────────────────────────── Tool for ArXiv ──────────────────────────────────────────────────────────── # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── ''' @tool def arvix_search(query: str) -> str: """Search Arxiv for a query and return up to top_k results as a JSON string.""" papers = ArxivLoader(query=query, load_max_docs=3).load() results: List[Dict] = [] for doc in papers: results.append({ "source": doc.metadata["source"], "id": doc.metadata.get("id", ""), "summary": doc.page_content[:1000], # just the abstract or first 1000 chars }) return {"arvix_results": format_search_docs(results)} # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # ─────────────────────────────────────────────── Tool for RAG (Document Intelligence) ───────────────────────────────────────────────────── # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── def rag_index_document(DOC_PATH: str) -> str: """ Indexes the given document into the RAG vector store. Supports text files or PDFs. Uses recursive text splitting for better chunking. """ global vector_store, rag_chain text = "" # Read text from file if DOC_PATH and DOC_PATH.lower().endswith(".pdf"): doc = fitz.open(DOC_PATH) for page in doc: text += page.get_text() else: with open(DOC_PATH, 'r', encoding='utf-8') as f: text = f.read() # Split text using recursive text splitter text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # You can adjust this (e.g., 500-1000) chunk_overlap=100 # Overlap for better context between chunks ) # Split into chunks texts = text_splitter.split_text(text) # Create Document objects with metadata docs = [Document(page_content=t, metadata={"source": DOC_PATH}) for t in texts] # Initialize or append to FAISS vector store embeddings = HuggingFaceEmbeddings(model_name='sentence-transformers/all-mpnet-base-v2') if vector_store is None: vector_store = FAISS.from_documents(docs, embeddings) else: vector_store.add_documents(docs) retriever = vector_store.as_retriever( search_type="mmr", search_kwargs={ "k": 10, "fetch_k": 10, "lambda_mult": 0.25 } ) # Build or update the RetrievalQA chain rag_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=retriever, return_source_documents=False ) def rag_answer(query: str) -> str: """ Answers a question using the RAG chain (on indexed documents). """ global rag_chain if rag_chain is None: return "No documents indexed. Please upload documents via /upload_doc." try: answer = rag_chain.run(query) return answer except Exception as e: return f"RAG error: {e}" # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # ───────────────────────────────────── Tool for Image (understading, captioning & classification) ───────────────────────────────────────── # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── ''' def _load_image(img_path: str, resize_to=(512, 512)) -> Image.Image: """ Load, verify, convert, and resize an image. Raises ValueError on failure. """ if not img_path: raise ValueError("No image path provided.") try: with Image.open(img_path) as img: img.verify() img = Image.open(img_path).convert("RGB") img = img.resize(resize_to) return img except UnidentifiedImageError: raise ValueError(f"File at {img_path} is not a valid image.") except Exception as e: raise ValueError(f"Failed to load image at {img_path}: {e}") def _encode_image_to_base64(img_path: str) -> str: """ Load an image, save optimized PNG into memory, and base64‑encode it. """ img = _load_image(img_path) buffer = BytesIO() img.save(buffer, format="PNG", optimize=True) return base64.b64encode(buffer.getvalue()).decode("utf-8") @tool def image_processing(prompt: str, img_path: str) -> str: """ First attempt to process the image with your 4‑vision LLM, falling back to Tesseract OCR on failure. Args: prompt: Instruction or question for the LLM. img_path: Local filesystem path to the image. Returns: The LLM's response or OCR text if the model call fails. """ try: b64 = _encode_image_to_base64(img_path) # Build a single markdown string with inline base64 image md = f"{prompt}\n\n![](data:image/png;base64,{b64})" message = HumanMessage(content=md) llm = ChatGroq(model="meta-llama/llama-4-maverick-17b-128e-instruct", streaming=False, temperature=0) resp = llm.invoke([message]) return resp.content.strip() except Exception as llama_err: print(f"[LLM vision failed] {llama_err}") try: img = _load_image(img_path) return pytesseract.image_to_string(img).strip() except Exception as ocr_err: print(f"[OCR fallback failed] {ocr_err}") return "Unable to process the image. Please check the file and try again." # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # ───────────────────────────────────── Tool for YT video (understading) ─────────────────────────────────────────────────────────────────── # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── ''' @tool def answer_youtube_video_question(youtube_url: str, question: str, chunk_size_seconds: int = 30): """ Load transcript from a YouTube video, chunk it, index it, and answer a question relevant to the video. Args: youtube_url (str): URL of the YouTube video. question (str): The question to ask about the video content. chunk_size_seconds (int): Size of transcript chunks in seconds (default 30). Returns: str: Answer generated by the LLM based on the transcript. """ # Step 1: Load transcript chunks from YouTube video loader = YoutubeLoader.from_youtube_url( youtube_url, add_video_info=True, transcript_format=TranscriptFormat.CHUNKS, chunk_size_seconds=chunk_size_seconds, ) documents = loader.load() # Step 2: Create embeddings and vector store from transcript chunks #embeddings = OpenAIEmbeddings() embeddings = HuggingFaceEmbeddings(model_name='sentence-transformers/all-mpnet-base-v2') vectorstore = FAISS.from_documents(documents, embeddings) # Step 3: Create RetrievalQA chain with OpenAI LLM #llm = OpenAI(temperature=0) llm = ChatGroq(model="deepseek-r1-distill-llama-70b", streaming=False) qa_chain = RetrievalQA.from_chain_type(llm=llm, retriever=vectorstore.as_retriever()) # Step 4: Run the query and return the answer answer = qa_chain.run(question) return answer python_repl = PythonREPLTool() ''' #### Create LangChain Tools #### # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # ─────────────────────────────────────────────── Assigning tools as list ────────────────────────────────────────────────────────────────── # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── ''' tool_list = [ Tool(name="browse", func=web_search, description="Search the web and scrape top results. Uses DuckDuckGo (safe mode) for query. Prefers Playwright for loading pages, with requests/BeautifulSoup as fallback. Filters out any explicit content. Returns JSON with titles, URLs, and page text."), Tool(name="multiply", func=multiply, description="""Multiply two numbers. Args: a: first int b: second int """), Tool(name="divide", func=divide, description="""Divide two numbers. Args: a: first int b: second int """), Tool(name="subtract", func=subtract, description="""Subtract two numbers. Args: a: first int b: second int """), Tool(name="add", func=add, description="""Add two numbers. Args: a: first int b: second int """), #Tool(name="date", func=get_current_date, description="Fetch the current date and time."), Tool(name="wiki", func=wiki_search, description="Search Wikipedia for a query and return up to top_k results as a JSON string."), Tool(name="arvix", func=arvix_search, description="Search Arxiv for a query and return up to top_k results as a JSON string."), Tool( name="python_repl", func=python_repl.run, description="A Python REPL: write and execute arbitrary Python snippets." ), Tool( name="youtube", func=answer_youtube_video_question, description=(" Load transcript from a YouTube video, chunk it, index it, and answer a question relevant to the video." "Args:" " youtube_url (str): URL of the YouTube video." " question (str): The question to ask about the video content." " chunk_size_seconds (int): Size of transcript chunks in seconds (default 30)." "Returns:" " str: Answer generated by the LLM based on the transcript."), ), ] ''' # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # ─────────────────────────────────────────────── Added Memory to Agent ──────────────────────────────────────────────────────────────────── # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # 1) instantiate with your LLM memory = AutoSummaryMemory( llm=llm, max_entries=20, # when chat ≥20 messages, trigger summary reduce_to=5 # keep only last 5 after summarizing ) # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # ─────────────────────────────────────────────── Initialize Agent ───────────────────────────────────────────────────────────────────────── # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # system_prompt # System prompt enforcing planning and final‐answer forma # build a retriever #embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-mpnet-base-v2") # dim=768 #supabase: Client = create_client( # os.environ.get("SUPABASE_URL"), # os.environ.get("SUPABASE_SERVICE_KEY")) #vector_store = SupabaseVectorStore( # client=supabase, # embedding= embeddings, # table_name="documents", # query_name="match_documents_langchain", #) #create_retriever_tool = create_retriever_tool( # retriever=vector_store.as_retriever(), # name="Question Search", # description="A tool to retrieve similar questions from a vector store.", #) # from pinecone import Pinecone as PineconeClient, ServerlessSpec from langchain.embeddings import HuggingFaceEmbeddings from langchain_pinecone import PineconeVectorStore # <- new import # 1) Init Pinecone client (v6+ SDK) pc = PineconeClient( api_key=os.environ["PINECONE_API_KEY"], environment=os.environ.get("PINECONE_ENV") # e.g. "us-east-1" ) # 2) Ensure index exists # ✅ THIS is now valid # ✅ THIS is what LangChain expects index_name = "my-qa-index" if index_name not in pc.list_indexes().names(): pc.create_index( name=index_name, dimension=768, # matches all-mpnet-base-v2 metric="cosine", spec=ServerlessSpec(cloud="aws", region="us-east-1") ) index_name = "my-qa-index" host= "https://my-qa-index-7lt4he8.svc.aped-4627-b74a.pinecone.io" pine_index = pc.Index(name=index_name, host=host) print("pine_index----->",pine_index) embeddings = HuggingFaceEmbeddings( model_name="sentence-transformers/all-mpnet-base-v2" ) vector_store = PineconeVectorStore.from_existing_index( index_name="my-qa-index", embedding=embeddings, text_key="text", # field where your document content lives namespace="default" # or None if you’re not using namespaces ) # 4) Wrap into a retriever tool as before from langchain.tools.retriever import create_retriever_tool retriever = vector_store.as_retriever() pinecone_tool = create_retriever_tool( retriever=retriever, name="Question Search", description="Retrieve similar questions from a Pinecone index." ) tools = [ multiply, add, subtract, divide, modulus, wiki_search, web_search, arvix_search, image_processing, answer_youtube_video_question, extract_article_text, youtube_audio_processing, transcribe_audio, read_text_from_docx, process_excel_to_text, download_file, PythonREPLTool(), pinecone_tool ] llm_with_tools = llm.bind_tools(tools) # Helper to pick tools relevant to a plan step def tools_condition(message: HumanMessage): text = message.content.lower() selected = [t for t in tools if t.__name__.replace("_", " ") in text or t.__name__ in text] return selected or tools[:3] # Build graph function from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage # Debug utility # Bind your LLM and tools once llm = ChatGroq(model="deepseek-r1-distill-llama-70b", temperature=0) llm_tools = llm.bind_tools(tools) sys_msg = SystemMessage(content=""" You are a helpful assistant tasked with answering questions using a set of tools. Now, I will ask you a question. Report your thoughts, and finish your answer with the following template: FINAL ANSWER: [YOUR FINAL ANSWER]. YOUR FINAL ANSWER should be a number OR as few words as possible OR a comma separated list of numbers and/or strings. If you are asked for a number, don't use comma to write your number neither use units such as $ or percent sign unless specified otherwise. If you are asked for a string, don't use articles, neither abbreviations (e.g. for cities), and write the digits in plain text unless specified otherwise. If you are asked for a comma separated list, apply the above rules depending of whether the element to be put in the list is a number or a string. Your answer should only start with "FINAL ANSWER: ", then follows with the answer. """.strip()) def build_graph(provider: str = "groq"): if provider != "groq": raise ValueError("Invalid provider. Expected 'groq'.") # 1) Orchestrator → produces a 'plan' + carries forward 'messages' def orchestrator(state: dict): if not state.get("messages") or not isinstance(state["messages"][0], HumanMessage): return { "plan": "No valid question provided", "messages": [HumanMessage(content="No valid question provided")] } user_q = state["messages"][0].content if not user_q: return { "plan": "Empty question", "messages": [HumanMessage(content="Empty question")] } plan_resp = llm_tools.invoke([ sys_msg, HumanMessage(content=f"Decompose this request into numbered steps:\n\n{user_q}") ]) return { "plan": plan_resp.content if hasattr(plan_resp, 'content') else "", "messages": [HumanMessage(content=user_q)] # Maintain original message } # 2) KnowledgeGraph → optional graph updates (you can persist these) def knowledge_graph(state: dict): kg = [] if "plan" not in state or not state["plan"]: return { "graph": ["No plan available"], "messages": state["messages"] } for step in state["plan"].split("\n"): if not step.strip(): continue ent = llm_tools.invoke([ sys_msg, HumanMessage(f"Extract key entities and relations from this step:\n\n{step}") ]) kg.append(ent.content if hasattr(ent, 'content') else "") return { "graph": kg, "messages": state["messages"] } # 3) Executor → runs each step either via retrieval or LLM def executor(state: dict): results = [] if "plan" not in state or not state["plan"]: return { "results": ["No plan available to execute"], "messages": state["messages"] } for step in state["plan"].split("\n"): if not step.strip(): continue if "search" in step.lower(): hits = vector_store.similarity_search(step, k=3) results.append(hits[0].page_content if hits else "No hits found.") else: resp = llm_tools.invoke(state["messages"] + [HumanMessage(step)]) results.append(resp.content if hasattr(resp, 'content') else "") return { "results": results, "messages": state["messages"] } # 4) Monitor → self-evaluate each result def monitor(state: dict): feedback = [] if "results" not in state: return { "feedback": [("No results available", "0")], "messages": state["messages"] } for r in state["results"]: score = llm_tools.invoke([ sys_msg, HumanMessage(f"On a scale of 1–10, rate the quality of this result:\n\n{r}") ]) feedback.append((r, score.content if hasattr(score, 'content') else "0")) return { "feedback": feedback, "messages": state["messages"] } # 5) Replanner → if any score <5, re-run orchestrator def replanner(state: dict): if "feedback" not in state: return orchestrator(state) low = [r for r, s in state["feedback"] if s.isdigit() and int(s) < 5] if low: return orchestrator(state) return { "messages": state["messages"] } # 6) Assistant → fold in graph and final results into a concise answer def assistant(state: dict): graph_text = "\n".join(state.get("graph", ["No graph available"])) finalize = HumanMessage( content=f"Using these notes:\n{graph_text}\nNow provide the FINAL ANSWER." ) resp = llm_tools.invoke(state["messages"] + [finalize]) return { "messages": [resp] } # 7) Assemble the graph builder = StateGraph(dict) builder.add_node("orchestrator", orchestrator) builder.add_node("knowledge_graph", knowledge_graph) builder.add_node("executor", executor) builder.add_node("monitor", monitor) builder.add_node("replanner", replanner) builder.add_node("assistant", assistant) builder.add_node("tools", ToolNode(tools)) builder.add_edge(START, "orchestrator") builder.add_edge("orchestrator", "knowledge_graph") builder.add_edge("knowledge_graph", "executor") builder.add_edge("executor", "monitor") builder.add_edge("monitor", "replanner") builder.add_edge("replanner", "assistant") builder.add_conditional_edges("assistant", tools_condition) builder.add_edge("tools", "assistant") return builder.compile() def get_final_answer(text): match = re.search(r'(FINAL ANSWER:.*)', text) return match.group(1).strip() if match else f"FINAL ANSWER: {text.strip()}" ############################################################################################################################################################################ ############################################################################################################################################################################ ############################################################################################################################################################################ # Initialize the agent with OpenAI and our tools. We use a zero-shot-react-description agent. ''' agent_executor = initialize_agent( tools=tool_list, llm=llm, agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION, memory=memory, verbose=True, handle_parsing_errors=True, agent_kwargs={"prompt": system_prompt}, #max_iterations=10, ) # ─── Streaming helper ──────────────────────────────────────────────────────────── def run_stream(query: str): """ Progressive token‐by‐token streaming from the agent. Args: query: The user’s natural-language question. data_paths: List of file paths (DB_PATH, DOC_PATH, IMG_PATH, OTH_PATH). """ # Stream the agent response hist = get_buffer_string(memory.chat_memory.messages) print("Memory now contains:", memory.chat_memory.messages) for chunk in agent_executor.stream({"input": query}): text = chunk.get("text") if text: yield text def run_full(query: str) -> str: """ Fallback single‐shot answer (for pure-tool or final completeness). """ return agent_executor.run(query) # Expose for Flask class AgentInterface: def __init__(self, executor): self.executor = executor def run_stream(self, q): return run_stream(q) def run_full(self, q): return run_full(q) agent = AgentInterface(agent_executor) __all__ = [ 'agent_executor', 'run_stream', 'run_full', 'AgentInterface', 'GLOBAL_DB_PATH', 'rag_index_document' ]''' # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # ─────────────────────────────────────────────── Refresh Memory Session ─────────────────────────────────────────────────────────────────── # ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────── # Refresh Memory def refresh_memory(): memory.clear() # clear memory at start of each new session memory.chat_memory.clear() # clear chat history # test if __name__ == "__main__": question = "When was a picture of St. Thomas Aquinas first added to the Wikipedia page on the Principle of double effect?" # Build the graph graph = build_graph(provider="groq") # Run the graph messages = [HumanMessage(content=question)] messages = graph.invoke({"messages": messages}) for m in messages["messages"]: m.pretty_print()