tatlock/docs/rag_search_implementation.md

Tatlock Integration Guide

Implementation instructions for integrating Library Desk search and content extraction endpoints into the Tatlock project.

Base Configuration

BASE_URL: http://library-desk:8089  (or your deployment URL)
AUTH_HEADER: Authorization: Bearer <LIBRARY_API_KEY>

---

1. RAG Search Endpoint

Use case: Librarian needs to research a topic by searching the web.

Endpoint

POST /rag/search

Request

{
  "query": "Python async programming best practices",
  "search_type": "web",
  "limit": 10,
  "user": "tatlock-librarian"
}

Field	Type	Default	Description
query	string	required	Search query (1-500 chars)
search_type	enum	"web"	"web", "news", or "images"
limit	int	10	Results to return (1-20)
user	string	"default"	User identifier for tracking

Response

{
  "query": "Python async programming best practices",
  "search_type": "web",
  "results": [
    {
      "title": "Async IO in Python: A Complete Walkthrough",
      "url": "https://realpython.com/async-io-python/",
      "content": "Full extracted article text via Trafilatura (~2000 chars max)...",
      "snippet": "Original search engine snippet (150-300 chars)...",
      "source": "realpython.com",
      "published_date": "2023-05-15"
    }
  ],
  "total_results": 10,
  "search_time_ms": 2340,
  "sources_summary": "## Sources\n- [Async IO in Python](https://realpython.com/async-io-python/)\n- ..."
}

Key Fields for Tatlock

Field	Usage
results[].content	Full extracted text - use this for LLM context
results[].snippet	Fallback if content extraction failed
sources_summary	Pre-formatted markdown for citations

Error Handling

HTTP Code	Meaning	Action
400	Invalid query	Check query length/format
502	SearXNG unavailable	Retry with backoff
504	Search timeout	Retry or reduce limit
500	Internal error	Log and notify

Example Usage (Python)

import httpx

async def search_web(query: str, limit: int = 10) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{BASE_URL}/rag/search",
            headers={"Authorization": f"Bearer {API_KEY}"},
            json={
                "query": query,
                "search_type": "web",
                "limit": limit,
                "user": "tatlock-librarian"
            },
            timeout=30.0
        )
        response.raise_for_status()
        return response.json()

# Usage
results = await search_web("machine learning transformers")
for r in results["results"]:
    # Prefer full content, fall back to snippet
    text = r["content"] or r["snippet"]
    print(f"{r['title']}: {len(text)} chars")

---

2. Content Extraction Endpoint

Use case: Librarian has a specific URL and needs to read its content.

Single URL Extraction

POST /content/extract

Request

{
  "url": "https://example.com/article",
  "include_metadata": true,
  "max_length": 2000
}

Response

{
  "result": {
    "url": "https://example.com/article",
    "title": "Article Title",
    "content": "Extracted main text content...",
    "author": "John Doe",
    "date": "2024-01-15",
    "language": "en",
    "success": true,
    "error": null
  },
  "extraction_time_ms": 1250
}

Batch URL Extraction

POST /content/extract/batch

Request

{
  "urls": [
    "https://example.com/article1",
    "https://example.com/article2",
    "https://example.com/article3"
  ],
  "include_metadata": true,
  "max_length": 2000
}

Response

{
  "results": [
    {
      "url": "https://example.com/article1",
      "title": "Article 1",
      "content": "Extracted content...",
      "success": true,
      "error": null
    },
    {
      "url": "https://example.com/article2",
      "title": null,
      "content": "",
      "success": false,
      "error": "Connection timeout"
    }
  ],
  "total_urls": 3,
  "successful": 2,
  "failed": 1,
  "extraction_time_ms": 3500
}

---

3. Error Pattern: Soft Failures

Important: Content extraction uses a soft failure pattern - individual URL failures do NOT throw HTTP errors.

Why Soft Failures?

When extracting content from multiple URLs (batch) or even single URLs:

• Some sites block bots
• Some URLs are temporarily down
• Some pages have no extractable content

Instead of failing the entire request, we return:

• success: true/false per result
• error: "reason" when failed
• Empty content: "" on failure

Handling Soft Failures

async def extract_with_fallback(url: str) -> str:
    response = await client.post(
        f"{BASE_URL}/content/extract",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={"url": url}
    )
    response.raise_for_status()  # Only throws on 4xx/5xx

    data = response.json()
    result = data["result"]

    if result["success"]:
        return result["content"]
    else:
        # Log the failure, return empty or handle gracefully
        logger.warning(f"Extraction failed for {url}: {result['error']}")
        return ""  # Or raise, or use cached version, etc.

Batch Processing Example

async def extract_batch_with_stats(urls: list[str]) -> dict:
    response = await client.post(
        f"{BASE_URL}/content/extract/batch",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={"urls": urls, "max_length": 3000}
    )
    response.raise_for_status()

    data = response.json()

    # Separate successful and failed
    successful = [r for r in data["results"] if r["success"]]
    failed = [r for r in data["results"] if not r["success"]]

    if failed:
        logger.warning(f"{len(failed)} URLs failed extraction:")
        for f in failed:
            logger.warning(f"  {f['url']}: {f['error']}")

    return {
        "contents": {r["url"]: r["content"] for r in successful},
        "failed_urls": [f["url"] for f in failed],
        "success_rate": data["successful"] / data["total_urls"]
    }

---

4. Recommended Patterns for Tatlock

Research Flow

async def librarian_research(topic: str) -> dict:
    """
    Full research flow: search + extract additional context.
    """
    # 1. Search for relevant pages
    search_results = await search_web(topic, limit=10)

    # 2. RAG search already includes extracted content
    #    Only extract more if you need deeper content

    # 3. Build context for LLM
    context_parts = []
    for r in search_results["results"]:
        content = r["content"] or r["snippet"]
        if content:
            context_parts.append(f"## {r['title']}\nSource: {r['url']}\n\n{content}")

    return {
        "context": "\n\n---\n\n".join(context_parts),
        "sources": search_results["sources_summary"],
        "result_count": search_results["total_results"]
    }

Reading a Specific Page

async def librarian_read_page(url: str) -> str:
    """
    Read a specific URL the user provided.
    """
    response = await client.post(
        f"{BASE_URL}/content/extract",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={"url": url, "max_length": 5000}  # Longer for deep reads
    )
    response.raise_for_status()

    result = response.json()["result"]

    if not result["success"]:
        raise ValueError(f"Could not read page: {result['error']}")

    # Format for LLM
    header = f"# {result['title'] or 'Untitled'}\n"
    if result["author"]:
        header += f"Author: {result['author']}\n"
    if result["date"]:
        header += f"Date: {result['date']}\n"

    return header + "\n" + result["content"]

---

5. Rate Limits & Best Practices

Recommendation	Reason
Use limit: 5-10 for searches	More results = longer extraction time
Batch URLs when possible	More efficient than sequential calls
Max 20 URLs per batch	Server limit
Set reasonable timeouts (30s)	Content extraction can be slow
Cache results client-side	Same URL rarely changes content
Use user parameter	Helps with debugging and rate limiting

---

6. Quick Reference

Endpoint	Method	Use Case
/rag/search	POST	Search web + get extracted content
/content/extract	POST	Read a single URL
/content/extract/batch	POST	Read multiple URLs
/health	GET	Check service status