Stop Building Dumb Chatbots: How to Create Smart LangChain Agents That Actually Think

# This is the mental model that clicked for me
Agent Flow:
1. User gives goal: "Find competitor pricing for our top 3 products"
2. Agent thinks: "I need to search websites and compare data"
3. Agent chooses tools: web_search, data_analyzer, report_generator
4. Agent executes: searches → analyzes → summarizes → reports
5. Agent evaluates: "Did I solve the user's problem completely?"

# Create isolated environment (trust me on this)
python -m venv langchain_agents
source langchain_agents/bin/activate  # macOS/Linux
# langchain_agents\Scripts\activate    # Windows

# Install exact versions I tested
pip install langchain==0.2.11
pip install langchain-openai==0.1.19
pip install langchain-community==0.2.10
pip install python-dotenv==1.0.0
pip install requests==2.31.0

# .env file (create this in your project root)
OPENAI_API_KEY=your_actual_api_key_here
SERP_API_KEY=your_serp_api_key_here  # Optional for web search

# config.py - Your agent settings
import os
from dotenv import load_dotenv

load_dotenv()

# Model configuration I actually use in production
AGENT_MODEL = "gpt-4"  # Worth the extra cost for reasoning
TEMPERATURE = 0.1      # Low for consistent logic
MAX_TOKENS = 2000      # Enough for complex thoughts
TIMEOUT = 30           # Prevents infinite loops

# API Keys
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
SERP_API_KEY = os.getenv("SERP_API_KEY")

# research_agent.py
from langchain.agents import initialize_agent, AgentType
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain.memory import ConversationBufferWindowMemory
import requests
from bs4 import BeautifulSoup
import json

class ResearchAgent:
    def __init__(self, api_key, model="gpt-4", temperature=0.1):
        """Initialize agent with specific settings I use in production"""
        self.llm = ChatOpenAI(
            api_key=api_key,
            model=model,
            temperature=temperature,
            max_tokens=2000
        )
        
        # Memory that actually works for multi-step tasks
        self.memory = ConversationBufferWindowMemory(
            memory_key="chat_history",
            k=10,  # Remember last 10 interactions
            return_messages=True
        )
        
        self.tools = self._create_tools()
        self.agent = self._initialize_agent()
    
    def _create_tools(self):
        """Tools I actually use for real research tasks"""
        return [
            Tool(
                name="web_scraper",
                description="Scrape content from any website URL. Input: URL string",
                func=self._scrape_website
            ),
            Tool(
                name="search_analyzer", 
                description="Analyze and summarize large amounts of text. Input: text to analyze",
                func=self._analyze_content
            ),
            Tool(
                name="data_formatter",
                description="Format data into structured reports. Input: raw data",
                func=self._format_data
            )
        ]
    
    def _scrape_website(self, url):
        """Web scraper that handles real-world websites"""
        try:
            headers = {
                'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'
            }
            response = requests.get(url, headers=headers, timeout=10)
            soup = BeautifulSoup(response.content, 'html.parser')
            
            # Extract meaningful content (not just raw HTML)
            text = soup.get_text()
            # Clean up whitespace
            clean_text = ' '.join(text.split())
            
            return clean_text[:5000]  # Limit to prevent token overload
        except Exception as e:
            return f"Error scraping {url}: {str(e)}"
    
    def _analyze_content(self, text):
        """Content analysis that finds key insights"""
        if len(text) < 100:
            return "Text too short to analyze meaningfully"
        
        # Simple but effective analysis
        word_count = len(text.split())
        sentences = text.split('.')
        avg_sentence_length = word_count / len(sentences) if sentences else 0
        
        analysis = f"""
        Content Analysis:
        - Word count: {word_count}
        - Sentences: {len(sentences)}
        - Avg sentence length: {avg_sentence_length:.1f} words
        - Key themes: {self._extract_themes(text)}
        """
        return analysis
    
    def _extract_themes(self, text):
        """Extract key themes from content"""
        # Simple keyword extraction (you could use more advanced NLP)
        words = text.lower().split()
        word_freq = {}
        
        for word in words:
            if len(word) > 4:  # Only meaningful words
                word_freq[word] = word_freq.get(word, 0) + 1
        
        # Return top 5 most frequent meaningful words
        top_words = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)[:5]
        return [word for word, freq in top_words]
    
    def _format_data(self, data):
        """Format data into readable reports"""
        try:
            # Try to structure the data intelligently
            formatted = f"""
            RESEARCH REPORT
            ==============
            
            Summary: {data[:200]}...
            
            Key Points:
            {self._extract_key_points(data)}
            
            Generated: {self._get_timestamp()}
            """
            return formatted
        except Exception as e:
            return f"Error formatting data: {str(e)}"
    
    def _extract_key_points(self, data):
        """Extract bullet points from data"""
        sentences = data.split('.')[:5]  # Top 5 sentences
        points = []
        for i, sentence in enumerate(sentences, 1):
            if sentence.strip():
                points.append(f"• {sentence.strip()}")
        return '\n'.join(points)
    
    def _get_timestamp(self):
        """Get current timestamp for reports"""
        from datetime import datetime
        return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    
    def _initialize_agent(self):
        """Initialize the agent with proper configuration"""
        return initialize_agent(
            tools=self.tools,
            llm=self.llm,
            agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,
            memory=self.memory,
            verbose=True,  # Shows thinking process
            handle_parsing_errors=True,  # Prevents crashes
            max_iterations=5  # Prevents infinite loops
        )
    
    def research(self, query):
        """Main method - give it any research task"""
        try:
            result = self.agent.run(query)
            return result
        except Exception as e:
            return f"Research failed: {str(e)}"

# test_agent.py
from research_agent import ResearchAgent
from config import OPENAI_API_KEY

def main():
    """Test the agent with increasingly complex tasks"""
    agent = ResearchAgent(OPENAI_API_KEY)
    
    print("🤖 Research Agent Ready")
    print("-" * 40)
    
    # Test 1: Simple web research
    print("\n📊 TEST 1: Company Research")
    result = agent.research("""
    Research OpenAI's latest pricing changes. 
    Find their current pricing page and summarize the key points.
    """)
    print(f"Result: {result}")
    
    # Test 2: Comparative analysis  
    print("\n📊 TEST 2: Competitive Analysis")
    result = agent.research("""
    Compare the pricing of OpenAI vs Anthropic for API usage.
    Create a structured comparison report.
    """)
    print(f"Result: {result}")
    
    # Test 3: Interactive research
    print("\n📊 TEST 3: Interactive Session")
    while True:
        query = input("\nWhat would you like me to research? (or 'quit'): ")
        if query.lower() == 'quit':
            break
        
        result = agent.research(query)
        print(f"\nResearch Result:\n{result}")

if __name__ == "__main__":
    main()

# Terminal command
python test_agent.py

# advanced_agent.py
from langchain.agents import initialize_agent, AgentType
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain.memory import ConversationSummaryBufferMemory
from langchain.schema import SystemMessage
import json
import os
from datetime import datetime

class AdvancedResearchAgent:
    def __init__(self, api_key, workspace_dir="./agent_workspace"):
        """Production-ready agent with persistence and planning"""
        self.workspace_dir = workspace_dir
        self.ensure_workspace()
        
        self.llm = ChatOpenAI(
            api_key=api_key,
            model="gpt-4",
            temperature=0.1
        )
        
        # Advanced memory that summarizes old conversations
        self.memory = ConversationSummaryBufferMemory(
            llm=self.llm,
            memory_key="chat_history",
            max_token_limit=1000,
            return_messages=True
        )
        
        self.tools = self._create_advanced_tools()
        self.agent = self._initialize_advanced_agent()
    
    def ensure_workspace(self):
        """Create workspace for persistent data"""
        if not os.path.exists(self.workspace_dir):
            os.makedirs(self.workspace_dir)
        
        # Create subdirectories for organization
        for subdir in ['reports', 'data', 'cache']:
            path = os.path.join(self.workspace_dir, subdir)
            if not os.path.exists(path):
                os.makedirs(path)
    
    def _create_advanced_tools(self):
        """Production tools with error handling and persistence"""
        return [
            Tool(
                name="web_researcher",
                description="Research any topic by scraping multiple websites. Input: research topic",
                func=self._multi_site_research
            ),
            Tool(
                name="report_generator", 
                description="Generate comprehensive reports with Data Analysis. Input: research data",
                func=self._generate_report
            ),
            Tool(
                name="task_planner",
                description="Break complex tasks into manageable steps. Input: complex task description", 
                func=self._plan_task
            ),
            Tool(
                name="data_persistence",
                description="Save research data for future reference. Input: data to save",
                func=self._save_research_data
            ),
            Tool(
                name="knowledge_retrieval",
                description="Retrieve previously saved research data. Input: search query",
                func=self._retrieve_knowledge
            )
        ]
    
    def _multi_site_research(self, topic):
        """Research across multiple sources intelligently"""
        try:
            # Plan the research strategy
            plan = self._create_research_plan(topic)
            results = []
            
            for source_type in plan['sources']:
                if source_type == 'official_sites':
                    # Research official websites
                    data = self._research_official_sources(topic)
                    results.append(data)
                elif source_type == 'news_sites':
                    # Research news sources
                    data = self._research_news_sources(topic) 
                    results.append(data)
                elif source_type == 'technical_docs':
                    # Research documentation
                    data = self._research_technical_sources(topic)
                    results.append(data)
            
            # Combine and analyze all results
            combined_research = self._combine_research_results(results)
            return combined_research
            
        except Exception as e:
            return f"Multi-site research failed: {str(e)}"
    
    def _create_research_plan(self, topic):
        """Create intelligent research strategy"""
        # Simple but effective planning logic
        plan = {
            'sources': ['official_sites', 'news_sites'],
            'priority': 'comprehensive',
            'time_limit': '5_minutes'
        }
        
        # Adjust plan based on topic
        if 'technical' in topic.lower() or 'api' in topic.lower():
            plan['sources'].append('technical_docs')
        
        if 'pricing' in topic.lower() or 'cost' in topic.lower():
            plan['priority'] = 'official_sources_first'
            
        return plan
    
    def _research_official_sources(self, topic):
        """Research official company/organization websites"""
        # This would contain your web scraping logic
        # Simplified for example
        return f"Official source data for: {topic}"
    
    def _research_news_sources(self, topic):
        """Research news and blog sources"""
        return f"News source data for: {topic}"
    
    def _research_technical_sources(self, topic):
        """Research technical documentation"""
        return f"Technical documentation for: {topic}"
    
    def _combine_research_results(self, results):
        """Intelligently combine research from multiple sources"""
        combined = {
            'summary': "Research completed across multiple sources",
            'sources_checked': len(results),
            'key_findings': [],
            'raw_data': results
        }
        
        # Extract key findings from each source
        for i, result in enumerate(results):
            combined['key_findings'].append(f"Source {i+1}: {result[:100]}...")
        
        return json.dumps(combined, indent=2)
    
    def _generate_report(self, data):
        """Generate comprehensive, formatted reports"""
        try:
            report_data = json.loads(data) if isinstance(data, str) else data
            
            report = f"""
# RESEARCH REPORT
**Generated:** {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
**Sources Analyzed:** {report_data.get('sources_checked', 'Multiple')}

## Executive Summary
{report_data.get('summary', 'Research completed successfully')}

## Key Findings
"""
            
            for i, finding in enumerate(report_data.get('key_findings', []), 1):
                report += f"\n{i}. {finding}"
            
            report += f"""

## Detailed Analysis
{report_data.get('analysis', 'Detailed analysis available in raw data')}

## Recommendations
Based on the research findings, key recommendations include:
- Further investigation of identified trends
- Regular monitoring of identified sources
- Follow-up research on emerging topics

---
*Report generated by Advanced Research Agent*
"""
            
            # Save report to workspace
            report_path = self._save_report(report)
            return f"Report generated and saved to: {report_path}\n\n{report}"
            
        except Exception as e:
            return f"Report generation failed: {str(e)}"
    
    def _plan_task(self, task_description):
        """Break complex tasks into actionable steps"""
        # This would use more sophisticated planning
        # Simplified for example
        steps = [
            f"1. Analyze task: '{task_description[:50]}...'",
            "2. Identify required information sources",
            "3. Gather data from identified sources", 
            "4. Analyze and synthesize information",
            "5. Generate comprehensive report",
            "6. Validate findings and recommendations"
        ]
        
        plan = {
            'task': task_description,
            'steps': steps,
            'estimated_time': '15-30 minutes',
            'complexity': 'moderate'
        }
        
        return json.dumps(plan, indent=2)
    
    def _save_research_data(self, data):
        """Save research data for future reference"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"research_data_{timestamp}.json"
        filepath = os.path.join(self.workspace_dir, 'data', filename)
        
        try:
            with open(filepath, 'w') as f:
                if isinstance(data, str):
                    f.write(data)
                else:
                    json.dump(data, f, indent=2)
            
            return f"Data saved successfully to: {filepath}"
        except Exception as e:
            return f"Failed to save data: {str(e)}"
    
    def _retrieve_knowledge(self, query):
        """Retrieve previously saved research"""
        data_dir = os.path.join(self.workspace_dir, 'data')
        
        if not os.path.exists(data_dir):
            return "No previous research data found"
        
        # Simple keyword search through saved files
        matching_files = []
        for filename in os.listdir(data_dir):
            if filename.endswith('.json'):
                try:
                    filepath = os.path.join(data_dir, filename)
                    with open(filepath, 'r') as f:
                        content = f.read()
                        if query.lower() in content.lower():
                            matching_files.append(filename)
                except Exception:
                    continue
        
        if matching_files:
            return f"Found {len(matching_files)} files matching '{query}': {matching_files}"
        else:
            return f"No saved research found matching: {query}"
    
    def _save_report(self, report_content):
        """Save generated reports"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"report_{timestamp}.md"
        filepath = os.path.join(self.workspace_dir, 'reports', filename)
        
        with open(filepath, 'w') as f:
            f.write(report_content)
        
        return filepath
    
    def _initialize_advanced_agent(self):
        """Initialize agent with advanced configuration"""
        system_message = SystemMessage(content="""
You are an advanced research agent. Your capabilities include:

1. Multi-source research across websites and databases
2. Intelligent task planning and decomposition  
3. Data persistence and knowledge retrieval
4. Comprehensive report generation
5. Error handling and recovery

Always think step-by-step:
1. Understand the user's request completely
2. Plan your approach using available tools
3. Execute research systematically
4. Analyze and synthesize findings
5. Generate actionable insights
6. Save important data for future reference

Be thorough but efficient. Prioritize accuracy over speed.
""")
        
        return initialize_agent(
            tools=self.tools,
            llm=self.llm,
            agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,
            memory=self.memory,
            verbose=True,
            handle_parsing_errors=True,
            max_iterations=10,  # Allow more complex reasoning
            system_message=system_message
        )
    
    def research(self, query, save_results=True):
        """Enhanced research with optional result persistence"""
        try:
            result = self.agent.run(query)
            
            if save_results:
                # Save the research session
                session_data = {
                    'query': query,
                    'result': result,
                    'timestamp': datetime.now().isoformat()
                }
                self._save_research_data(session_data)
            
            return result
        except Exception as e:
            return f"Advanced research failed: {str(e)}"

# agent_cli.py
import argparse
from advanced_agent import AdvancedResearchAgent
from config import OPENAI_API_KEY

def main():
    parser = argparse.ArgumentParser(description='Research Agent CLI')
    parser.add_argument('--query', '-q', type=str, help='Research query')
    parser.add_argument('--interactive', '-i', action='store_true', help='Interactive mode')
    parser.add_argument('--workspace', '-w', type=str, default='./workspace', help='Workspace directory')
    
    args = parser.parse_args()
    
    # Initialize agent
    agent = AdvancedResearchAgent(OPENAI_API_KEY, args.workspace)
    
    if args.interactive:
        interactive_mode(agent)
    elif args.query:
        single_query_mode(agent, args.query)
    else:
        print("Use --query for single research or --interactive for session mode")

def interactive_mode(agent):
    """Interactive research session"""
    print("🤖 Advanced Research Agent Ready")
    print("Type 'help' for commands, 'quit' to exit")
    print("-" * 50)
    
    while True:
        try:
            query = input("\n🔍 Research Query: ").strip()
            
            if query.lower() == 'quit':
                print("👋 Research session ended")
                break
            elif query.lower() == 'help':
                show_help()
                continue
            elif query.lower().startswith('load '):
                # Load previous research
                search_term = query[5:]  # Remove 'load '
                result = agent.research(f"Retrieve knowledge about: {search_term}")
                print(f"\n📚 Previous Research:\n{result}")
                continue
            elif query == '':
                continue
                
            print(f"\n🔬 Researching: {query}")
            print("=" * 60)
            
            result = agent.research(query)
            print(f"\n✅ Research Complete:\n{result}")
            
        except KeyboardInterrupt:
            print("\n👋 Research interrupted by user")
            break
        except Exception as e:
            print(f"\n❌ Error: {str(e)}")

def single_query_mode(agent, query):
    """Single research query mode"""
    print(f"🔬 Researching: {query}")
    print("=" * 60)
    
    result = agent.research(query)
    print(f"\n✅ Research Result:\n{result}")

def show_help():
    """Show available commands"""
    help_text = """
Available Commands:
------------------
• [any text]     - Research the topic
• load [topic]   - Load previous research about topic  
• help           - Show this help message
• quit           - Exit the research session

Example Queries:
---------------
• "Compare pricing for OpenAI vs Anthropic APIs"
• "Research latest Python web frameworks in 2025"
• "Find competitor analysis for SaaS pricing models"
• "load pricing"  (loads previous pricing research)
"""
    print(help_text)

if __name__ == "__main__":
    main()

# Single research query
python agent_cli.py --query "Research LangChain alternatives in 2025"

# Interactive session
python agent_cli.py --interactive

# Custom workspace
python agent_cli.py --interactive --workspace /path/to/my/research