langxchange 0.5.0
AI Framework for fast integration of Private Data and LLM
Requires Python: >=3.8

# LangXChange Framework

<div align="center">

![LangXChange](https://img.shields.io/badge/LangXChange-Toolkit-blue)
![Python](https://img.shields.io/badge/Python-3.8%2B-blue)
![License](https://img.shields.io/badge/License-MIT-green)
![Version](https://img.shields.io/badge/Version-v0.5.0-orange)

**A comprehensive Python Framework for LLM operations, vector databases, RAG implementations, database integration, MCP Service Management and local model management**

[Installation](#installation) • [Quick Start](#quick-start) • [Documentation](#modules) • [Examples](#examples) • [RAG Tutorial](#rag-tutorial)

</div>

## 🌟 Overview

LangXChange is a powerful, comprehensive Python toolkit designed to streamline LLM operations and Retrieval-Augmented Generation (RAG) implementations. It provides unified interfaces for multiple LLM providers, vector databases, document processing, database integration, and local model management.

### 🚀 Key Features

- **🤖 Multi-LLM Support**: OpenAI, Anthropic, Google GenAI, DeepSeek, Llama
- **🔍 Vector Databases**: ChromaDB, Pinecone, FAISS integration
- **📄 Document Processing**: Universal document loader with multiple formats
- **🧠 RAG Implementation**: Two-stage retrieval with cross-encoder
- **🏠 Local LLM**: Model downloading, fine-tuning, quantization
- **💾 Database Integration**: MySQL, MongoDB support
- **💰 Cost Tracking**: Built-in cost monitoring for API calls
- **⚡ Performance**: Caching, async operations, batch processing

## 📦 Installation

### Via PyPI (Recommended)
```bash
pip install langxchange
```

<!-- ### From Source
```bash
git clone https://github.com/yourusername/langxchange.git
cd langxchange
pip install -e .
```

### Development Installation
```bash
git clone https://github.com/yourusername/langxchange.git
cd langxchange
pip install -e ".[dev]"
``` -->

### Environment Variables
Create a `.env` file in your project directory:

```bash
# OpenAI
OPENAI_API_KEY=your_openai_key

# MySQL Configuration
MYSQL_HOST=localhost
MYSQL_DB=your_database
MYSQL_USER=your_username
MYSQL_PASSWORD=your_password
MYSQL_PORT=3306
MYSQL_CHARSET=utf8mb4

# ChromaDB
CHROMA_PERSIST_PATH=./chroma_db

# Vector Databases
PINECONE_API_KEY=your_pinecone_key
PINECONE_ENVIRONMENT=your_environment

# Milvus Configuration
MILVUS_HOST=localhost
MILVUS_PORT=19530
MILVUS_API_KEY=your_milvus_token  # Optional for local, required for Zilliz Cloud

# Elasticsearch Configuration
ELASTICSEARCH_HOST=http://localhost:9200
```

## 🚀 Quick Start

```python
import os
from langxchange.openai_helper import EnhancedOpenAIHelper, OpenAIConfig
from langxchange.chroma_helper import EnhancedChromaHelper, ChromaConfig
from langxchange.documentloader import DocumentLoaderHelper, ChunkingStrategy
from langxchange.embeddings import EmbeddingHelper

# Set API key (or use environment variable)
os.environ["OPENAI_API_KEY"] = "your-api-key"

# Configure OpenAI with enhanced settings
openai_config = OpenAIConfig(
    chat_model="gpt-4",
    enable_caching=True,
    enable_cost_tracking=True,
    max_retries=3
)

# Initialize OpenAI client
llm = EnhancedOpenAIHelper(openai_config)

# Configure ChromaDB with enhanced settings
chroma_config = ChromaConfig(
    persist_directory="./chroma_db",
    batch_size=100,
    progress_bar=True
)

# Initialize ChromaDB vector store
chroma = EnhancedChromaHelper(llm, chroma_config)

# Load and process documents with semantic chunking
loader = DocumentLoaderHelper(
    chunking_strategy=ChunkingStrategy.SEMANTIC,
    chunk_size=800,
    preserve_formatting=True
)

# Process documents and store in vector database
documents = list(loader.load("document.pdf"))
chroma.insert_documents(
    collection_name="my_collection",
    documents=[doc.content for doc in documents],
    metadatas=[doc.metadata for doc in documents],
    generate_embeddings=True
)

# Query the vector database
results = chroma.query_collection(
    collection_name="my_collection",
    query_text="What is machine learning?",
    top_k=5
)
```

## 📚 Modules

### 🤖 LLM Providers

#### OpenAI Integration (`openai_helper.py`)
```python
from langxchange.openai_helper import EnhancedOpenAIHelper, OpenAIConfig

# Configure OpenAI with enhanced settings
open_ai_config = OpenAIConfig(
    chat_model="gpt-4",
    enable_caching=True,
    enable_cost_tracking=True,
    max_retries=3,
    log_level="INFO"
)

openai = EnhancedOpenAIHelper(open_ai_config)

response = openai.generate(
    prompt="Explain quantum computing in simple terms.",
    system_message="You are a helpful AI assistant."
)

# Cost tracking
cost = openai.get_cost_summary()
print(f"Total cost: ${cost['total_cost']:.4f}")
```

**Features:**
- ✅ API key management and validation
- ✅ Response caching for cost optimization
- ✅ Cost tracking and reporting
- ✅ Support for all OpenAI models (GPT-3.5, GPT-4, embeddings)
- ✅ Batch processing capabilities
- ✅ Error handling and retry logic

#### Anthropic Integration (`anthropic_helper.py`)
```python
import os
from langxchange.anthropic_helper import EnhancedAnthropicHelper, AnthropicConfig

# Set API key (or use environment variable)
os.environ["ANTHROPIC_API_KEY"] = "your-anthropic-key"

# Configure Anthropic with enhanced settings
anthropic_config = AnthropicConfig(
    model="claude-3-sonnet-20240229",
    enable_caching=True,
    enable_cost_tracking=True,
    max_retries=3,
    log_level="INFO"
)

# Initialize Anthropic client
anthropic = EnhancedAnthropicHelper(anthropic_config)

response = anthropic.generate(
    prompt="Analyze the following text for sentiment and key themes.",
    max_tokens=500,
    system_message="You are a helpful AI assistant."
)

# Cost tracking
cost = anthropic.get_cost_summary()
print(f"Total cost: ${cost['total_cost']:.4f}")
```

**Features:**
- ✅ Claude model support (Haiku, Sonnet, Opus)
- ✅ Enhanced configuration and caching
- ✅ Cost tracking and reporting
- ✅ Context window optimization
- ✅ Token counting and cost estimation
- ✅ Streaming responses
- ✅ Error handling and retry logic

#### Google GenAI Integration (`google_genai_helper.py`) — **Updated in v0.4.7**

```python
import os
import asyncio
from langxchange.google_genai_helper import EnhancedGoogleGenAIHelper, GoogleGenAIHelper

# Set API key (or use environment variable)
os.environ["GOOGLE_API_KEY"] = "your-google-key"

# ── EnhancedGoogleGenAIHelper (recommended) ──────────────────────────────────
google_genai = EnhancedGoogleGenAIHelper(
    api_key="your-google-key",
    chat_model="gemini-2.0-flash",
    vision_model="gemini-2.0-flash",
    tts_model="gemini-2.0-flash-preview-tts",
    enable_usage_tracking=True,
    enable_context_caching=True
)

# ── Synchronous multi-turn chat ───────────────────────────────────────────────
messages = [
    {"role": "system",    "content": "You are a helpful research assistant."},
    {"role": "user",      "content": "Summarize the following research paper."},
]
response, usage = google_genai.chat(messages, temperature=0.3, max_tokens=1000)
print(response)

# ── Async chat (non-blocking, for FastAPI / async backends) ───────────────────
async def async_example():
    response_text, usage, tool_calls = await google_genai.chat_async(
        messages=messages,
        temperature=0.7,
        max_tokens=2000,
    )
    print(response_text)
    print(f"Tokens used: {usage.get('total_tokens', 0)}")

asyncio.run(async_example())

# ── Streaming chat ────────────────────────────────────────────────────────────
async def stream_example():
    async for chunk in google_genai.chat_stream(messages, temperature=0.7):
        print(chunk, end="", flush=True)

asyncio.run(stream_example())

# ── Tool / function calling ───────────────────────────────────────────────────
tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get current weather for a city",
        "parameters": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"]
        }
    }
}]

async def tool_example():
    content, usage, tool_calls = await google_genai.chat_async(
        messages=[{"role": "user", "content": "What's the weather in London?"}],
        tools=tools
    )
    if tool_calls:
        for call in tool_calls:
            print(f"Tool: {call['function']['name']}, Args: {call['function']['arguments']}")

asyncio.run(tool_example())

# ── Multi-modal vision processing ─────────────────────────────────────────────
vision_response = google_genai.chat_with_vision(
    text="Analyze this image for key insights",
    image_path="research_chart.png"
)

# ── Text-to-speech generation ─────────────────────────────────────────────────
audio_path = google_genai.text_to_speech(
    text="Welcome to the research presentation",
    voice="Zephyr",
    output_format="wav"
)

# ── Usage tracking ────────────────────────────────────────────────────────────
stats = google_genai.get_usage_statistics()
print(f"Total requests: {stats.chat_requests}")
print(f"Input tokens:   {stats.total_input_tokens}")

# ── Lightweight GoogleGenAIHelper (simple use cases) ─────────────────────────
llm = GoogleGenAIHelper(chat_model="gemini-2.0-flash")
async def simple_chat():
    reply = await llm.chat(messages)
    print(reply)
asyncio.run(simple_chat())
```

**Features (v0.5.0):**
- ✅ **Native multi-turn chat** — messages correctly formatted as `types.Content` with `user`/`model` roles
- ✅ **System instruction support** — `system` messages extracted and passed via `GenerateContentConfig.system_instruction`
- ✅ **Async chat** (`chat_async`) — non-blocking, returns `(text, usage, tool_calls)` tuple
- ✅ **Streaming chat** (`chat_stream`) — async generator yielding text chunks in real time
- ✅ **Tool / function calling** — OpenAI-style tool definitions auto-converted to `types.FunctionDeclaration`; tool calls returned in OpenAI-compatible format
- ✅ **Dual client** — both sync `genai.Client` and async `genai.AsyncClient` initialized on startup
- ✅ Gemini 2.0 Flash model support
- ✅ Multi-modal capabilities (text, images, audio)
- ✅ Text-to-speech and speech-to-text generation
- ✅ Context management and usage statistics
- ✅ Safety filtering and content moderation
- ✅ Embedding generation with `text-embedding-004`
- ✅ Caching and performance optimization

#### DeepSeek Integration (`deepseek_helper.py`)
```python
import os
from langxchange.deepseek_helper import EnhancedDeepSeekHelper, ModelType, ContextManagementStrategy

# Set API key (or use environment variable)
os.environ["DEEPSEEK_API_KEY"] = "your-deepseek-key"

# Initialize Enhanced DeepSeek client with advanced configuration
deepseek = EnhancedDeepSeekHelper(
    api_key="your-deepseek-key",
    base_url="https://api.deepseek.com/v1",
    default_model=ModelType.CHAT.value,
    embed_model=ModelType.EMBEDDING.value,
    vision_model=ModelType.VISION.value,
    timeout=30,
    max_retries=3,
    enable_logging=True,
    log_level="INFO",
    max_context_tokens=30000,
    context_strategy=ContextManagementStrategy.SLIDING_WINDOW
)

# Generate text response with enhanced features
response = deepseek.generate(
    prompt="Write a Python function for binary search.",
    max_tokens=500,
    temperature=0.3,
    system_message="You are a helpful coding assistant.",
    context_messages=[{"role": "user", "content": "Previous context"}]
)

# Code generation with syntax highlighting
code_response = deepseek.generate_code(
    prompt="Create a REST API endpoint in Flask",
    language="python",
    include_docs=True,
    error_handling=True
)

# Batch processing
batch_responses = deepseek.batch_generate([
    {"prompt": "Explain recursion", "max_tokens": 200},
    {"prompt": "Define Big O notation", "max_tokens": 200}
], temperature=0.7)

# Usage and cost tracking
cost_summary = deepseek.get_cost_summary()
print(f"Total cost: ${cost_summary['total_cost']:.4f}")
print(f"Total tokens: {cost_summary['total_tokens']}")
```

**Features:**
- ✅ Cost-effective alternative to OpenAI with enhanced features
- ✅ Multiple model types (chat, embedding, vision)
- ✅ Advanced context management with sliding window strategy
- ✅ Code generation with syntax highlighting and documentation
- ✅ Batch processing capabilities for multiple requests
- ✅ Streaming support with real-time response handling
- ✅ Usage tracking and cost monitoring
- ✅ Error handling and retry logic
- ✅ Compatible with OpenAI API format

#### Llama Integration (`llama_helper.py`)
```python
from langxchange.llama_helper import EnhancedLLaMAHelper, LLaMAConfig
import os

# Set Hugging Face token (or use environment variable)
os.environ["HUGGINGFACE_TOKEN"] = "your-hf-token"

# Configure LLaMA with enhanced settings
llama_config = LLaMAConfig(
    chat_model="meta-llama/Llama-2-7b-chat-hf",
    embed_model="all-MiniLM-L6-v2",
    device="auto",
    max_memory_per_gpu="8GB",
    load_in_8bit=False,
    load_in_4bit=True,
    cache_dir="./llama_cache",
    trust_remote_code=False
)

# Initialize Enhanced LLaMA client
llama = EnhancedLLaMAHelper(config=llama_config)

# Generate text response
response = llama.generate(
    prompt="Explain machine learning fundamentals.",
    temperature=0.7,
    max_tokens=2048,
    system_message="You are a knowledgeable AI assistant specializing in ML.",
    do_sample=True,
    top_p=0.9
)

# Advanced text generation with stopping criteria
advanced_response = llama.generate_advanced(
    prompt="Write a Python class for neural networks",
    stopping_criteria=["def ", "class ", "\n\n"],
    temperature=0.5,
    repetition_penalty=1.1,
    no_repeat_ngram_size=3
)

# Batch text processing
batch_responses = llama.batch_generate([
    {"prompt": "What is deep learning?", "max_tokens": 200},
    {"prompt": "Explain neural networks", "max_tokens": 200},
    {"prompt": "Define backpropagation", "max_tokens": 200}
], temperature=0.7)

# Token counting and optimization
token_count = llama.count_tokens("This is a sample text for token counting.")
print(f"Token count: {token_count}")

# Model performance metrics
metrics = llama.get_model_metrics()
print(f"Memory usage: {metrics['memory_usage_gb']:.2f} GB")
print(f"Inference speed: {metrics['tokens_per_second']:.2f} tokens/sec")
```

**Features:**
- ✅ Local model deployment with Hugging Face integration
- ✅ Enhanced quantization support (4-bit, 8-bit)
- ✅ GPU acceleration with memory optimization
- ✅ Advanced configuration with LLaMAConfig
- ✅ Multiple quantization modes for different hardware
- ✅ Custom stopping criteria for precise generation control
- ✅ Batch processing capabilities
- ✅ Token counting and performance monitoring
- ✅ Memory management and optimization
- ✅ Support for various LLaMA model variants
- ✅ Hugging Face model hub integration

#### Model Context Protocol (MCP) Integration (`mcp_helper.py`)
```python
import asyncio
from langxchange.mcp_helper import MCPServiceManager

async def main():
    # Initialize manager with JSON config
    manager = MCPServiceManager("mcp_config.json")
    
    # 1. Register functional capabilities for servers with priority
    manager.register_server_capabilities("filesystem", ["files", "local_storage"], priority=10)
    manager.register_server_capabilities("brave_search", ["web_search", "research"], priority=5)
    
    # 2. Intelligent Routing: Resolve server by tool name or capability
    # Automatically selects the best server (health + priority aware)
    server = await manager.select_best_server_for_tool("read_file")
    
    # Direct namespace resolution
    server = manager.resolve_tool_server("filesystem::read_file")
    
    # Name-based resolution with capability hints
    context = {"preferred_capability": "web_search"}
    server = manager.resolve_tool_server("search", context=context)
    
    # 3. Health & Priority Aware Selection
    # Automatically picks the best server based on priority, error rates and latency
    best_server = manager.select_best_server(["server_a", "server_b"])
    
    # 4. Discovery: Fetch all tools with routing metadata
    all_tools = await manager.get_all_tools_with_metadata()
    
    # 5. Call a tool (standard way)
    result = await manager.call_tool(
        server_name="filesystem",
        tool_name="read_file",
        arguments={"path": "data.txt"}
    )
    
    await manager.shutdown()

if __name__ == "__main__":
    asyncio.run(main())
```

**Features:**
- ✅ **Standardized Interoperability**: Connect to any MCP-compliant server (stdio or SSE)
- ✅ **Intelligent Routing**: Automatic server resolution based on tool names and namespaces (`server::tool`)
- ✅ **Capability-Based Selection**: Route tasks to servers based on functional tags (e.g., "web_search", "filesystem")
- ✅ **Health & Priority Aware Selection**: Automatically prioritizes healthy servers and uses priority scores for optimal routing
- ✅ **Tool Registry**: Comprehensive metadata registry for all available tools across all connected servers
- ✅ **Lifecycle Management**: Automatic server startup, health monitoring, and recovery
- ✅ **Discovery**: Dynamic tool discovery with TTL-based caching
- ✅ **Production-Ready**: Graceful shutdown, detailed logging, and error handling

#### 🛠️ Complete Example: Calculator Server

This example demonstrates a full setup including a mock server, configuration, and execution script.

````carousel
```python
# mcp_calculator_server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Calculator")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """Multiply two numbers"""
    return a * b

if __name__ == "__main__":
    mcp.run(transport="stdio")
```
<!-- slide -->
```json
// mcp_test_config.json
{
  "servers": [
    {
      "name": "calculator",
      "transport": "stdio",
      "command": "python3",
      "args": ["mcp_calculator_server.py"]
    }
  ]
}
```
<!-- slide -->
```python
# mcp_test_execution.py
import asyncio
from langxchange.mcp_helper import MCPServiceManager

async def run_test():
    manager = MCPServiceManager("mcp_test_config.json")
    await manager.initialize()
    
    try:
        # Call 'add' tool
        result = await manager.call_tool(
            server_name="calculator",
            tool_name="add",
            arguments={"a": 5, "b": 3}
        )
        print(f"Add Result: {result}")
        
    finally:
        await manager.shutdown()

if __name__ == "__main__":
    asyncio.run(run_test())
```
<!-- slide -->
```text
# Expected Output
INFO:langxchange.mcp:MCPServiceManager initialized
INFO:langxchange.mcp:Started MCP server 'calculator'
Add Result: content=[TextContent(type='text', text='8', ...)] structuredContent={'result': 8}
INFO:langxchange.mcp:Stopped MCP server 'calculator'
```
````

### 🤖 Autonomous Agents (`EnhancedAgent.py`)

LangXChange 0.4.6 introduces a powerful `EnhancedLLMAgentHelper` for building production-ready autonomous agents. It features dynamic tool discovery, semantic memory, per-tool circuit breakers, and automatic observation summarization.

#### 🚀 Quick Start: Manual Tool Definition

```python
import asyncio
from langxchange.EnhancedAgent import EnhancedLLMAgentHelper
from langxchange.agent_memory_helper import AgentMemoryHelper
from langxchange.openai_helper import EnhancedOpenAIHelper, OpenAIConfig

async def main():
    # 1. Initialize LLM and Memory
    llm = EnhancedOpenAIHelper(OpenAIConfig(chat_model="gpt-4o"))
    memory = AgentMemoryHelper(sqlite_path="agent_memory.db")

    # 2. Define tools with JSON Schema for strict parameter generation
    async def get_weather(params):
        return f"The weather in {params['location']} is sunny, 25°C."

    tools = [{
        "action": "get_weather",
        "description": "Get current weather for a location",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "City and country"}
            },
            "required": ["location"]
        },
        "func": get_weather
    }]

    # 3. Initialize Agent
    agent = EnhancedLLMAgentHelper(
        llm=llm,
        action_space=tools,
        external_memory_helper=memory,
        debug=True
    )

    # 4. Run autonomously to achieve a goal
    agent.set_goal("What is the weather in London?")
    results = await agent.run_autonomous(max_cycles=5)
    
    for res in results:
        print(f"Thought: {res['thought']}")
        print(f"Action: {res['decision']['action']}")
        print(f"Result: {res['outcome']['result']}")

if __name__ == "__main__":
    asyncio.run(main())
```

#### 🔌 Native MCP Integration

Instead of manually defining tools, you can pass an MCP configuration. The agent will automatically discover all tools from the configured servers and route calls dynamically.

```python
mcp_config = {
    "servers": [
        {
            "name": "filesystem",
            "transport": "stdio",
            "command": "mcp-server-filesystem",
            "args": ["/home/user/Downloads"]
        },
        {
            "name": "brave-search",
            "transport": "stdio",
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-brave-search"],
            "env": {"BRAVE_API_KEY": "your-key"}
        }
    ]
}

agent = EnhancedLLMAgentHelper(
    llm=llm,
    mcp_config=mcp_config, # Native MCP support
    external_memory_helper=memory
)

# The agent will now have access to all tools from both filesystem and brave-search
agent.set_goal("Find the latest PDF in my Downloads and search for its main topic on the web.")
await agent.run_autonomous()
```

#### ✨ Key Features

- ✅ **Autonomous Loops**: Use `run_autonomous()` for multi-step goal achievement with automatic state management.
- ✅ **Dynamic Discovery**: Tools are discovered at runtime from MCP servers or via a `discovery_callback`.
- ✅ **Schema-Strict Parameters**: Uses JSON Schema hints in prompts to ensure the LLM generates valid parameters.
- ✅ **Per-Tool Circuit Breakers**: Failures in one tool (e.g., a flaky API) won't crash the entire agent.
- ✅ **Auto-Summarization**: Large tool outputs are automatically summarized to preserve context window space.
- ✅ **Semantic Memory**: Integrates with `AgentMemoryHelper` for long-term history and semantic retrieval.
- ✅ **Observability**: Built-in Prometheus-style metrics and correlation ID tracing across all operations.

### 🔍 Vector Database Integration

#### ChromaDB Integration (`chroma_helper.py`)
```python
from langxchange.chroma_helper import EnhancedChromaHelper, ChromaConfig

# Configure Chroma with enhanced performance settings
chroma_config = ChromaConfig(
    persist_directory="./chroma_db",
    batch_size=100,
    max_workers=8,
    progress_bar=True
)

chroma = EnhancedChromaHelper(llm, chroma_config)

# Insert documents with metadata
chroma.insert_documents(
    collection_name="my_collection",
    documents=["Document content here"],
    metadatas=[{"source": "file1.txt", "type": "text"}],
    generate_embeddings=True
)

# Query with similarity search
results = chroma.query_collection(
    collection_name="my_collection",
    query_text="What is machine learning?",
    top_k=5
)
```

**Features:**
- ✅ Persistent storage
- ✅ Metadata filtering
- ✅ Batch operations
- ✅ Collection management
- ✅ Performance optimization

#### Pinecone Integration (`pinecone_helper.py`)
```python
import os
from langxchange.pinecone_helper import EnhancedPineconeHelper, PineconeConfig, CloudProvider, MetricType
from langxchange.openai_helper import EnhancedOpenAIHelper, OpenAIConfig

# Set API key (or use environment variable)
os.environ["PINECONE_API_KEY"] = "your-pinecone-key"

# Configure LLM helper for embeddings
openai_config = OpenAIConfig(enable_caching=True)
llm_helper = EnhancedOpenAIHelper(openai_config)

# Configure Pinecone with enhanced settings
pinecone_config = PineconeConfig(
    api_key="your-pinecone-key",
    environment="us-west1-gcp",
    cloud_service=CloudProvider.GCP,
    index_name="my-index",
    dimension=1536,  # OpenAI embedding dimension
    metric=MetricType.COSINE,
    batch_size=100,
    max_workers=10,
    progress_bar=True
)

# Initialize Enhanced Pinecone client
pinecone = EnhancedPineconeHelper(
    llm_helper=llm_helper,
    config=pinecone_config
)

# Insert documents with automatic embedding generation
documents = [
    "Machine learning is a subset of AI that focuses on algorithms.",
    "Deep learning uses neural networks with multiple layers.",
    "Natural language processing enables computers to understand text."
]

pinecone.insert_documents(
    collection_name="my-index",
    documents=documents,
    metadatas=[{"source": "article1", "topic": "AI"}, 
               {"source": "article2", "topic": "ML"}, 
               {"source": "article3", "topic": "NLP"}],
    generate_embeddings=True,
    namespace="default"
)

# Query for similar vectors with filters
results = pinecone.query(
    vector=None,  # Will auto-generate embedding from query_text
    query_text="What is machine learning?",
    top_k=5,
    filter_metadata={"topic": "AI"},
    include_metadata=True
)

# DataFrame ingestion with batch processing
import pandas as pd

df = pd.DataFrame({
    'text': ['Sample text 1', 'Sample text 2', 'Sample text 3'],
    'category': ['tech', 'science', 'tech']
})

stats = pinecone.ingest_dataframe(
    collection_name="my-index",
    dataframe=df,
    text_column='text',
    metadata_columns=['category'],
    namespace="default"
)

print(f"Inserted {stats['inserted']} documents")
print(f"Skipped {stats['skipped']} documents")
```

**Features:**
- ✅ Cloud-based vector storage with auto-scaling
- ✅ Automatic embedding generation using LLM helpers
- ✅ DataFrame ingestion with batch processing
- ✅ Advanced querying with metadata filters
- ✅ Performance monitoring and statistics
- ✅ Enterprise-grade error handling and retries
- ✅ Memory-efficient operations for large datasets
- ✅ Namespace management and resource cleanup
- ✅ Real-time updates with comprehensive logging

#### Milvus Integration (`milvus_helper.py`)
```python
from langxchange.milvus_helper import EnhancedMilvusHelper, MilvusConfig
from langxchange.openai_helper import EnhancedOpenAIHelper, OpenAIConfig

# Configure LLM helper for embeddings
openai_config = OpenAIConfig(enable_caching=True)
llm_helper = EnhancedOpenAIHelper(openai_config)

# Configure Milvus with enhanced settings
milvus_config = MilvusConfig(
    host="localhost",
    port="19530",
    api_key="your-milvus-token", # Optional for local
    collection_prefix="lx_",
    embedding_dim=1536,
    batch_size=100,
    progress_bar=True
)

# Initialize Enhanced Milvus client
milvus = EnhancedMilvusHelper(
    llm_helper=llm_helper,
    config=milvus_config
)

# Insert documents with automatic embedding generation
documents = [
    "Milvus is an open-source vector database built for AI applications.",
    "It supports high-performance vector similarity search and analytics.",
    "Milvus can handle billions of vectors with millisecond latency."
]

milvus.insert_documents(
    collection_name="ai_docs",
    documents=documents,
    metadatas=[{"category": "database"}, {"category": "search"}, {"category": "performance"}],
    generate_embeddings=True
)

# Query for similar vectors
results = milvus.query(
    collection_name="ai_docs",
    query_text="What is Milvus?",
    top_k=3
)

for hit in results[0]:
    print(f"Score: {hit.score}")
    print(f"Document: {hit.entity.get('document')}")
```

**Features:**
- ✅ High-performance vector search with HNSW index
- ✅ Support for local Milvus and Zilliz Cloud (via API key/token)
- ✅ Automatic collection creation and schema management
- ✅ Batch insertion and DataFrame ingestion
- ✅ Metadata filtering and JSON support
- ✅ Enterprise-grade error handling

#### Elasticsearch Integration (`elasticsearch_helper.py`)
```python
from langxchange.elasticsearch_helper import EnhancedElasticsearchHelper, ElasticsearchConfig
from langxchange.openai_helper import EnhancedOpenAIHelper, OpenAIConfig

# Configure LLM helper for embeddings
openai_config = OpenAIConfig(enable_caching=True)
llm_helper = EnhancedOpenAIHelper(openai_config)

# Configure Elasticsearch with enhanced settings
es_config = ElasticsearchConfig(
    host="http://localhost:9200",
    index_prefix="lx_",
    embedding_dim=1536,
    batch_size=100,
    progress_bar=True
)

# Initialize Enhanced Elasticsearch client
es = EnhancedElasticsearchHelper(
    llm_helper=llm_helper,
    config=es_config
)

# Insert documents with automatic embedding generation
documents = [
    "Elasticsearch is a distributed, RESTful search and analytics engine.",
    "It provides a distributed, multitenant-capable full-text search engine.",
    "Elasticsearch is developed in Java and is open source."
]

es.insert_documents(
    collection_name="tech_docs",
    documents=documents,
    metadatas=[{"topic": "search"}, {"topic": "analytics"}, {"topic": "java"}],
    generate_embeddings=True
)

# Query for similar vectors
results = es.query(
    collection_name="tech_docs",
    query_text="What is Elasticsearch?",
    top_k=3
)

for hit in results['hits']['hits']:
    print(f"Score: {hit['_score']}")
    print(f"Document: {hit['_source'].get('document')}")
```

**Features:**
- ✅ High-performance vector search with `dense_vector` type
- ✅ Support for script-based cosine similarity scoring
- ✅ Automatic index creation and mapping management
- ✅ Batch insertion and DataFrame ingestion
- ✅ Metadata filtering support
- ✅ Unified interface consistent with Chroma and Milvus helpers

#### FAISS Integration (`faiss_helper.py`)
```python
import os
import pandas as pd
from langxchange.faiss_helper import EnhancedFAISSHelper

# Initialize Enhanced FAISS helper with advanced configuration
faiss = EnhancedFAISSHelper(
    dim=768,  # Vector dimension
    index_type="ivf",  # Use IVF index for better performance on large datasets
    normalize_vectors=True,  # Normalize vectors for cosine similarity
    nlist=100,  # Number of clusters for IVF
    auto_train=True  # Automatically train IVF indices
)

# Insert individual vectors with documents and metadata
documents = [
    "Machine learning is a subset of artificial intelligence.",
    "Deep learning uses neural networks with multiple layers.",
    "Natural language processing enables computers to understand text.",
    "Computer vision allows machines to interpret visual information."
]

metadatas = [
    {"source": "article1", "topic": "ML", "date": "2025-01-15"},
    {"source": "article2", "topic": "DL", "date": "2025-01-16"},
    {"source": "article3", "topic": "NLP", "date": "2025-01-17"},
    {"source": "article4", "topic": "CV", "date": "2025-01-18"}
]

# Generate embeddings (this would come from your embedding model)
embeddings = [
    [0.1, 0.2, 0.3] * 256,  # 768-dimensional embedding
    [0.4, 0.5, 0.6] * 256,
    [0.7, 0.8, 0.9] * 256,
    [0.2, 0.3, 0.4] * 256
]

faiss.insert(
    vectors=embeddings,
    documents=documents,
    metadatas=metadatas
)

# DataFrame integration with batch processing
df = pd.DataFrame({
    'embeddings': embeddings,
    'documents': documents,
    'metadata': metadatas
})

faiss.insert_dataframe(
    dataframe=df,
    embeddings_col="embeddings",
    documents_col="documents", 
    metadata_col="metadata"
)

# Query for similar vectors with comprehensive results
query_vector = [0.1, 0.2, 0.3] * 256  # Sample query embedding
results = faiss.query(
    embedding_vector=query_vector,
    top_k=3,
    include_distances=True  # Include similarity scores
)

print(f"Found {len(results)} similar documents:")
for i, result in enumerate(results, 1):
    print(f"{i}. Score: {result['distance']:.3f}")
    print(f"   Document: {result['document'][:100]}...")
    print(f"   Metadata: {result['metadata']}")
    print()

# Batch querying for multiple queries at once
query_vectors = [
    [0.1, 0.2, 0.3] * 256,
    [0.4, 0.5, 0.6] * 256
]

batch_results = faiss.query_batch(
    embedding_vectors=query_vectors,
    top_k=2,
    include_distances=True
)

print(f"Batch query results: {len(batch_results)} result sets")

# Retrieve documents by ID
doc_results = faiss.get_by_ids(["id_0", "id_1"])
print(f"Retrieved {len(doc_results)} documents by ID")

# Get comprehensive statistics
stats = faiss.get_stats()
print(f"Index Statistics:")
print(f"  Total vectors: {stats['total_vectors']}")
print(f"  Index type: {stats['index_type']}")
print(f"  Dimension: {stats['dimension']}")
print(f"  Is trained: {stats.get('is_trained', 'N/A')}")
print(f"  Number of clusters: {stats.get('nlist', 'N/A')}")

# Persistence - save and load index
index_path = "./faiss_index.bin"
metadata_path = "./faiss_metadata.pkl"

faiss.save(index_path, metadata_path)
print(f"Saved index to {index_path}")

# Load saved index
loaded_faiss = EnhancedFAISSHelper(dim=768, index_type="ivf")
loaded_faiss.load(index_path, metadata_path)
print(f"Loaded index with {loaded_faiss.count()} vectors")

# Index management
index_vector_count = faiss.count()
print(f"Current index contains {index_vector_count} vectors")

# Delete specific document
deleted = faiss.delete_by_id("id_0")
print(f"Document deleted: {deleted}")

# Rebuild index with different configuration
rebuilt_count = faiss.rebuild_index(index_type="hnsw")
print(f"Rebuilt index with {rebuilt_count} vectors using HNSW")
```

**Features:**
- ✅ Multiple index types (Flat, IVF, HNSW) with automatic training
- ✅ High-performance similarity search with vector normalization
- ✅ Comprehensive DataFrame integration for pandas workflow
- ✅ Batch operations for efficient processing
- ✅ Advanced querying with distance scores and metadata
- ✅ Persistence and index management
- ✅ Document retrieval by ID and batch operations
- ✅ Comprehensive statistics and performance monitoring
- ✅ Index rebuilding and optimization
- ✅ Memory-efficient operations with proper validation

### 📄 Document Processing

#### Document Loader (`documentloader.py`)
```python
"""
Demo script showing the enhanced DocumentLoaderHelper capabilities
optimized for LLM processing.
"""
import os
import sys
from langxchange.documentloader import DocumentLoaderHelper, ChunkingStrategy, ImageProcessingStrategy

# Initialize Document Loader with different configurations
loader = DocumentLoaderHelper(
    chunk_size=800,
    overlap_size=100,
    chunking_strategy=ChunkingStrategy.SEMANTIC,
    preserve_formatting=True
)

# Test different chunking strategies
strategies = [
    (ChunkingStrategy.CHARACTER, "Character-based chunking"),
    (ChunkingStrategy.SENTENCE, "Sentence-aware chunking"),
    (ChunkingStrategy.PARAGRAPH, "Paragraph-aware chunking"),
    (ChunkingStrategy.SEMANTIC, "Semantic chunking (recommended)"),
]

# Add token-based if available
try:
    import tiktoken
    strategies.append((ChunkingStrategy.TOKEN, "Token-based chunking"))
except ImportError:
    print("Note: tiktoken not available, skipping token-based chunking")

# Process document with specific strategy
file_path = "documents/sample.txt"

for strategy, description in strategies:
    print(f"\n--- {description} ---")
    
    # Initialize loader with strategy
    loader = DocumentLoaderHelper(
        chunk_size=800,
        overlap_size=100,
        chunking_strategy=strategy,
        preserve_formatting=True
    )
    
    try:
        chunks = list(loader.load(file_path))
        
        print(f"Total chunks created: {len(chunks)}")
        print(f"Processing time: {loader.stats['times']['total']:.3f}s")
        
        # Show first few chunks
        for i, chunk in enumerate(chunks[:3]):
            print(f"\nChunk {i+1}:")
            print(f"  Length: {len(chunk.content)} chars")
            if chunk.metadata.token_count:
                print(f"  Tokens: {chunk.metadata.token_count}")
            print(f"  Content preview: {chunk.content[:100]}...")
            
    except Exception as e:
        print(f"Error with {strategy}: {e}")

# Multi-format document processing
loader = DocumentLoaderHelper(
    chunk_size=800,
    chunking_strategy=ChunkingStrategy.SEMANTIC,
    preserve_formatting=True
)

files_to_process = [
    "documents/sample.txt",
    "documents/data.csv",
    "documents/report.pdf"
]

for file_path in files_to_process:
    if os.path.exists(file_path):
        print(f"\n--- Processing: {file_path} ---")
        
        try:
            chunks = list(loader.load(file_path))
            
            print(f"File type: {chunks[0].metadata.file_type}")
            print(f"Total chunks: {len(chunks)}")
            
            # Show metadata for first chunk
            first_chunk = chunks[0]
            print(f"First chunk metadata:")
            print(f"  Source: {first_chunk.metadata.source_file}")
            print(f"  Section: {first_chunk.metadata.section_title}")
            print(f"  Content length: {len(first_chunk.content)} chars")
            
        except Exception as e:
            print(f"Error processing {file_path}: {e}")

# Advanced configuration examples
configs = [
    {
        "name": "Minimal overlap",
        "params": {"chunk_size": 400, "overlap_size": 20, "min_chunk_size": 30}
    },
    {
        "name": "High overlap",
        "params": {"chunk_size": 400, "overlap_size": 100, "min_chunk_size": 50}
    },
    {
        "name": "Preserve formatting",
        "params": {"chunk_size": 400, "preserve_formatting": True}
    },
    {
        "name": "Normalize text",
        "params": {"chunk_size": 400, "preserve_formatting": False}
    }
]

file_path = "documents/sample.txt"

for config in configs:
    print(f"\n--- CONFIG: {config['name']} ---")
    
    loader = DocumentLoaderHelper(
        chunking_strategy=ChunkingStrategy.SEMANTIC,
        **config['params']
    )
    
    try:
        chunks = list(loader.load(file_path))
        stats = loader.get_statistics()
        
        print(f"Chunks created: {len(chunks)}")
        print(f"Avg chunk size: {sum(len(c.content) for c in chunks) / len(chunks):.0f} chars")
        print(f"Processing time: {stats['processing_stats']['times']['total']:.3f}s")
        
    except Exception as e:
        print(f"Error: {e}")

# Image processing support
try:
    from PIL import Image
    pil_available = True
    print("✅ PIL (Pillow) support: Available")
except ImportError:
    pil_available = False
    print("❌ PIL (Pillow) support: Not available")

try:
    import pytesseract
    tesseract_available = True
    print("✅ OCR (pytesseract) support: Available")
except ImportError:
    tesseract_available = False
    print("❌ OCR (pytesseract) support: Not available")

# Image processing strategies
strategies = [
    (ImageProcessingStrategy.OCR_TEXT, "Extract text using OCR"),
    (ImageProcessingStrategy.DESCRIPTION, "Generate image d