Structured vs. Unstructured Retrieval: Documents, APIs, and Databases

Not all data lives in the same format. Documents are unstructured text. Databases are structured tables. APIs return structured JSON. Your agent needs to know which retrieval strategy to use for which data type—and how to combine results across formats.

The Retrieval Spectrum

Fully Structured                    Fully Unstructured
(SQL, Tables)                       (Documents, Images)
      │                                    │
      ▼                                    ▼
┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐
│   SQL    │  │  Graph   │  │  Vector  │  │  Full    │
│  Query   │  │  Query   │  │  Search  │  │  Text    │
│          │  │          │  │          │  │  Search  │
└──────────┘  └──────────┘  └──────────┘  └──────────┘
   Exact         Relationship  Semantic     Keyword
   Matches       Traversal     Matching     Matching

Structured Retrieval

SQL-Based Retrieval

from sqlalchemy import create_engine, text
import pandas as pd

class SQLRetriever:
    def __init__(self, connection_string: str):
        self.engine = create_engine(connection_string)
    
    def retrieve(self, query: str, schema_info: str) -> str:
        """Convert natural language to SQL and execute."""
        
        sql_prompt = f"""Given the database schema:
        {schema_info}
        
        Convert this question to SQL: "{query}"
        
        Return ONLY the SQL query, no explanation:"""
        
        sql_query = llm.invoke([HumanMessage(content=sql_prompt)]).content
        sql_query = sql_query.strip().strip("```sql").strip("```").strip()
        
        try:
            df = pd.read_sql(text(sql_query), self.engine)
            return df.to_markdown(index=False)
        except Exception as e:
            return f"SQL execution error: {e}"
    
    def get_schema_info(self) -> str:
        """Extract schema for LLM context."""
        with self.engine.connect() as conn:
            tables = conn.execute(text(
                "SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'"
            )).fetchall()
            
            schema_parts = []
            for (table,) in tables:
                columns = conn.execute(text(
                    f"SELECT column_name, data_type FROM information_schema.columns "
                    f"WHERE table_name = '{table}'"
                )).fetchall()
                
                col_str = ", ".join([f"{name} ({dtype})" for name, dtype in columns])
                schema_parts.append(f"TABLE {table}: {col_str}")
            
            return "\n".join(schema_parts)

Graph-Based Retrieval

from neo4j import GraphDatabase

class GraphRetriever:
    def __init__(self, uri: str, user: str, password: str):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))
    
    def retrieve(self, query: str) -> str:
        """Convert natural language to Cypher and query the graph."""
        
        schema = self.get_schema()
        
        cypher_prompt = f"""Given this graph schema:
        {schema}
        
        Convert to Cypher: "{query}"
        
        Return ONLY the Cypher query:"""
        
        cypher = llm.invoke([HumanMessage(content=cypher_prompt)]).content
        
        try:
            with self.driver.session() as session:
                result = session.run(cypher)
                records = [dict(r) for r in result]
                return str(records)
        except Exception as e:
            return f"Graph query error: {e}"
    
    def get_schema(self) -> str:
        with self.driver.session() as session:
            result = session.run("CALL db.schema.visualization()")
            return str(result.single())

Unstructured Retrieval

Vector Search

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

class VectorRetriever:
    def __init__(self, documents_path: str):
        documents = SimpleDirectoryReader(documents_path).load_data()
        self.index = VectorStoreIndex.from_documents(documents)
    
    def retrieve(self, query: str, top_k: int = 5) -> str:
        query_engine = self.index.as_query_engine(
            similarity_top_k=top_k,
            response_mode="compact"
        )
        response = query_engine.query(query)
        return str(response)

Hybrid Search

from llama_index.core.retrievers import VectorIndexRetriever, KeywordTableRetriever
from llama_index.core.schema import QueryBundle

class HybridRetriever:
    def __init__(self, vector_index, keyword_index, weights: tuple = (0.7, 0.3)):
        self.vector_retriever = VectorIndexRetriever(index=vector_index, similarity_top_k=5)
        self.keyword_retriever = KeywordTableRetriever(index=keyword_index, similarity_top_k=5)
        self.weights = weights
    
    def retrieve(self, query: str) -> list:
        # Get results from both
        vector_results = self.vector_retriever.retrieve(query)
        keyword_results = self.keyword_retriever.retrieve(query)
        
        # Reciprocal Rank Fusion
        fused_scores = {}
        for rank, node in enumerate(vector_results):
            fused_scores[node.node.node_id] = fused_scores.get(node.node.node_id, 0) + \
                self.weights[0] / (rank + 60)
        
        for rank, node in enumerate(keyword_results):
            fused_scores[node.node.node_id] = fused_scores.get(node.node.node_id, 0) + \
                self.weights[1] / (rank + 60)
        
        # Sort by fused score
        sorted_ids = sorted(fused_scores.keys(), key=lambda x: fused_scores[x], reverse=True)
        
        all_nodes = {n.node.node_id: n for n in vector_results + keyword_results}
        return [all_nodes[nid] for nid in sorted_ids[:10]]

Adaptive Retrieval Router

from enum import Enum

class DataType(Enum):
    STRUCTURED = "structured"
    SEMI_STRUCTURED = "semi_structured"
    UNSTRUCTURED = "unstructured"
    MIXED = "mixed"

class AdaptiveRetriever:
    def __init__(self, sql_retriever, vector_retriever, graph_retriever):
        self.retrievers = {
            "sql": sql_retriever,
            "vector": vector_retriever,
            "graph": graph_retriever,
        }
    
    def classify_query(self, query: str) -> DataType:
        """Determine what type of data the query needs."""
        
        prompt = f"""Classify this query by what data type it needs:
        
        Query: "{query}"
        
        Options:
        - STRUCTURED: Needs data from databases, tables, specific records
        - SEMI_STRUCTURED: Needs data from JSON, APIs, config files
        - UNSTRUCTURED: Needs understanding from documents, text
        - MIXED: Needs both structured and unstructured data
        
        Classification:"""
        
        response = llm.invoke([HumanMessage(content=prompt)]).content.strip()
        
        try:
            return DataType(response.lower())
        except ValueError:
            return DataType.MIXED
    
    def retrieve(self, query: str) -> dict:
        data_type = self.classify_query(query)
        
        if data_type == DataType.STRUCTURED:
            return {"sql": self.retrievers["sql"].retrieve(query)}
        
        elif data_type == DataType.UNSTRUCTURED:
            return {"vector": self.retrievers["vector"].retrieve(query)}
        
        elif data_type == DataType.SEMI_STRUCTURED:
            return {"vector": self.retrievers["vector"].retrieve(query)}
        
        else:  # MIXED
            results = {}
            for name, retriever in self.retrievers.items():
                results[name] = retriever.retrieve(query)
            return results

Cross-Format Answer Synthesis

class MultiSourceSynthesizer:
    def __init__(self, llm):
        self.llm = llm
    
    def synthesize(self, query: str, results_by_source: dict[str, str]) -> str:
        """Combine results from multiple retrieval sources into a coherent answer."""
        
        source_descriptions = []
        for source, result in results_by_source.items():
            source_descriptions.append(f"=== Source: {source} ===\n{result}")
        
        combined_context = "\n\n".join(source_descriptions)
        
        prompt = f"""Answer the question using information from multiple sources below.
        
        If sources conflict, mention the discrepancy.
        If a source doesn't have relevant information, ignore it.
        
        Question: {query}
        
        Sources:
        {combined_context}
        
        Answer:"""
        
        response = self.llm.invoke([HumanMessage(content=prompt)])
        return response.content

When to Use Each Strategy

RETRIEVAL_DECISIONS = {
    "exact_lookup": {
        "strategy": "sql",
        "when": "User asks for specific records, counts, aggregations",
        "example": "How many users signed up last month?"
    },
    "conceptual_search": {
        "strategy": "vector",
        "when": "User asks about ideas, concepts, or similar documents",
        "example": "Find articles about machine learning best practices"
    },
    "relationship_query": {
        "strategy": "graph",
        "when": "User asks about connections between entities",
        "example": "Which teams work with the data engineering team?"
    },
    "multi_source": {
        "strategy": "hybrid",
        "when": "Answer requires both structured and unstructured data",
        "example": "What do our docs say about the API that has the most errors?"
    },
}

Evaluation Framework

def evaluate_retrieval_strategy(strategy_name: str, test_cases: list) -> dict:
    results = {"precision": [], "recall": [], "latency": [], "cost": []}
    
    for case in test_cases:
        start = time.time()
        
        retrieved = strategy.retrieve(case["query"])
        latency = time.time() - start
        
        # Precision: what fraction of retrieved items are relevant?
        relevant_retrieved = count_relevant(retrieved, case["relevant_ids"])
        precision = relevant_retrieved / max(len(retrieved), 1)
        
        # Recall: what fraction of relevant items were retrieved?
        recall = relevant_retrieved / max(len(case["relevant_ids"]), 1)
        
        results["precision"].append(precision)
        results["recall"].append(recall)
        results["latency"].append(latency)
    
    return {
        "avg_precision": sum(results["precision"]) / len(results["precision"]),
        "avg_recall": sum(results["recall"]) / len(results["recall"]),
        "avg_latency": sum(results["latency"]) / len(results["latency"]),
        "f1": 2 * (sum(results["precision"]) / len(results["precision"])) * (sum(results["recall"]) / len(results["recall"])) / 
              max((sum(results["precision"]) / len(results["precision"])) + (sum(results["recall"]) / len(results["recall"])), 0.001),
    }

The best retrieval strategy depends on your data, not ideology. Most production systems need a combination—SQL for facts, vectors for concepts, graphs for relationships. Build an adaptive router and let the agent decide.