如何使用LlamaIndex和网页抓取构建生产就绪的RAG（2025指南）

python -m venv venv
source venv/bin/activate   # On Windows: venv\Scripts\activate

pip install llama-index llama-index-vector-stores-chroma llama-index-embeddings-openai \
llama-index-embeddings-huggingface llama-index-llms-openai \
fastapi uvicorn chromadb sentence-transformers \
python-dotenv requests beautifulsoup4

ollama pull llama3.2:3b

# API
RAG_API_KEY=your-secret-api-key
HOST=0.0.0.0
PORT=8000

# Vector store
CHROMA_PERSIST_DIR=./chroma_db

# Decodo API (used by scraper.py)
DECODO_USERNAME=your-decodo-username
DECODO_PASSWORD=your-decodo-password

# LLM (optional, set ENABLE_LLM=true to use)
ENABLE_LLM=false
OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxx
OPENAI_MODEL=gpt-3.5-turbo

# Web Framework
fastapi==0.115.*
uvicorn[standard]==0.30.*

# Vector Database
chromadb==0.5.*

# Embeddings
sentence-transformers==2.7.*

# Web Scraping
requests==2.32.*
beautifulsoup4==4.12.*

# Environment Configuration
python-dotenv==1.0.*

# Data Validation (FastAPI dependency)
pydantic==2.5.*

# Optional for full RAG
openai==1.40.*

import os
import requests
from bs4 import BeautifulSoup
import logging
from typing import Optional
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

logger = logging.getLogger(__name__)

class DecodoScraper:
    """Web scraper using Decodo's Web Scraping API"""  
    def __init__(self):
        self.username = os.getenv("DECODO_USERNAME")
        self.password = os.getenv("DECODO_PASSWORD")
        self.base_url = "https://scraper-api.decodo.com/v2/scrape"
        self.session = requests.Session()       

        if self.username and self.password:
            self.session.auth = (self.username, self.password)
            self.session.headers.update({'Content-Type': 'application/json'})
        else:
            logger.warning("Decodo credentials not found in environment variables")  

    def scrape_url(self, url: str) -> str:
        """Scrape a webpage using Decodo's Web Scraping API"""
        try:
            logger.info(f"Scraping URL: {url}")           

            task_params = {
                'url': url,
                'parse': True  # Enable automatic parsing
            }           

            response = self.session.post(
                self.base_url,
                json=task_params,
                timeout=30
            )

           logger.info(f"Response status: {response.status_code}")
           
            if response.status_code == 200:
                data = response.json()
                logger.info(f"Response data keys: {list(data.keys())}")
                logger.info(f"Full response structure: {data}")              

                # Check if we have results
                if 'results' in data and len(data['results']) > 0:
                    result = data['results'][0]
                    logger.info(f"Result keys: {list(result.keys())}")                   

                    # Try to extract the actual HTML content
                    if 'content' in result:
                        content = result['content']
                        logger.info(f"Content keys: {list(content.keys()) if isinstance(content, dict) else 'Not a dict'}") 

                        # If we have HTML content directly
                        if 'html' in content:
                            soup = BeautifulSoup(content['html'], 'html.parser')
                            text = soup.get_text()
                            return text
                        # If we have text content directly
                        elif 'text' in content:
                            return content['text']
                        # If we have description
                        elif 'description' in content:
                            return content['description']
                        # If content is a string
                        elif isinstance(content, str):
                            return content
                        # If we have parsed content
                        elif 'results' in content:
                            return str(content['results'])
                        else:
                            return str(content)
                    # If result has HTML directly
                    elif 'html' in result:
                        soup = BeautifulSoup(result['html'], 'html.parser')
                        text = soup.get_text()
                        return text
                    # If result has text directly
                    elif 'text' in result:
                        return result['text']
                    # If result has description
                    elif 'description' in result:
                        return result['description']
                    else:
                        # Return the entire result as string for debugging
                        logger.warning(f"Unexpected result structure: {result}")
                        return str(result)
                else:
                    raise Exception(f"No results found in response: {data}")
            else:
                raise Exception(f"HTTP error {response.status_code}: {response.text}")      
        except requests.exceptions.RequestException as e:
            logger.error(f"Request failed: {e}")
            raise Exception(f"Request failed: {e}")
        except Exception as e:
            logger.error(f"Scraping failed: {e}")
            raise Exception(f"Scraping failed: {e}")   

    def is_ready(self) -> bool:
        """Check if scraper is ready (has credentials)"""
        return bool(self.username and self.password)   

    def test_connection(self) -> bool:
        """Test the connection to Decodo API"""
        try:
            # Test with a simple URL
            test_url = "https://httpbin.org/html"
            content = self.scrape_url(test_url)
            return len(content) > 0
        except Exception as e:
            logger.error(f"Connection test failed: {e}")
            return False


# Global scraper instance
_scraper_instance = None


def get_scraper() -> DecodoScraper:
    """Get the global scraper instance"""
    global _scraper_instance
    if _scraper_instance is None:
        _scraper_instance = DecodoScraper()
    return _scraper_instance


# Test function
if __name__ == "__main__":
    import asyncio
    import logging   

    # Set up logging to see debug information
    logging.basicConfig(level=logging.INFO)  

    async def test_scraper():
        scraper = get_scraper()

        if not scraper.is_ready():
            print("Scraper not ready - missing credentials")
            print("Make sure to set DECODO_USERNAME and DECODO_PASSWORD in .env file")
            return

        print("Scraper ready")

        try:
            # Test with a simple URL
            test_url = "https://httpbin.org/html"
            print(f"Testing with URL: {test_url}")           

            content = scraper.scrape_url(test_url)
            print(f"Successfully scraped {len(content)} characters")
            print(f"Content preview: {content[:200]}...") 

            # Also test with a more complex URL
            print("\n" + "="*50)
            test_url2 = "https://example.com"
            print(f"Testing with URL: {test_url2}")           
            content2 = scraper.scrape_url(test_url2)
            print(f"Successfully scraped {len(content2)} characters")

            print(f"Content preview: {content2[:200]}...")

        except Exception as e:
            print(f"Scraping failed: {e}")   

    asyncio.run(test_scraper())

# rag_system.py
import os
import asyncio
import logging
from typing import List, Dict, Any, Optional
from datetime import datetime, timezone
import uuid

# Core dependencies
from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, HttpUrl, Field
import uvicorn
from dotenv import load_dotenv

load_dotenv()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# LlamaIndex imports
try:
    from llama_index.core import Document, VectorStoreIndex, StorageContext
    from llama_index.core import Settings as LISettings
    from llama_index.core.node_parser import SentenceSplitter
    from llama_index.embeddings.huggingface import HuggingFaceEmbedding
    from llama_index.embeddings.openai import OpenAIEmbedding
    from llama_index.vector_stores.chroma import ChromaVectorStore
    from llama_index.llms.openai import OpenAI as LIOpenAI
    import chromadb
    from chromadb.config import Settings as ChromaSettings
    LLAMAINDEX_AVAILABLE = True
except ImportError as e:
    LLAMAINDEX_AVAILABLE = False
    logger.warning(f"LlamaIndex not available: {e}")

class Settings:
    """Application settings from environment variables"""   
    # API Configuration
    API_KEY = os.getenv("RAG_API_KEY", "your-secret-api-key")
    HOST = os.getenv("HOST", "0.0.0.0")
    PORT = int(os.getenv("PORT", 8000))   

    # Database Configuration
    CHROMA_PERSIST_DIR = os.getenv("CHROMA_PERSIST_DIR", "./chroma_db")    

    # LLM Configuration
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
    OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-3.5-turbo")
    ENABLE_LLM = os.getenv("ENABLE_LLM", "true").lower() == "true"    

    # Embedding Configuration
    EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "openai")
    HUGGINGFACE_MODEL = os.getenv("HUGGINGFACE_MODEL", "BAAI/bge-small-en-v1.5")    

    # Chunking Configuration
    CHUNK_SIZE = int(os.getenv("CHUNK_SIZE", 1024))
    CHUNK_OVERLAP = int(os.getenv("CHUNK_OVERLAP", 200))

class LlamaIndexVectorDB:
    """Vector DB via LlamaIndex on top of Chroma"""

    def __init__(self, persist_directory: str):
        self.persist_directory = persist_directory
        self.client = None
        self.is_initialized = False

    async def initialize(self, settings: Settings):
        if not LLAMAINDEX_AVAILABLE:
            raise Exception("LlamaIndex not available. Install with: pip install llama-index")

        os.makedirs(self.persist_directory, exist_ok=True)
        self.client = chromadb.PersistentClient(
            path=self.persist_directory,
            settings=ChromaSettings(anonymized_telemetry=False, allow_reset=True),
        )        

        # Configure embedding model through LlamaIndex
        if settings.EMBEDDING_MODEL.lower() == "openai" and settings.OPENAI_API_KEY:
            LISettings.embed_model = OpenAIEmbedding(
                model="text-embedding-3-small", 
                api_key=settings.OPENAI_API_KEY
           )
        else:
            LISettings.embed_model = HuggingFaceEmbedding(
                model_name=settings.HUGGINGFACE_MODEL
            )

        # Configure chunking through LlamaIndex
        LISettings.node_parser = SentenceSplitter(
            chunk_size=settings.CHUNK_SIZE, 
            chunk_overlap=settings.CHUNK_OVERLAP
        )

        self.is_initialized = True
        logger.info("LlamaIndexVectorDB initialized")

    async def add_document(self, collection_name: str, content: str, metadata: Dict[str, Any]):
        collection = self.client.get_or_create_collection(collection_name)
        vector_store = ChromaVectorStore(chroma_collection=collection)
        storage_context = StorageContext.from_defaults(vector_store=vector_store)

        # Let LlamaIndex handle chunking automatically
        docs = [Document(text=content, metadata=metadata)]
        VectorStoreIndex.from_documents(docs, storage_context=storage_context)
        logger.info(f"Indexed document into '{collection_name}'")

    async def query(self, collection_name: str, query_text: str, n_results: int = 5) -> Dict[str, Any]:
        collection = self.client.get_or_create_collection(collection_name)
        vector_store = ChromaVectorStore(chroma_collection=collection)
        index = VectorStoreIndex.from_vector_store(vector_store)
        qe = index.as_query_engine(similarity_top_k=n_results)
        resp = await asyncio.to_thread(qe.query, query_text)

        docs, metas, dists = [], [], []
        for sn in getattr(resp, "source_nodes", []):
            docs.append(sn.get_text())
            metas.append(sn.node.metadata or {})
            # sn.score is similarity (higher is better)
            relevance_score = max(0.0, min(1.0, sn.score or 0.0))
            dists.append(relevance_score) 

        return {"documents": docs, "metadatas": metas, "distances": dists}

class LlamaIndexRAGSystem:
    """Production-ready RAG system using LlamaIndex"""
    def __init__(self, settings: Settings):
        self.settings = settings
        self.vector_db = LlamaIndexVectorDB(settings.CHROMA_PERSIST_DIR)
        self.is_initialized = False

    async def initialize(self):
        """Initialize LlamaIndex components"""
        try:
            await self.vector_db.initialize(self.settings)

            # Configure LlamaIndex LLM
            if self.settings.ENABLE_LLM and self.settings.OPENAI_API_KEY:
                LISettings.llm = LIOpenAI(
                    model=self.settings.OPENAI_MODEL,
                    api_key=self.settings.OPENAI_API_KEY
                )
                logger.info(f"LlamaIndex LLM configured: {self.settings.OPENAI_MODEL}")

            self.is_initialized = True
            logger.info("LlamaIndex RAG system initialized successfully")
            

        except Exception as e:
            logger.error(f"Failed to initialize LlamaIndex RAG system: {e}")
            raise

    async def add_document(self, content: str, metadata: Dict[str, Any], collection_name: str = None):
        """Add a document to LlamaIndex"""
        try:
            if not self.is_initialized:
                raise Exception("RAG system not initialized")

            await self.vector_db.add_document(
                collection_name or "default",
                content,
                metadata
            )

            logger.info(f"Added document to LlamaIndex with {len(content)} characters")

        except Exception as e:
            logger.error(f"Failed to add document: {e}")
            raise

   async def query(self, query_text: str, collection_name: str = None) -> Dict[str, Any]:
        """Query the LlamaIndex RAG system"""
        try:
            if not self.is_initialized:
                raise Exception("RAG system not initialized")

            # Query with LlamaIndex
            results = await self.vector_db.query(
                collection_name or "default",
                query_text,
                n_results=5
            )            if results["documents"]:
                context = "\n\n".join(results["documents"])            

                # Try to use LlamaIndex LLM for response generation
                if hasattr(LISettings, 'llm') and LISettings.llm:
                    try:
                        prompt = f"""Based on the following context, please answer the question.
Context:
{context}

Question: {query_text}

Answer:"""
                        response = await asyncio.to_thread(LISettings.llm.complete, prompt)
                        answer = str(response)
                    except Exception as e:
                        logger.warning(f"LlamaIndex LLM generation failed, using simple assembly: {e}")
                        answer = f"Based on the context: {context[:500]}..."
                else:
                    # Simple assembly (R+A mode)
                    answer = f"Based on the context: {context[:500]}..."

                sources = []
                for i, (doc, metadata) in enumerate(zip(results["documents"], results["metadatas"])):
                    sources.append({
                        "content": doc[:200] + "..." if len(doc) > 200 else doc,
                        "metadata": metadata,
                        "relevance_score": results["distances"][i] if i < len(results["distances"]) else 0
                    })

                mode = "LlamaIndex RAG" if getattr(LISettings, "llm", None) else "LlamaIndex R+A"
                return {
                    "answer": answer,
                    "sources": sources,
                    "mode": mode
                }

            else:
                mode = "LlamaIndex RAG" if getattr(LISettings, "llm", None) else "LlamaIndex R+A"
                return {
                    "answer": "No relevant information found.",
                    "sources": [],
                    "mode": mode
                }

        except Exception as e:
            logger.error(f"Query failed: {e}")
            raise   

    async def scrape_and_store(self, url: str, collection_name: str = None, max_content_length: int = 1_000_000) -> str:
        """Scrape a URL and store the content in LlamaIndex"""
        try:
            from scraper import get_scraper   
            scraper = get_scraper()
            content = scraper.scrape_url(url)

            # Guard against empty pages
            if not content or len(content.strip()) < 50:
                raise Exception(f"Scraped content too short or empty: {len(content)} characters")

            # Size limit check
            if len(content) > max_content_length:
                logger.warning(f"Content truncated from {len(content)} to {max_content_length} characters")
                content = content[:max_content_length]

            # Create metadata
            metadata = {
                "url": url,
                "scraped_at": datetime.now(timezone.utc).isoformat(),
                "task_id": uuid.uuid4().hex,
                "content_length": len(content)
            }

            await self.add_document(content, metadata, collection_name)

            logger.info(f"Successfully scraped and stored content from {url}")
            return collection_name or "default"

        except Exception as e:
            logger.error(f"Failed to scrape and store: {e}")
            raise