LLM Document Reading API¶

from nexus import connect

async def main():
    nx = connect()

    # Simple usage - just get the answer
    answer = await nx.llm_read(
        path="/reports/q4.pdf",
        prompt="What were the top 3 challenges?",
        model="claude-sonnet-4"
    )
    print(answer)

async def llm_read(
    path: str,
    prompt: str,
    model: str = "claude-sonnet-4",
    max_tokens: int = 1000,
    api_key: str | None = None,
    use_search: bool = True,
    search_mode: str = "semantic",
    provider: LLMProvider | None = None,
) -> str

answer = await nx.llm_read(
    "/reports/q4.pdf",
    "What were the key achievements?",
    model="claude-sonnet-4",
    max_tokens=500
)
print(answer)

async def llm_read_detailed(
    path: str,
    prompt: str,
    model: str = "claude-sonnet-4",
    max_tokens: int = 1000,
    api_key: str | None = None,
    use_search: bool = True,
    search_mode: str = "semantic",
    search_limit: int = 10,
    include_citations: bool = True,
    provider: LLMProvider | None = None,
) -> DocumentReadResult

result = await nx.llm_read_detailed(
    "/docs/**/*.md",
    "How does authentication work?",
    model="claude-sonnet-4"
)

print(result.answer)
print(f"\nSources ({len(result.citations)}):")
for citation in result.citations:
    print(f"  • {citation.path} (score: {citation.score:.2f})")
print(f"\nCost: ${result.cost:.4f}")
print(f"Tokens: {result.tokens_used:,}")

async def llm_read_stream(
    path: str,
    prompt: str,
    model: str = "claude-sonnet-4",
    max_tokens: int = 1000,
    api_key: str | None = None,
    use_search: bool = True,
    search_mode: str = "semantic",
    provider: LLMProvider | None = None,
) -> AsyncIterator[str]

async for chunk in nx.llm_read_stream(
    "/report.pdf",
    "Analyze the financial trends",
    model="claude-sonnet-4"
):
    print(chunk, end="", flush=True)
print()  # Newline after streaming

def create_llm_reader(
    provider: LLMProvider | None = None,
    model: str | None = None,
    api_key: str | None = None,
    system_prompt: str | None = None,
    max_context_tokens: int = 3000,
) -> LLMDocumentReader

# Create custom reader
reader = nx.create_llm_reader(
    model="claude-opus-4",
    system_prompt=(
        "You are a technical documentation expert. "
        "Provide detailed, precise answers with code examples when relevant."
    )
)

# Use the reader
result = await reader.read(
    path="/docs/**/*.md",
    prompt="Explain the architecture",
    max_tokens=1000
)
print(result.answer)

# Python
answer = await nx.llm_read(
    "/docs/**/*.md",
    "How does authentication work?",
    search_mode="semantic"  # Default
)

# CLI
nexus llm read "/docs/**/*.md" "authentication" --search-mode semantic

# Python
answer = await nx.llm_read(
    "/src/**/*.py",
    "Find JWT token implementation",
    search_mode="keyword"
)

# CLI
nexus llm read "/src/**/*.py" "JWT token" --search-mode keyword

# Python
answer = await nx.llm_read(
    "/docs/**/*.md",
    "API security best practices",
    search_mode="hybrid"
)

# CLI
nexus llm read "/docs/**/*.md" "API security" --search-mode hybrid

# Python
answer = await nx.llm_read(
    "/config.yaml",
    "Explain this configuration",
    use_search=False
)

# CLI
nexus llm read /config.yaml "Explain this" --no-search

@dataclass
class DocumentReadResult:
    answer: str                      # LLM's answer
    citations: list[Citation]        # Source citations
    sources: list[str]               # Source file paths
    tokens_used: int | None          # Total tokens used
    cost: float | None               # Cost in USD
    cached: bool                     # Whether cached
    cache_savings: float | None      # Cache savings in USD

@dataclass
class Citation:
    path: str                        # Source file path
    chunk_index: int | None          # Chunk index in file
    score: float | None              # Relevance score (0-1)
    start_offset: int | None         # Start position in file
    end_offset: int | None           # End position in file

# Create reader with custom prompt
reader = nx.create_llm_reader(
    model="claude-sonnet-4",
    system_prompt=(
        "You are an executive assistant. Provide concise, "
        "bullet-point summaries focused on key business metrics and "
        "actionable insights. Use executive language and avoid technical jargon."
    )
)

# Use the customized reader
result = await reader.read(
    path="/reports/q4-2024.txt",
    prompt="Summarize Q4 performance for the executive team",
    max_tokens=400
)
print(result.answer)

from nexus.llm import LiteLLMProvider, LLMConfig

# Configure provider
config = LLMConfig(
    model="claude-sonnet-4",
    api_key="your-api-key",
    temperature=0.7,
    max_output_tokens=2000,
    timeout=60
)
provider = LiteLLMProvider(config)

# Use with LLM reading
answer = await nx.llm_read(
    "/docs/api.md",
    "What endpoints are available?",
    provider=provider
)

# Get detailed result with cost
result = await nx.llm_read_detailed(
    "/docs/**/*.md",
    "Comprehensive analysis",
    model="claude-sonnet-4"
)

print(f"Cost: ${result.cost:.4f}")
print(f"Tokens: {result.tokens_used:,}")

# Use cheaper model for simple queries
answer = await nx.llm_read(
    "/doc.txt",
    "Simple question",
    model="claude-haiku-4",  # Cheaper
    max_tokens=200           # Limit response
)

# Use keyword search to avoid embedding costs
answer = await nx.llm_read(
    "/code/**/*.py",
    "Find function definition",
    search_mode="keyword"  # No embedding costs
)

# Ask in Chinese
answer = await nx.llm_read(
    "/doc.pdf",
    "请用中文总结这个文档",
    model="claude-sonnet-4"
)

# Ask in Spanish
answer = await nx.llm_read(
    "/report.pdf",
    "Resume este informe en español",
    model="claude-sonnet-4"
)

import os

# Set remote server credentials
os.environ["NEXUS_URL"] = "http://localhost:2026"
os.environ["NEXUS_API_KEY"] = "your-api-key"

# Connect to remote server
nx = connect()

# LLM reading works exactly the same
answer = await nx.llm_read(
    "/workspace/doc.pdf",
    "What are the key points?",
    model="claude-sonnet-4"
)

from nexus.core.exceptions import (
    NexusFileNotFoundError,
    NexusPermissionError,
    ValidationError
)

try:
    answer = await nx.llm_read(
        "/reports/q4.pdf",
        "What were the challenges?",
        model="claude-sonnet-4"
    )
    print(answer)
except NexusFileNotFoundError:
    print("Document not found")
except NexusPermissionError:
    print("Permission denied")
except ValidationError as e:
    print(f"Invalid input: {e}")
except Exception as e:
    print(f"Error: {e}")

# Index documents for semantic search
nx.search_index("/workspace/docs")

# Now LLM reading will use semantic search automatically
result = await nx.llm_read_detailed(
    "/workspace/docs/**/*.md",
    "How does the system work?",
    model="claude-sonnet-4"
)

import asyncio
from nexus import connect

async def analyze_documentation():
    """Analyze project documentation with LLM."""
    nx = connect()

    # Index documents for better search
    print("Indexing documents...")
    nx.search_index("/workspace/docs")

    # Ask questions
    questions = [
        "What are the main features?",
        "How do I get started?",
        "What are the API endpoints?",
    ]

    for question in questions:
        print(f"\nQ: {question}")

        result = await nx.llm_read_detailed(
            path="/workspace/docs/**/*.md",
            prompt=question,
            model="claude-sonnet-4",
            max_tokens=500
        )

        print(f"A: {result.answer}")
        print(f"\nSources: {len(result.sources)}")
        print(f"Cost: ${result.cost:.4f}")
        print("-" * 60)

if __name__ == "__main__":
    asyncio.run(analyze_documentation())