use langchain

=2.0.0

@@ -0,0 +1,15 @@
+Collecting torch
+  Downloading torch-2.7.1-cp312-none-macosx_11_0_arm64.whl.metadata (29 kB)
+Requirement already satisfied: filelock in /Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/site-packages (from torch) (3.16.1)
+Requirement already satisfied: typing-extensions>=4.10.0 in /Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/site-packages (from torch) (4.13.2)
+Requirement already satisfied: setuptools in /Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/site-packages (from torch) (75.1.0)
+Requirement already satisfied: sympy>=1.13.3 in /Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/site-packages (from torch) (1.13.3)
+Requirement already satisfied: networkx in /Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/site-packages (from torch) (3.3)
+Requirement already satisfied: jinja2 in /Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/site-packages (from torch) (3.1.4)
+Requirement already satisfied: fsspec in /Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/site-packages (from torch) (2024.3.1)
+Requirement already satisfied: mpmath<1.4,>=1.1.0 in /Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/site-packages (from sympy>=1.13.3->torch) (1.3.0)
+Requirement already satisfied: MarkupSafe>=2.0 in /Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/site-packages (from jinja2->torch) (2.1.5)
+Downloading torch-2.7.1-cp312-none-macosx_11_0_arm64.whl (68.6 MB)
+   ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 68.6/68.6 MB 53.9 MB/s eta 0:00:00
+Installing collected packages: torch
+Successfully installed torch-2.7.1

agent.py

@@ -1,11 +1,6 @@
 """
-AI AGENT WITH LANGGRAPH + UTILS SYSTEM
-
-Architecture:
-- LangChain/LangGraph workflow với AI-driven routing
-- Qwen3-8B làm main reasoning engine  
-- Utils system cung cấp tools
-- AI tự quyết định tools và logic xử lý
+AI AGENT WITH LANGGRAPH + HUGGINGFACE INTEGRATION
+Clean architecture with LangChain HuggingFace Pipeline
 """
 
 import os
@@ -14,26 +9,20 @@ import time
 from typing import Dict, Any, List, Optional, Annotated
 from dotenv import load_dotenv
 
-# LangChain imports
 from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
 from langchain_core.prompts import ChatPromptTemplate
 from langchain_core.output_parsers import JsonOutputParser, StrOutputParser
 
-# LangGraph imports  
 from langgraph.graph import StateGraph, END
 from langgraph.graph.message import add_messages
 from typing_extensions import TypedDict
 
-# HuggingFace imports
-from huggingface_hub import InferenceClient
-
-# Groq imports for fallback
-from groq import Groq
-
-# Pydantic for structured output
 from pydantic import BaseModel, Field
 
-# Utils system imports
+# LangChain HuggingFace Integration
+from langchain_huggingface import HuggingFacePipeline, ChatHuggingFace, HuggingFaceEndpoint
+from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
+
 from utils import (
     process_question_with_tools,
     get_agent_state,
@@ -45,407 +34,274 @@ from utils import (
     analyze_question_type
 )
 
-# Load environment
 load_dotenv()
 
-# =============================================================================
-# LANGGRAPH STATE DEFINITION
-# =============================================================================
-
 class AgentState(TypedDict):
-    """LangGraph state for AI agent"""
     messages: Annotated[List, add_messages]
     question: str
     task_id: str
-    
-    # AI Analysis
     ai_analysis: Dict[str, Any]
     should_use_tools: bool
-    
-    # Tool processing
     tool_processing_result: Dict[str, Any]
-    
-    # Final response
     final_answer: str
     processing_complete: bool
 
-# =============================================================================
-# PYDANTIC SCHEMAS FOR STRUCTURED OUTPUT
-# =============================================================================
-
 class QuestionAnalysis(BaseModel):
-    """Schema for AI question analysis"""
     question_type: str = Field(description="Type: youtube|image|audio|wiki|file|text|math")
     needs_tools: bool = Field(description="Whether tools are needed")
     reasoning: str = Field(description="AI reasoning for the decision")
     confidence: str = Field(description="Confidence level: high|medium|low")
-    can_answer_directly: bool = Field(description="Can answer without tools")
-    suggested_approach: str = Field(description="Brief description of approach")
 
-class TextDecision(BaseModel):
-    """Schema for reversed text decision"""
-    chosen_version: str = Field(description="original|reversed")
-    reasoning: str = Field(description="Reasoning for the choice")
-    confidence: str = Field(description="Confidence level: high|medium|low")
-
-# =============================================================================
-# AI BRAIN WITH LANGCHAIN
-# =============================================================================
-
-class LangChainQwen3Brain:
-    """AI Brain using LangChain + HuggingFace with Groq fallback"""
-    
+class AIBrain:
     def __init__(self):
-        # Primary: HuggingFace
-        self.hf_client = InferenceClient(
-            provider="auto", 
-            api_key=os.environ.get("HF_TOKEN", "")
-        )
-        self.hf_model = "Qwen/Qwen3-8B"
+        self.model_name = "Qwen/Qwen3-8B"
         
-        # Fallback: Groq  
-        self.groq_client = Groq(
-            api_key=os.environ.get("GROQ_API_KEY", "")
+        print("🧠 Initializing Qwen3-8B with LangChain HuggingFace...")
+        
+        # Load tokenizer with thinking disabled
+        self.tokenizer = AutoTokenizer.from_pretrained(self.model_name)
+        
+        # Create text generation pipeline with Qwen3
+        self.hf_pipeline = pipeline(
+            "text-generation",
+            model=self.model_name,
+            tokenizer=self.tokenizer,
+            torch_dtype="auto",
+            device_map="auto",
+            max_new_tokens=2048,
+            temperature=0.7,
+            top_p=0.9,
+            do_sample=True,
+            pad_token_id=self.tokenizer.eos_token_id if self.tokenizer.eos_token_id else self.tokenizer.pad_token_id
         )
-        self.groq_model = "llama3-8b-8192"
         
-        # Setup parsers
-        self.json_parser = JsonOutputParser()
-        self.str_parser = StrOutputParser()
+        # Wrap with LangChain HuggingFacePipeline
+        self.llm = HuggingFacePipeline(pipeline=self.hf_pipeline)
         
-        print("🧠 LangChain Hybrid Brain initialized (HF + Groq fallback)")
-    
-    def _create_structured_model(self, schema: BaseModel):
-        """Create model with structured output"""
-        try:
-            # Try HuggingFace with structured output
-            from langchain_huggingface import ChatHuggingFace
-            hf_model = ChatHuggingFace(
-                llm=self.hf_client,
-                model_id=self.hf_model
-            )
-            return hf_model.with_structured_output(schema)
-        except Exception as hf_error:
-            print(f"⚠️ HF structured output failed: {str(hf_error)[:50]}...")
-            try:
-                # Fallback to Groq with structured output
-                from langchain_groq import ChatGroq
-                groq_model = ChatGroq(
-                    api_key=os.environ.get("GROQ_API_KEY", ""),
-                    model=self.groq_model
-                )
-                return groq_model.with_structured_output(schema)
-            except Exception as groq_error:
-                print(f"⚠️ Both structured output failed")
-                return None
-
-    def _invoke_model(self, messages: List[Dict[str, str]]) -> str:
-        """Invoke model with messages - try HF first, fallback to Groq"""
+        # Create ChatHuggingFace for chat interface
+        self.chat_model = ChatHuggingFace(llm=self.llm)
         
-        # Try HuggingFace first
+        print("✅ Qwen3 AI Brain with LangChain HuggingFace initialized")
+    
+    def _generate_with_qwen3(self, prompt: str, max_tokens: int = 2048) -> str:
+        """Generate text with Qwen3 via LangChain - thinking disabled"""
         try:
-            completion = self.hf_client.chat.completions.create(
-                model=self.hf_model,
-                messages=messages,
-                max_tokens=2048,
-                temperature=0.7
+            # Prepare messages for chat template with thinking DISABLED
+            messages = [{"role": "user", "content": prompt}]
+            
+            # Apply chat template with enable_thinking=False
+            text = self.tokenizer.apply_chat_template(
+                messages,
+                tokenize=False,
+                add_generation_prompt=True,
+                enable_thinking=False  # CRITICAL: Disable thinking mode
             )
-            return completion.choices[0].message.content
-        except Exception as hf_error:
-            print(f"⚠️ HuggingFace failed: {str(hf_error)[:100]}...")
-            print("🔄 Falling back to Groq...")
             
-            # Fallback to Groq
+            # Use LangChain HuggingFace pipeline for generation
+            response = self.llm.invoke(text)
+            
+            # Clean up response - remove input prompt
+            if text in response:
+                response = response.replace(text, "").strip()
+            
+            return response
+            
+        except Exception as e:
+            print(f"⚠️ Qwen3 generation error: {str(e)}")
+            # Fallback to direct pipeline call
             try:
-                completion = self.groq_client.chat.completions.create(
-                    model=self.groq_model,
-                    messages=messages,
-                    max_tokens=2048,
-                    temperature=0.7
-                )
-                return completion.choices[0].message.content
-            except Exception as groq_error:
-                return f"AI Error: Both HF ({str(hf_error)[:50]}) and Groq ({str(groq_error)[:50]}) failed"
-
+                result = self.hf_pipeline(prompt, max_new_tokens=max_tokens)
+                return result[0]['generated_text'].replace(prompt, "").strip()
+            except Exception as e2:
+                return f"AI generation failed: {str(e2)}"
+    
     def analyze_question(self, question: str, task_id: str = "") -> Dict[str, Any]:
-        """AI analyzes question and decides approach with structured output"""
+        """Analyze question type using Qwen3 with strict JSON output"""
         
-        # Create structured model
-        structured_model = self._create_structured_model(QuestionAnalysis)
+        prompt = f"""<instruction>
+Analyze this question and determine the correct tool approach. Return ONLY valid JSON.
+</instruction>
+
+<question>{question}</question>
+<task_id>{task_id}</task_id>
+
+<classification_rules>
+- YouTube URLs (youtube.com, youtu.be): "youtube"
+- Images, photos, chess positions, visual content: "image" 
+- Audio files, voice, sound, mp3: "audio"
+- Excel, CSV, documents, file uploads: "file"
+- Wikipedia searches, historical facts, people info: "wiki"
+- Math calculations, logic, text analysis: "text"
+</classification_rules>
+
+Return this exact JSON format:
+{{
+    "question_type": "youtube|image|audio|wiki|file|text",
+    "needs_tools": true,
+    "reasoning": "Brief explanation of classification",
+    "confidence": "high"
+}}"""
         
-        if structured_model:
-            analysis_prompt = f"""
-Analyze this question and decide the approach:
-
-Question: "{question}"
-Task ID: "{task_id}"
-
-Important rules:
-- If question asks about Mercedes Sosa albums, Wikipedia, historical facts -> use "wiki"
-- If YouTube URL present -> use "youtube"  
-- If mentions image, photo, chess position -> use "image"
-- If mentions audio, voice, mp3 -> use "audio"
-- If mentions file attachment, Excel, CSV -> use "file"
-- For math, tables, logic problems -> use "text" but needs_tools=false
-- Be accurate about question_type to trigger correct tools
-
-/no_thinking
-"""
+        try:
+            response = self._generate_with_qwen3(prompt, 512)
             
-            try:
-                result = structured_model.invoke(analysis_prompt)
-                return result.dict()
-            except Exception as e:
-                print(f"⚠️ Structured analysis failed: {str(e)[:50]}...")
-        
-        # Fallback analysis
-        question_type = analyze_question_type(question)
-        return {
-            "question_type": question_type,
-            "needs_tools": bool(task_id) or question_type in ["wiki", "youtube", "image", "audio", "file"],
-            "reasoning": "Fallback analysis - structured output failed",
-            "confidence": "medium",
-            "can_answer_directly": question_type == "text" and not task_id,
-            "suggested_approach": f"Use {question_type} processing"
-        }
+            # Extract JSON from response
+            import re
+            json_pattern = r'\{[^{}]*\}'
+            json_match = re.search(json_pattern, response)
+            
+            if json_match:
+                result = json.loads(json_match.group())
+                
+                # Validate required fields
+                required_fields = ["question_type", "needs_tools", "reasoning", "confidence"]
+                if all(field in result for field in required_fields):
+                    return result
+                    
+            raise ValueError("Invalid JSON structure in response")
+            
+        except Exception as e:
+            print(f"⚠️ Qwen3 analysis failed: {str(e)[:100]}...")
+            
+            # Fallback analysis
+            question_type = analyze_question_type(question)
+            return {
+                "question_type": question_type,
+                "needs_tools": question_type in ["wiki", "youtube", "image", "audio", "file"],
+                "reasoning": f"Fallback classification: detected {question_type}",
+                "confidence": "medium"
+            }
     
-    def generate_final_answer(self, question: str, tool_results: Dict[str, Any], context: str = "") -> str:
-        """Generate final answer using LangChain"""
+    def generate_answer(self, question: str, tool_results: Dict[str, Any]) -> str:
+        """Generate final answer using Qwen3 with context"""
         
-        # Build context summary
         if tool_results and tool_results.get("tool_results"):
-            context_summary = build_context_summary(
+            context = build_context_summary(
                 tool_results.get("tool_results", []),
                 tool_results.get("cached_data", {})
             )
         else:
-            context_summary = context or "No additional context available"
-        
-        answer_prompt = get_response_prompt(
-            "final_answer",
-            question=question,
-            context_summary=context_summary
-        ) + "\n\n/no_thinking"
-        
-        messages = [
-            {"role": "system", "content": get_system_prompt("reasoning_agent")},
-            {"role": "user", "content": answer_prompt}
-        ]
-        
-        return self._invoke_model(messages)
-    
-    def decide_on_reversed_text(self, original: str, reversed: str) -> Dict[str, Any]:
-        """AI decides which version of text to use with structured output"""
+            context = "No additional context available"
         
-        # Create structured model
-        structured_model = self._create_structured_model(TextDecision)
-        
-        if structured_model:
-            decision_prompt = f"""
-You are analyzing two versions of the same text to determine which makes more sense:
+        prompt = f"""<instruction>
+Generate a comprehensive answer to the user's question using the provided context.
+</instruction>
 
-Original: "{original}"
-Reversed: "{reversed}"
+<question>{question}</question>
 
-Analyze both versions and decide which one is more likely to be the correct question.
-Consider grammar, word order, and meaning.
+<context>
+{context}
+</context>
 
-/no_thinking
-"""
-            
-            try:
-                result = structured_model.invoke(decision_prompt)
-                return result.dict()
-            except Exception as e:
-                print(f"⚠️ Structured decision failed: {str(e)[:50]}...")
-        
-        # Fallback decision
-        return {
-            "chosen_version": "reversed" if len(reversed.split()) > 3 else "original",
-            "reasoning": "Fallback decision based on text structure",
-            "confidence": "low"
-        }
+<output_rules>
+- Provide direct, accurate answers
+- Use context information when relevant
+- Be concise but complete
+- No thinking process in output
+- Professional tone
+</output_rules>
 
-# =============================================================================
-# LANGGRAPH NODES
-# =============================================================================
+Answer:"""
+        
+        response = self._generate_with_qwen3(prompt, 2048)
+        
+        # Clean up response
+        if "Answer:" in response:
+            response = response.split("Answer:")[-1].strip()
+        
+        return response
 
-# Initialize AI brain
-ai_brain = LangChainQwen3Brain()
+# Initialize AI Brain globally
+ai_brain = AIBrain()
 
 def analyze_question_node(state: AgentState) -> AgentState:
-    """AI analyzes the question and decides approach"""
+    """Analyze question using Qwen3 AI Brain"""
     question = state["question"]
     task_id = state.get("task_id", "")
     
-    print(f"🔍 AI analyzing question: {question[:50]}...")
-    
-    # Get AI analysis
+    print("🔍 Analyzing question with Qwen3...")
     analysis = ai_brain.analyze_question(question, task_id)
-    state["ai_analysis"] = analysis
     
-    # Determine if tools are needed
+    state["ai_analysis"] = analysis
     state["should_use_tools"] = analysis.get("needs_tools", True)
     
-    print(f"📊 AI Analysis:")
-    print(f"   Type: {analysis.get('question_type', 'unknown')}")
-    print(f"   Needs tools: {analysis.get('needs_tools', True)}")
-    print(f"   Confidence: {analysis.get('confidence', 'medium')}")
-    print(f"   Reasoning: {analysis.get('reasoning', 'No reasoning provided')}")
-    
+    print(f"📊 Type: {analysis.get('question_type')} | Tools: {analysis.get('needs_tools')} | Confidence: {analysis.get('confidence')}")
     return state
 
 def process_with_tools_node(state: AgentState) -> AgentState:
-    """Process question using utils tool system"""
+    """Process question with appropriate tools"""
     question = state["question"]
     task_id = state.get("task_id", "")
     
-    print(f"🔧 Processing with tools...")
+    print("🔧 Processing with specialized tools...")
+    tool_results = process_question_with_tools(question, task_id)
+    state["tool_processing_result"] = tool_results
     
-    try:
-        # Use utils tool orchestrator
-        result = process_question_with_tools(question, task_id)
-        state["tool_processing_result"] = result
-        
-        print(f"✅ Tool processing completed:")
-        print(f"   Question type: {result.get('question_type', 'unknown')}")
-        print(f"   Successful tools: {result.get('successful_tools', [])}")
-        print(f"   Failed tools: {result.get('failed_tools', [])}")
-        
-    except Exception as e:
-        print(f"❌ Tool processing failed: {str(e)}")
-        state["tool_processing_result"] = {
-            "error": str(e),
-            "processed_question": question,
-            "question_type": "error",
-            "tools_used": [],
-            "successful_tools": [],
-            "failed_tools": [],
-            "tool_results": [],
-            "cached_data": {}
-        }
+    successful_tools = [result.tool_name for result in tool_results.get("tool_results", []) if result.success]
+    if successful_tools:
+        print(f"✅ Successful tools: {successful_tools}")
+    else:
+        print("⚠️ No tools succeeded")
     
     return state
 
 def answer_directly_node(state: AgentState) -> AgentState:
-    """Answer question directly without tools"""
+    """Answer directly without tools using Qwen3"""
     question = state["question"]
     
-    print(f"💭 AI answering directly...")
-    
-    # Generate direct answer
-    direct_prompt = f"""
-Answer this question directly based on your knowledge:
-
-Question: {question}
-
-Provide a clear, accurate, and helpful answer.
-"""
-    
-    messages = [
-        {"role": "system", "content": get_system_prompt("reasoning_agent")},
-        {"role": "user", "content": direct_prompt}
-    ]
-    
-    answer = ai_brain._invoke_model(messages)
+    print("💭 Generating direct answer with Qwen3...")
+    answer = ai_brain.generate_answer(question, {})
     state["final_answer"] = answer
     state["processing_complete"] = True
     
     return state
 
 def generate_final_answer_node(state: AgentState) -> AgentState:
-    """Generate final answer using AI + tool results"""
+    """Generate final answer combining tool results and AI analysis"""
     question = state["question"]
     tool_results = state.get("tool_processing_result", {})
     
-    print(f"🎯 Generating final answer...")
-    
-    # Generate comprehensive answer
-    answer = ai_brain.generate_final_answer(question, tool_results)
-    
+    print("🎯 Generating final answer with context...")
+    answer = ai_brain.generate_answer(question, tool_results)
     state["final_answer"] = answer
     state["processing_complete"] = True
     
-    print(f"✅ Final answer generated")
-    
     return state
 
-# =============================================================================
-# LANGGRAPH WORKFLOW
-# =============================================================================
-
 def create_agent_workflow():
-    """Create LangGraph workflow"""
-    
+    """Create LangGraph workflow for question processing"""
     workflow = StateGraph(AgentState)
     
     # Add nodes
-    workflow.add_node("analyze", analyze_question_node)
-    workflow.add_node("use_tools", process_with_tools_node)
-    workflow.add_node("direct_answer", answer_directly_node)
-    workflow.add_node("generate_answer", generate_final_answer_node)
+    workflow.add_node("analyze_question", analyze_question_node)
+    workflow.add_node("process_with_tools", process_with_tools_node) 
+    workflow.add_node("answer_directly", answer_directly_node)
+    workflow.add_node("generate_final_answer", generate_final_answer_node)
     
-    # Routing logic
+    # Define routing logic
     def should_use_tools(state: AgentState) -> str:
-        """AI-driven routing decision"""
-        should_use = state.get("should_use_tools", True)
-        can_answer_directly = state.get("ai_analysis", {}).get("can_answer_directly", False)
-        
-        if can_answer_directly and not should_use:
-            print("🚀 AI decided to answer directly")
-            return "direct_answer"
-        else:
-            print("🔧 AI decided to use tools")
-            return "use_tools"
-    
-    # Add conditional edges
-    workflow.add_conditional_edges(
-        "analyze",
-        should_use_tools,
-        {
-            "use_tools": "use_tools",
-            "direct_answer": "direct_answer"
-        }
-    )
-    
-    # Connect tool processing to final answer
-    workflow.add_edge("use_tools", "generate_answer")
-    
-    # End edges
-    workflow.add_edge("direct_answer", END)
-    workflow.add_edge("generate_answer", END)
+        return "process_with_tools" if state.get("should_use_tools", True) else "answer_directly"
     
-    # Set entry point
-    workflow.set_entry_point("analyze")
+    # Set up the flow
+    workflow.set_entry_point("analyze_question")
+    workflow.add_conditional_edges("analyze_question", should_use_tools)
+    workflow.add_edge("process_with_tools", "generate_final_answer")
+    workflow.add_edge("answer_directly", END)
+    workflow.add_edge("generate_final_answer", END)
     
     return workflow.compile()
 
-# =============================================================================
-# MAIN AGENT CLASS
-# =============================================================================
-
 class LangGraphUtilsAgent:
-    """Main AI Agent using LangGraph + Utils system"""
-    
     def __init__(self):
-        self.workflow = create_agent_workflow()
-        self.ai_brain = ai_brain
-        
-        print("🤖 LangGraph Utils Agent initialized!")
-        print("🧠 AI Brain: LangChain + HuggingFace with Groq fallback")
-        print("🔧 Tools: YouTube, Image OCR, Audio Transcript, Wikipedia, File Reader, Text Processor")
-        print("⚡ Features: AI-driven routing, Smart tool selection, Multimodal processing")
+        self.app = create_agent_workflow()
+        print("🚀 LangGraph Agent with Qwen3 + Utils System ready")
     
     def process_question(self, question: str, task_id: str = "") -> str:
-        """Main entry point for processing questions"""
+        """Process question through the workflow"""
         try:
-            print(f"\n🚀 Processing question: {question}")
-            print(f"📄 Task ID: {task_id or 'None'}")
-            
-            # Reset agent state for new question
-            reset_agent_state()
+            print(f"\n🎯 Processing: {question[:100]}...")
             
-            # Initialize LangGraph state
+            # Initialize state
             initial_state = {
                 "messages": [HumanMessage(content=question)],
                 "question": question,
@@ -457,38 +313,29 @@ class LangGraphUtilsAgent:
                 "processing_complete": False
             }
             
-            # Execute workflow
-            print("\n🔄 Starting LangGraph workflow...")
+            # Run workflow
             start_time = time.time()
+            result = self.app.invoke(initial_state)
+            elapsed_time = time.time() - start_time
             
-            final_state = self.workflow.invoke(initial_state)
-            
-            execution_time = time.time() - start_time
-            print(f"\n⏱️ Total execution time: {execution_time:.2f} seconds")
+            final_answer = result.get("final_answer", "No answer generated")
+            print(f"✅ Completed in {elapsed_time:.2f}s")
             
-            # Return final answer
-            answer = final_state.get("final_answer", "No answer generated")
-            
-            print(f"\n✅ Question processed successfully!")
-            return answer
+            return final_answer
             
         except Exception as e:
-            error_msg = f"Agent processing error: {str(e)}"
-            print(f"\n❌ {error_msg}")
-            import traceback
-            traceback.print_exc()
-            return error_msg
-
-# =============================================================================
-# GLOBAL AGENT INSTANCE
-# =============================================================================
+            print(f"❌ Agent error: {str(e)}")
+            return f"I apologize, but I encountered an error processing your question: {str(e)}"
 
-# Create global agent
+# Global agent instance
 agent = LangGraphUtilsAgent()
 
 def process_question(question: str, task_id: str = "") -> str:
-    """Global function for processing questions"""
-    return agent.process_question(question, task_id)
+    """Main entry point for question processing"""
+    if not question or not question.strip():
+        return "Please provide a valid question."
+    
+    return agent.process_question(question.strip(), task_id)
 
 # =============================================================================
 # TESTING
@@ -534,4 +381,13 @@ if __name__ == "__main__":
         
         print(f"\n{'-'*60}")
     
-    print("\n✅ All tests completed!") 
+    print("\n✅ All tests completed!")
+
+# Initialize Qwen3 with thinking mode disabled
+primary_brain = HuggingFaceEndpoint(
+    repo_id=primary_model,
+    temperature=0.7,
+    max_new_tokens=300,
+    huggingfacehub_api_token=os.getenv("HF_API_KEY"),
+    model_kwargs={"enable_thinking": False, "thinking_prompt": "/no_thinking"}
+) 

requirements.txt

@@ -1,26 +1,38 @@
-gradio==4.44.0
-huggingface-hub==0.33.1
-groq==0.13.0
-python-dotenv==1.0.0
-requests==2.31.0
-wikipedia==1.4.0
-pandas==2.0.3
-openpyxl==3.1.2
-Pillow==10.0.0
-yt-dlp>=2024.12.23
-
-# LangChain + LangGraph for AI-driven workflows
-langchain==0.3.13
-langchain-core==0.3.29
-langgraph==0.2.61
+# AI Core Dependencies - LangChain HuggingFace Integration
+langchain>=0.3.0
 langchain-huggingface>=0.1.0
-langchain-groq>=0.2.0
+langchain-groq>=0.2.0  
+langgraph>=0.2.0
+
+# HuggingFace Core
+transformers>=4.51.0
+
+# Tool Dependencies  
+groq>=0.11.0
+wikipedia>=1.4.0
+requests>=2.31.0
+pillow>=10.0.0
+python-dotenv>=1.0.0
 pydantic>=2.0.0
 
-# Transformers for multimodal models
-transformers>=4.44.0
-torch>=2.0.0
-accelerate>=0.21.0
+# YouTube Tools
+yt-dlp>=2024.1.0
+pytube>=15.0.0
+youtube-transcript-api>=0.6.0
+
+# File Processing
+pandas>=2.0.0
+openpyxl>=3.1.0
+
+# Audio/Image Processing  
+openai-whisper>=20231117
+opencv-python>=4.8.0
+
+# Optional optimizations
+typing-extensions>=4.8.0
+
+# LangChain + LangGraph for AI-driven workflows
+langchain-core==0.3.29
 
 # Optional dependencies for enhanced functionality
 # flash-attn>=2.0.0  # For flash attention (requires CUDA)

utils/__init__.py

@@ -2,8 +2,8 @@
 Utils package for AI Agent tools
 """
 
-# Individual tools
-from .youtube_tool import get_youtube_content
+# Individual tools  
+from .youtube_tool import YouTubeTool
 from .text_tool import reverse_text_if_needed
 from .image_tool import ocr_image_with_nanonets
 from .audio_tool import transcribe_audio_groq
@@ -30,7 +30,7 @@ from .tool_orchestrator import (
 
 __all__ = [
     # Individual tools
-    "get_youtube_content",
+    "YouTubeTool",
     "reverse_text_if_needed", 
     "ocr_image_with_nanonets",
     "transcribe_audio_groq",

utils/audio_tool.py

@@ -1,11 +1,139 @@
 """
-Audio Tool - Transcribe audio với Groq Whisper API
+AUDIO PROCESSING TOOL - Groq Audio Only
+Handles audio file transcription using Groq Whisper API
 """
 
 import os
 import tempfile
 import requests
-from typing import Optional
+from typing import Dict, Any, Optional
+from groq import Groq
+from .state_manager import get_agent_state
+
+class AudioTool:
+    def __init__(self):
+        self.client = Groq(api_key=os.environ.get("GROQ_API_KEY", ""))
+        self.model = "whisper-large-v3"
+        print("🎵 Audio Tool (Groq Whisper) initialized")
+    
+    def process_audio(self, audio_input: str, **kwargs) -> Dict[str, Any]:
+        """
+        Process audio files using Groq Whisper API
+        Supports URLs, file paths, and base64 audio
+        """
+        try:
+            audio_path = self._prepare_audio_file(audio_input)
+            if not audio_path:
+                return self._error_result("Could not prepare audio file")
+            
+            # Transcribe using Groq Whisper
+            transcript = self._transcribe_with_groq(audio_path)
+            
+            # Cleanup temp file if created
+            if audio_path.startswith(tempfile.gettempdir()):
+                os.unlink(audio_path)
+            
+            result = {
+                "transcript": transcript,
+                "source": audio_input,
+                "model": self.model,
+                "tool": "groq_whisper"
+            }
+            
+            # Update agent state
+            state = get_agent_state()
+            state.cached_data["audio_analysis"] = result
+            
+            return {
+                "success": True,
+                "data": result,
+                "summary": f"Audio transcribed: {transcript[:100]}..."
+            }
+            
+        except Exception as e:
+            error_msg = f"Audio processing failed: {str(e)}"
+            print(f"❌ {error_msg}")
+            return self._error_result(error_msg)
+    
+    def _prepare_audio_file(self, audio_input: str) -> Optional[str]:
+        """Prepare audio file for processing"""
+        try:
+            # If it's a URL, download it
+            if audio_input.startswith(('http://', 'https://')):
+                return self._download_audio(audio_input)
+            
+            # If it's a local file path
+            if os.path.exists(audio_input):
+                return audio_input
+            
+            # If it's base64, decode it
+            if self._is_base64(audio_input):
+                return self._decode_base64_audio(audio_input)
+            
+            return None
+            
+        except Exception as e:
+            print(f"⚠️ Audio prep error: {str(e)}")
+            return None
+    
+    def _download_audio(self, url: str) -> str:
+        """Download audio from URL to temp file"""
+        response = requests.get(url, stream=True)
+        response.raise_for_status()
+        
+        # Create temp file with audio extension
+        suffix = '.mp3'  # Default
+        if '.' in url:
+            suffix = '.' + url.split('.')[-1].split('?')[0]
+        
+        with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp_file:
+            for chunk in response.iter_content(chunk_size=8192):
+                tmp_file.write(chunk)
+            return tmp_file.name
+    
+    def _is_base64(self, s: str) -> bool:
+        """Check if string is base64 encoded"""
+        import base64
+        try:
+            if isinstance(s, str):
+                s_bytes = bytes(s, 'ascii')
+            elif isinstance(s, bytes):
+                s_bytes = s
+            else:
+                return False
+            return base64.b64encode(base64.b64decode(s_bytes)) == s_bytes
+        except Exception:
+            return False
+    
+    def _decode_base64_audio(self, b64_string: str) -> str:
+        """Decode base64 audio to temp file"""
+        import base64
+        
+        audio_data = base64.b64decode(b64_string)
+        with tempfile.NamedTemporaryFile(delete=False, suffix='.mp3') as tmp_file:
+            tmp_file.write(audio_data)
+            return tmp_file.name
+    
+    def _transcribe_with_groq(self, audio_path: str) -> str:
+        """Transcribe audio using Groq Whisper API"""
+        with open(audio_path, "rb") as audio_file:
+            transcript = self.client.audio.transcriptions.create(
+                file=audio_file,
+                model=self.model,
+                language="en",  # Auto-detect or specify
+                response_format="text"
+            )
+        
+        return transcript if isinstance(transcript, str) else transcript.text
+    
+    def _error_result(self, error_msg: str) -> Dict[str, Any]:
+        """Standard error result format"""
+        return {
+            "success": False,
+            "error": error_msg,
+            "data": None,
+            "summary": f"Audio processing failed: {error_msg}"
+        }
 
 def download_audio_file(task_id: str) -> Optional[str]:
     """
@@ -58,7 +186,6 @@ def transcribe_audio_groq(task_id: str = "", audio_path: str = "", language: str
     
     try:
         # Initialize Groq client
-        from groq import Groq
         groq_api_key = os.environ.get("GROQ_API_KEY")
         
         if not groq_api_key:

utils/text_tool.py

@@ -1,100 +1,43 @@
 """
-Text Tool - Xử lý câu hỏi viết ngược
+Text Tool - Process reversed text questions
 """
 
 from typing import Dict, Any
 
-def reverse_text(text: str) -> str:
-    """
-    Đảo ngược text
-    """
-    return text[::-1]
-
 def is_likely_reversed(text: str) -> bool:
-    """
-    Kiểm tra xem text có khả năng bị viết ngược không
-    Dựa trên các dấu hiệu:
-    - Câu kết thúc bằng dấu phẩy thay vì dấu chấm hỏi
-    - Có từ "rewsna" (answer ngược)
-    - Có từ kết thúc bằng các ký tự đặc biệt
-    """
-    # Dấu hiệu câu hỏi viết ngược
-    reverse_indicators = [
-        "rewsna",  # "answer" ngược
-        "noitseuq",  # "question" ngược  
-        "ecnetnes",  # "sentence" ngược
-        "dnatsrednu",  # "understand" ngược
-        "etirw",  # "write" ngược
-        text.strip().endswith(","),  # Kết thúc bằng dấu phẩy
-        text.strip().startswith("?"),  # Bắt đầu bằng dấu hỏi
-    ]
+    words = text.split()
+    if len(words) < 3:
+        return False
     
-    # Đếm số dấu hiệu
-    indicators_found = sum([
-        1 for indicator in reverse_indicators 
-        if (isinstance(indicator, str) and indicator.lower() in text.lower()) or 
-           (isinstance(indicator, bool) and indicator)
-    ])
+    reversed_indicators = [
+        text.startswith('.'),
+        text.endswith('fI'),
+        'ecnetnes' in text,
+        'rewsna' in text,
+        'noitseuq' in text
+    ]
     
-    # Nếu có >= 2 dấu hiệu thì có thể là text ngược
-    return indicators_found >= 2
+    return sum(reversed_indicators) >= 2
+
+def reverse_text(text: str) -> str:
+    return text[::-1]
 
 def reverse_text_if_needed(question: str, ai_brain=None) -> Dict[str, Any]:
-    """
-    Main function: Kiểm tra và cung cấp thông tin về câu hỏi có thể bị viết ngược
-    
-    Args:
-        question: Câu hỏi gốc
-        ai_brain: AI brain instance để hỏi lại (optional)
+    if is_likely_reversed(question):
+        reversed_question = reverse_text(question)
         
-    Returns:
-        Dict chứa thông tin phân tích text
-    """
-    analysis = {
-        "original_text": question,
-        "reversed_text": reverse_text(question),
-        "likely_reversed": is_likely_reversed(question),
-        "should_reverse": False,
-        "processed_text": question
-    }
+        return {
+            "should_reverse": True,
+            "original_text": question,
+            "processed_text": reversed_question,
+            "confidence": "high",
+            "reasoning": "Text appears to be written backwards"
+        }
     
-    # Nếu có dấu hiệu bị viết ngược
-    if analysis["likely_reversed"]:
-        print(f"🔄 Detected likely reversed text: {question[:50]}...")
-        
-        # Nếu có AI brain, hỏi AI quyết định
-        if ai_brain:
-            check_prompt = f"""
-            Original: {question}
-            Reversed: {analysis["reversed_text"]}
-            
-            Which version makes more sense as a question? Answer "original" or "reversed" only.
-            """
-            try:
-                ai_response = ai_brain.think(check_prompt).strip().lower()
-                analysis["should_reverse"] = "reversed" in ai_response
-                analysis["ai_decision"] = ai_response
-            except:
-                # Fallback nếu AI không hoạt động
-                analysis["should_reverse"] = True
-                analysis["ai_decision"] = "fallback_reverse"
-        else:
-            # Không có AI, AI sẽ quyết định sau
-            analysis["should_reverse"] = None  # Để AI quyết định
-        
-        if analysis["should_reverse"]:
-            analysis["processed_text"] = analysis["reversed_text"]
-            print(f"🔄 Reversed to: {analysis['reversed_text'][:50]}...")
-    
-    return analysis
-
-# Test function
-if __name__ == "__main__":
-    # Test case từ đề bài
-    test_question = ".rewsna eht sa \"tfel\" drow eht fo etisoppo eht etirw ,ecnetnes siht dnatsrednu uoy fI"
-    
-    print("Original:", test_question)
-    print("Is likely reversed:", is_likely_reversed(test_question))
-    
-    analysis = reverse_text_if_needed(test_question)
-    print("Analysis:", analysis) 
+    return {
+        "should_reverse": False,
+        "original_text": question,
+        "processed_text": question,
+        "confidence": "high",
+        "reasoning": "Text appears normal"
+    } 

utils/tool_orchestrator.py

@@ -6,7 +6,7 @@ import time
 from typing import Dict, Any, List, Optional, Callable
 from .state_manager import AgentState, ToolResult, get_agent_state, analyze_question_type, detect_urls_in_question
 from .text_tool import reverse_text_if_needed
-from .youtube_tool import get_youtube_content
+from .youtube_tool import YouTubeTool
 from .image_tool import ocr_image_with_nanonets, fallback_ocr_image
 from .audio_tool import transcribe_audio_groq, fallback_audio_info
 from .wiki_tool import search_wikipedia_from_question
@@ -198,8 +198,9 @@ class ToolOrchestrator:
     def _run_youtube_tool(self, question: str, task_id: str = "") -> Dict[str, Any]:
         """Run YouTube content extraction"""
         try:
-            result = get_youtube_content(question)
-            return result
+            youtube_tool = YouTubeTool()
+            result = youtube_tool.process_youtube(question)
+            return result.get("data", {}) if result.get("success") else {"success": False, "error": result.get("error", "Unknown error")}
         except Exception as e:
             return {"success": False, "error": str(e)}
     

utils/youtube_tool.py

@@ -1,161 +1,189 @@
 """
-YouTube Tool - Auto detect YouTube URLs and extract metadata + transcript/thumbnail
+YOUTUBE PROCESSING TOOL
+Enhanced with cookies support for bot detection bypass
 """
 
+import os
 import re
-import requests
-from typing import Dict, Any, Optional
+import json
+import tempfile
+from typing import Dict, Any, Optional, List
+from urllib.parse import urlparse, parse_qs
 
-def extract_youtube_url(text: str) -> Optional[str]:
-    """
-    Tự động regex tìm link YouTube trong câu hỏi task
-    """
-    # Regex lấy link YouTube
-    pattern = r'(https?://(?:www\.)?(?:youtube\.com/watch\?v=|youtu\.be/)[\w\-]+)'
-    match = re.search(pattern, text)
-    if match:
-        return match.group(1)
-    return None
+try:
+    from pytube import YouTube
+    from youtube_transcript_api import YouTubeTranscriptApi
+    import yt_dlp
+except ImportError as e:
+    print(f"⚠️ YouTube dependencies missing: {e}")
 
-def extract_youtube_id(url: str) -> Optional[str]:
-    """
-    Lấy video_id từ YouTube URL
-    """
-    # Hỗ trợ cả dạng youtube.com/watch?v=... và youtu.be/...
-    pattern = r'(?:v=|\/)([0-9A-Za-z_-]{11})'
-    match = re.search(pattern, url)
-    if match:
-        return match.group(1)
-    return None
+from .state_manager import get_agent_state
 
-def get_youtube_thumbnail_url(video_url: str) -> Optional[str]:
-    """
-    Lấy link thumbnail từ YouTube URL
-    """
-    video_id = extract_youtube_id(video_url)
-    if not video_id:
-        return None
-    # Link thumbnail chuẩn của YouTube
-    return f"https://img.youtube.com/vi/{video_id}/hqdefault.jpg"
-
-def get_txt_content_from_url(url: str) -> str:
-    """
-    Lấy nội dung file .txt từ URL (dành cho transcript link)
-    """
-    try:
-        response = requests.get(url, timeout=30)
-        response.raise_for_status()
-        return response.text
-    except Exception as e:
-        return f"Error downloading text file: {str(e)}"
-
-def get_youtube_content(question: str) -> Dict[str, Any]:
-    """
-    Main function: Tự động detect YouTube URL trong câu hỏi và lấy metadata + transcript/thumbnail
-    
-    Args:
-        question: Câu hỏi task có thể chứa YouTube URL
-        
-    Returns:
-        Dict chứa metadata, transcript (nếu có), thumbnail URL
-    """
-    # Auto detect YouTube URL
-    youtube_url = extract_youtube_url(question)
-    
-    if not youtube_url:
-        return {
-            "has_youtube": False,
-            "error": "No YouTube URL found in question"
-        }
+class YouTubeTool:
+    def __init__(self):
+        # Path to cookies file in project root
+        self.cookies_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "cookies.txt")
+        print("🎬 YouTube Tool with cookies support initialized")
     
-    print(f"Found YouTube URL: {youtube_url}")
+    def process_youtube(self, youtube_input: str, **kwargs) -> Dict[str, Any]:
+        """
+        Process YouTube content with cookie authentication
+        """
+        try:
+            # Extract video ID from URL or use as-is
+            video_id = self._extract_video_id(youtube_input)
+            if not video_id:
+                return self._error_result("Invalid YouTube URL or video ID")
+            
+            print(f"🎬 Processing YouTube video: {video_id}")
+            
+            # Try multiple extraction methods
+            video_data = self._extract_with_cookies(video_id) or self._extract_with_pytube(video_id)
+            
+            if not video_data:
+                return self._error_result("Could not extract video data")
+            
+            # Update agent state
+            state = get_agent_state()
+            state.cached_data["youtube_analysis"] = video_data
+            
+            return {
+                "success": True,
+                "data": video_data,
+                "summary": f"YouTube video processed: {video_data.get('title', 'Unknown')[:50]}..."
+            }
+            
+        except Exception as e:
+            error_msg = f"YouTube processing failed: {str(e)}"
+            print(f"❌ {error_msg}")
+            return self._error_result(error_msg)
     
-    try:
-        # Sử dụng yt-dlp để lấy metadata an toàn với cookies
-        import yt_dlp
-        import os
-        
-        # Path to cookies file
-        cookies_path = "cookies.txt"
+    def _extract_video_id(self, url_or_id: str) -> Optional[str]:
+        """Extract video ID from YouTube URL or return if already ID"""
+        if len(url_or_id) == 11 and url_or_id.isalnum():
+            return url_or_id
         
-        ydl_opts = {
-            'writesubtitles': True,
-            'writeautomaticsub': True,
-            'subtitleslangs': ['en'],
-            'skip_download': True,
-            'quiet': True,
-            'no_warnings': True
-        }
+        # Extract from various YouTube URL formats
+        patterns = [
+            r'(?:youtube\.com/watch\?v=|youtu\.be/|youtube\.com/embed/)([a-zA-Z0-9_-]{11})',
+            r'youtube\.com/.*[?&]v=([a-zA-Z0-9_-]{11})',
+        ]
         
-        # Add cookies if file exists
-        if os.path.exists(cookies_path):
-            ydl_opts['cookiefile'] = cookies_path
-            print(f"🍪 Using cookies from {cookies_path}")
-        else:
-            print("⚠️ No cookies.txt found, trying without cookies")
+        for pattern in patterns:
+            match = re.search(pattern, url_or_id)
+            if match:
+                return match.group(1)
         
-        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
-            info = ydl.extract_info(youtube_url, download=False)
+        return None
+    
+    def _extract_with_cookies(self, video_id: str) -> Optional[Dict[str, Any]]:
+        """Extract using yt-dlp with cookies for better success rate"""
+        try:
+            ydl_opts = {
+                'quiet': True,
+                'no_warnings': True,
+                'extractaudio': False,
+                'extract_flat': False,
+            }
             
-            title = info.get('title', 'Unknown Title')
-            description = info.get('description', 'No description')
+            # Add cookies if file exists
+            if os.path.exists(self.cookies_path):
+                ydl_opts['cookiefile'] = self.cookies_path
+                print(f"🍪 Using cookies from: {self.cookies_path}")
             
-            # Lấy thumbnail
-            thumbnail_url = get_youtube_thumbnail_url(youtube_url)
+            url = f"https://www.youtube.com/watch?v={video_id}"
             
-            # Kiểm tra transcript
-            transcript_content = None
-            if 'subtitles' in info and info['subtitles']:
-                # Có subtitle/transcript
-                for lang in ['en', 'en-US', 'en-GB']:
-                    if lang in info['subtitles']:
-                        subtitle_info = info['subtitles'][lang]
-                        if subtitle_info and len(subtitle_info) > 0:
-                            transcript_url = subtitle_info[0].get('url')
-                            if transcript_url:
-                                transcript_content = get_txt_content_from_url(transcript_url)
-                                break
+            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                info = ydl.extract_info(url, download=False)
+                
+                # Extract transcript using youtube-transcript-api
+                transcript = self._get_transcript(video_id)
+                
+                return {
+                    "video_id": video_id,
+                    "title": info.get('title', ''),
+                    "description": info.get('description', ''),
+                    "channel": info.get('uploader', ''),
+                    "duration": info.get('duration', 0),
+                    "view_count": info.get('view_count', 0),
+                    "transcript": transcript,
+                    "thumbnail_url": info.get('thumbnail', ''),
+                    "upload_date": info.get('upload_date', ''),
+                    "url": url,
+                    "extraction_method": "yt-dlp_with_cookies"
+                }
+        
+        except Exception as e:
+            print(f"⚠️ yt-dlp extraction failed: {str(e)}")
+            return None
+    
+    def _extract_with_pytube(self, video_id: str) -> Optional[Dict[str, Any]]:
+        """Fallback extraction using pytube"""
+        try:
+            url = f"https://www.youtube.com/watch?v={video_id}"
+            yt = YouTube(url)
             
-            # Kiểm tra automatic_captions nếu không có subtitles
-            if not transcript_content and 'automatic_captions' in info and info['automatic_captions']:
-                for lang in ['en', 'en-US', 'en-GB']:
-                    if lang in info['automatic_captions']:
-                        caption_info = info['automatic_captions'][lang]
-                        if caption_info and len(caption_info) > 0:
-                            # Tìm format .vtt hoặc .txt
-                            for caption in caption_info:
-                                if caption.get('ext') in ['vtt', 'txt']:
-                                    transcript_url = caption.get('url')
-                                    if transcript_url:
-                                        transcript_content = get_txt_content_from_url(transcript_url)
-                                        break
-                            if transcript_content:
-                                break
+            transcript = self._get_transcript(video_id)
             
             return {
-                "has_youtube": True,
-                "title": title,
-                "description": description[:1000],  # Giới hạn description
-                "transcript": transcript_content,
-                "thumbnail_url": thumbnail_url,
-                "video_url": youtube_url
+                "video_id": video_id,
+                "title": yt.title or '',
+                "description": yt.description or '',
+                "channel": yt.author or '',
+                "duration": yt.length or 0,
+                "view_count": yt.views or 0,
+                "transcript": transcript,
+                "thumbnail_url": yt.thumbnail_url or '',
+                "upload_date": str(yt.publish_date) if yt.publish_date else '',
+                "url": url,
+                "extraction_method": "pytube_fallback"
             }
             
-    except Exception as e:
-        # Fallback: Ít nhất trả về thumbnail
-        thumbnail_url = get_youtube_thumbnail_url(youtube_url)
+        except Exception as e:
+            print(f"⚠️ PyTube extraction failed: {str(e)}")
+            return None
+    
+    def _get_transcript(self, video_id: str) -> str:
+        """Get video transcript using youtube-transcript-api"""
+        try:
+            # Try to get transcript in multiple languages
+            languages = ['en', 'en-US', 'auto', 'vi']
+            
+            for lang in languages:
+                try:
+                    transcript_list = YouTubeTranscriptApi.get_transcript(video_id, languages=[lang])
+                    transcript_text = ' '.join([entry['text'] for entry in transcript_list])
+                    if transcript_text.strip():
+                        return transcript_text
+                except:
+                    continue
+            
+            # If no manual transcript, try auto-generated
+            try:
+                transcript_list = YouTubeTranscriptApi.get_transcript(video_id)
+                return ' '.join([entry['text'] for entry in transcript_list])
+            except:
+                return "No transcript available"
+                
+        except Exception as e:
+            print(f"⚠️ Transcript extraction failed: {str(e)}")
+            return "Transcript extraction failed"
+    
+    def is_youtube_url(self, text: str) -> bool:
+        """Check if text contains YouTube URL"""
+        youtube_patterns = [
+            r'youtube\.com/watch\?v=',
+            r'youtu\.be/',
+            r'youtube\.com/embed/',
+            r'youtube\.com/.*[?&]v='
+        ]
+        return any(re.search(pattern, text, re.IGNORECASE) for pattern in youtube_patterns)
+    
+    def _error_result(self, error_msg: str) -> Dict[str, Any]:
+        """Standard error result format"""
         return {
-            "has_youtube": True,
-            "title": "Could not fetch title",
-            "description": "Could not fetch description",
-            "transcript": None,
-            "thumbnail_url": thumbnail_url,
-            "video_url": youtube_url,
-            "error": f"YouTube extraction error: {str(e)}"
-        }
-
-# Test function
-if __name__ == "__main__":
-    test_question = "What is this video about? https://www.youtube.com/watch?v=dQw4w9WgXcQ"
-    result = get_youtube_content(test_question)
-    print("Result:", result) 
+            "success": False,
+            "error": error_msg,
+            "data": None,
+            "summary": f"YouTube processing failed: {error_msg}"
+        }