Real-Time Scenario: Semantic Search for Research Papers
Let's explore a practical implementation of vector indexing for semantic search of research papers using Python, the arXiv API, and CrewAI.
Understanding the Data Structure
Our example uses a YAML file structured with research paper metadata. Here's a more comprehensive structure of the YAML with research pipeline configuration:
research_pipeline:
arxiv_query: "cat:cs.AI AND submittedDate:[20230501 TO 20240501]"
embedding_model: "sentence-transformers/all-mpnet-base-v2"
vector_index:
type: "HNSW"
dimensions: 768
ef_construction: 200
M: 16
papers:
- arxiv_id: "2405.12345v1"
title: "Advancements in Transformer Architectures"
authors: ["Smith, J.", "Doe, A."]
- arxiv_id: "2405.12346v1"
title: "Efficient Neural Network Compression Techniques"
# Original structure is also supported
- category: Recent Papers
contents:
- date: '2025-05-15'
link: https://arxiv.org/abs/2505.10185
title: 'The CoT Encyclopedia: Analyzing, Predicting, and Controlling how a Reasoning Model will Think'
- date: '2025-05-13'
link: https://arxiv.org/abs/2505.08775
title: 'HealthBench: Evaluating Large Language Models Towards Improved Human Health'
This hierarchical structure enables easy maintenance and parameter tuning. It includes pipeline configuration details such as the arXiv search query, embedding model selection, vector index configuration parameters, and manual paper entries.
Step 1: Loading Paper Metadata from YAML
First, we need to load and parse the YAML file containing our research paper metadata:
import yaml
def load_papers(yaml_path):
"""Load research papers from YAML file"""
with open(yaml_path, 'r') as file:
data = yaml.safe_load(file)
papers = []
# Check for research pipeline configuration
if 'research_pipeline' in data:
config = data['research_pipeline']
# Process papers directly listed in configuration
if 'papers' in config:
for paper in config['papers']:
papers.append({
'arxiv_id': paper['arxiv_id'],
'title': paper['title'],
'authors': paper.get('authors', []),
'link': f"https://arxiv.org/abs/{paper['arxiv_id']}"
})
# Extract other configuration details
embedding_model = config.get('embedding_model', 'all-MiniLM-L6-v2')
vector_index_config = config.get('vector_index', {})
return papers, embedding_model, vector_index_config
# Process original format (for backward compatibility)
for category in data:
for paper in category['contents']:
papers.append({
'title': paper['title'],
'link': paper['link'],
'date': paper['date'],
'category': category['category']
})
return papers, 'all-MiniLM-L6-v2', {}
# Load papers from YAML
papers, embedding_model, vector_index_config = load_papers('_data/research_papers.yml')
print(f"Loaded {len(papers)} papers from YAML")
print(f"Using embedding model: {embedding_model}")
print(f"Vector index configuration: {vector_index_config}")
Step 2: Fetching Paper Abstracts from arXiv
Next, we extract arXiv IDs from the links and fetch the corresponding abstracts, implementing batch processing and error handling:
import arxiv
import re
import time
from typing import List, Dict, Any
def extract_arxiv_id(arxiv_url):
"""Extract arXiv ID from URL"""
match = re.search(r'abs/([0-9]+\.[0-9]+v?[0-9]*)', arxiv_url)
if match:
return match.group(1)
return None
def fetch_paper_abstracts(papers: List[Dict[str, Any]], arxiv_query=None):
"""Fetch abstracts for papers and optionally search for additional papers"""
client = arxiv.Client(page_size=100, delay_seconds=3)
results = []
# Process existing papers
for paper in papers:
# Check if paper has arxiv_id directly
arxiv_id = paper.get('arxiv_id')
# If no direct ID, try to extract from link
if not arxiv_id and 'link' in paper:
arxiv_id = extract_arxiv_id(paper['link'])
if arxiv_id:
try:
search = arxiv.Search(id_list=[arxiv_id])
paper_results = list(client.results(search))
if paper_results:
paper_info = paper_results[0]
paper['abstract'] = paper_info.summary
paper['authors'] = [author.name for author in paper_info.authors]
paper['categories'] = paper_info.categories
paper['published'] = paper_info.published.strftime('%Y-%m-%d')
print(f"Fetched abstract for: {paper['title']}")
else:
paper['abstract'] = ""
print(f"No results found for: {arxiv_id}")
except Exception as e:
print(f"Failed to fetch abstract for {arxiv_id}: {e}")
paper['abstract'] = ""
# Implement exponential backoff
time.sleep(5)
else:
paper['abstract'] = ""
results.append(paper)
# If provided with an arxiv query, search for additional papers
if arxiv_query:
try:
print(f"Searching arXiv with query: {arxiv_query}")
search = arxiv.Search(
query=arxiv_query,
max_results=100,
sort_by=arxiv.SortCriterion.SubmittedDate
)
for result in client.results(search):
arxiv_id = result.entry_id.split('/')[-1]
# Check if we already have this paper
if not any(p.get('arxiv_id') == arxiv_id for p in results):
new_paper = {
'arxiv_id': arxiv_id,
'title': result.title,
'abstract': result.summary,
'authors': [author.name for author in result.authors],
'categories': result.categories,
'published': result.published.strftime('%Y-%m-%d'),
'link': result.entry_id
}
results.append(new_paper)
print(f"Added new paper from search: {result.title}")
except Exception as e:
print(f"Error searching arXiv: {e}")
print(f"Total papers processed: {len(results)}")
return results
# Fetch abstracts, possibly using arxiv_query from config
papers_with_abstracts = fetch_paper_abstracts(
papers,
arxiv_query=vector_index_config.get('arxiv_query')
)
Step 3: Generating Vector Embeddings
Now we convert paper abstracts into vector embeddings using a pre-trained model, with support for different embedding models:
from sentence_transformers import SentenceTransformer
import numpy as np
def generate_embeddings(papers, model_name="all-MiniLM-L6-v2"):
"""Generate embeddings for paper abstracts"""
# Load pre-trained model
model = SentenceTransformer(model_name)
# Get embedding dimensions for logging
dimensions = model.get_sentence_embedding_dimension()
print(f"Using {model_name} with {dimensions} dimensions")
# Process papers in batches for efficiency
batch_size = 32
embeddings = []
for i in range(0, len(papers), batch_size):
batch = papers[i:i + batch_size]
# Combine title and abstract for better semantic representation
texts = [f"{paper['title']} {paper.get('abstract', '')}" for paper in batch]
batch_embeddings = model.encode(texts, show_progress_bar=True)
embeddings.extend(batch_embeddings)
print(f"Processed batch {i//batch_size + 1}/{(len(papers) + batch_size - 1)//batch_size}")
# Add embeddings to papers
for paper, embedding in zip(papers, embeddings):
paper['embedding'] = embedding
print(f"Generated embeddings for {len(papers)} papers")
return papers, dimensions
# Generate embeddings using model from config
papers_with_embeddings, embedding_dim = generate_embeddings(
papers_with_abstracts,
model_name=embedding_model
)
Step 4: Storing Vectors in a Database
We'll use Qdrant, a vector database, to store and index our embeddings, implementing HNSW index configuration:
from qdrant_client import QdrantClient
from qdrant_client.http import models
def setup_vector_db(dimensions=384, config=None):
"""Initialize vector database with HNSW index configuration"""
client = QdrantClient("localhost", port=6333)
# Extract HNSW parameters from config or use defaults
if config and 'vector_index' in config:
index_config = config['vector_index']
ef_construction = index_config.get('ef_construction', 128)
m = index_config.get('M', 16)
else:
ef_construction = 128 # Default
m = 16 # Default
# Create collection with HNSW index configuration
try:
client.create_collection(
collection_name="research_papers",
vectors_config=models.VectorParams(
size=dimensions,
distance=models.Distance.COSINE,
# HNSW index configuration
hnsw_config=models.HnswConfigDiff(
m=m, # Number of connections per layer
ef_construct=ef_construction, # Construction-time quality parameter
)
),
# Optional sparse vectors configuration for hybrid search
optimizers_config=models.OptimizersConfigDiff(
memmap_threshold=20000 # Enable memory mapping for collections > 20K vectors
)
)
print(f"Created new collection 'research_papers' with HNSW index (ef={ef_construction}, m={m})")
except Exception as e:
print(f"Collection might already exist: {e}")
return client
def store_embeddings(client, papers):
"""Store paper embeddings in vector database with optimization for large datasets"""
# Batch size for optimal performance
batch_size = 100
total_points = 0
for i in range(0, len(papers), batch_size):
batch = papers[i:i + batch_size]
points = []
for j, paper in enumerate(batch):
# Create point with embedding and metadata
point_id = i + j
# Include rich metadata for filtering and post-processing
payload = {
'title': paper['title'],
'link': paper.get('link', ''),
'date': paper.get('date', paper.get('published', '')),
'category': paper.get('category', '')
}
# Add authors if available
if 'authors' in paper:
payload['authors'] = paper['authors']
# Add arxiv categories if available
if 'categories' in paper:
payload['arxiv_categories'] = paper['categories']
# Include arxiv_id for direct reference
if 'arxiv_id' in paper:
payload['arxiv_id'] = paper['arxiv_id']
# Create the point structure
point = models.PointStruct(
id=point_id,
vector=paper['embedding'].tolist(),
payload=payload
)
points.append(point)
# Store batch
client.upsert(
collection_name="research_papers",
points=points,
wait=True # Ensure points are indexed before proceeding
)
total_points += len(points)
print(f"Stored batch {i//batch_size + 1}/{(len(papers) + batch_size - 1)//batch_size} ({total_points}/{len(papers)} total)")
print(f"Stored {total_points} papers in vector database")
# Optimize collection if needed
if total_points > 10000:
print("Running optimization for large collection...")
client.update_collection(
collection_name="research_papers",
optimizers_config=models.OptimizersConfigDiff(
indexing_threshold=0 # Force reindexing
)
)
# Setup database with dimensions from embedding
db_client = setup_vector_db(dimensions=embedding_dim, config=vector_index_config)
store_embeddings(db_client, papers_with_embeddings)
Step 5: Semantic Search with Vector Indexing
Now we can perform semantic search to find papers related to specific topics, with advanced query processing and filtering:
def semantic_search(client, model, query, limit=5, filter_categories=None, min_score=0.6):
"""Perform semantic search for research papers with advanced options"""
# Convert query to embedding
query_vector = model.encode(query).tolist()
# Prepare filters if needed
filter_query = None
if filter_categories:
filter_query = models.Filter(
must=[
models.FieldCondition(
key="arxiv_categories",
match=models.MatchAny(any=filter_categories)
)
]
)
# Configure search parameters for optimal quality
search_params = models.SearchParams(
hnsw_ef=max(128, limit * 4), # Dynamic ef adjustment based on result count
exact=False # Use approximate search for speed
)
# Search for similar papers
search_results = client.search(
collection_name="research_papers",
query_vector=query_vector,
limit=limit * 2, # Fetch more than needed for score filtering
filter=filter_query,
search_params=search_params,
with_payload=True,
score_threshold=min_score # Minimum similarity score
)
# Process and refine results
results = []
for result in search_results[:limit]:
result_data = {
'title': result.payload['title'],
'link': result.payload['link'],
'similarity': round(result.score, 4),
'date': result.payload.get('date', '')
}
# Add category if available
if 'category' in result.payload:
result_data['category'] = result.payload['category']
# Add authors if available
if 'authors' in result.payload:
result_data['authors'] = result.payload['authors']
# Add arXiv ID if available
if 'arxiv_id' in result.payload:
result_data['arxiv_id'] = result.payload['arxiv_id']
results.append(result_data)
return results
# Example advanced search
model = SentenceTransformer(embedding_model)
results = semantic_search(
db_client,
model,
"Latest developments in AGI and reasoning",
limit=5,
filter_categories=["cs.AI", "cs.LG"], # Only AI and ML papers
min_score=0.70 # Higher relevance threshold
)
# Display results
print("\nSearch Results:")
for i, result in enumerate(results, 1):
print(f"{i}. {result['title']} (Similarity: {result['similarity']})")
print(f" Link: {result['link']}")
if 'authors' in result:
print(f" Authors: {', '.join(result['authors'][:3])}" +
(f" and {len(result['authors'])-3} more" if len(result['authors']) > 3 else ""))
if 'category' in result:
print(f" Category: {result['category']}")
print()
Using CrewAI for Coordinated Workflow
CrewAI provides a framework for orchestrating multiple agents to work together on complex tasks. Here's how we can implement our semantic search system using CrewAI:
from crewai import Agent, Task, Crew
# Define specialized agents
data_loader_agent = Agent(
name="Data Loader",
role="Loads and parses YAML data",
goal="Efficiently load and parse research paper data from YAML file",
backstory="Expert in data parsing and validation",
verbose=True
)
arxiv_fetcher_agent = Agent(
name="ArXiv Fetcher",
role="Fetches paper abstracts from arXiv",
goal="Retrieve paper abstracts using arXiv IDs",
backstory="Specialized in working with academic APIs",
verbose=True
)
embedding_agent = Agent(
name="Embedding Generator",
role="Generates embeddings from text",
goal="Convert paper abstracts into vector embeddings",
backstory="Expert in NLP and vector representations",
verbose=True
)
db_agent = Agent(
name="Database Manager",
role="Manages vector database operations",
goal="Store and manage vector embeddings efficiently",
backstory="Database expert specialized in vector stores",
verbose=True
)
search_agent = Agent(
name="Search Expert",
role="Performs semantic searches",
goal="Find relevant papers based on semantic queries",
backstory="Search optimization specialist",
verbose=True
)
# Create tasks for the workflow
task1 = Task(
description="Load research papers from YAML file",
expected_output="List of papers with metadata",
agent=data_loader_agent
)
task2 = Task(
description="Fetch abstracts for all papers from arXiv",
expected_output="Papers enriched with abstracts",
agent=arxiv_fetcher_agent
)
task3 = Task(
description="Generate vector embeddings for papers",
expected_output="Papers with vector embeddings",
agent=embedding_agent
)
task4 = Task(
description="Store embeddings in vector database",
expected_output="Confirmation of successful storage",
agent=db_agent
)
task5 = Task(
description="Perform semantic search for papers on AGI reasoning",
expected_output="Ranked list of relevant papers",
agent=search_agent
)
# Create crew with agents and tasks
crew = Crew(
agents=[data_loader_agent, arxiv_fetcher_agent, embedding_agent, db_agent, search_agent],
tasks=[task1, task2, task3, task4, task5],
verbose=2
)
# Execute the workflow
result = crew.kickoff()
print(result)
Detailed Agent Operations and Tools
Let's explore in detail how each CrewAI agent operates, what tools they use, and what output they produce:
| Agent |
Tools & Libraries |
Example Output |
| Data Loader |
- PyYAML for parsing YAML
- Custom validation functions
- Error handling utilities
|
{
'status': 'success',
'papers_loaded': 245,
'categories': ['Recent Papers', 'Recommended Papers'],
'data': [
{'title': 'The CoT Encyclopedia', 'link': '...', 'date': '2025-05-15', 'category': 'Recent Papers'},
{'title': 'HealthBench', 'link': '...', 'date': '2025-05-13', 'category': 'Recent Papers'},
...
]
}
|
| ArXiv Fetcher |
- arxiv Python library
- Regular expressions for ID extraction
- Rate limiting management
- Error handling for API failures
|
{
'status': 'completed',
'papers_processed': 245,
'abstracts_fetched': 237,
'failed': 8,
'sample_abstract': 'In this paper, we present a comprehensive analysis of chain-of-thought reasoning...',
'data': [
{'title': '...', 'abstract': '...', 'link': '...', ...},
...
]
}
|
| Embedding Generator |
- SentenceTransformers library
- Hugging Face transformers
- Text preprocessing utilities
- Batch processing for efficiency
|
{
'status': 'success',
'embedding_model': 'all-MiniLM-L6-v2',
'embedding_dimension': 384,
'papers_embedded': 237,
'sample_embedding': [0.0257, 0.0153, -0.0621, ...],
'data': [
{'title': '...', 'abstract': '...', 'embedding': [...], ...},
...
]
}
|
| Database Manager |
- Qdrant client library
- Connection management tools
- Batch insertion optimization
- Index configuration utilities
|
{
'status': 'completed',
'database': 'Qdrant',
'collection': 'research_papers',
'points_inserted': 237,
'index_type': 'HNSW',
'configuration': {
'distance': 'cosine',
'ef_construction': 128,
'M': 16
},
'storage_details': {
'disk_usage': '12.4 MB',
'vector_dimension': 384
}
}
|
| Search Expert |
- Qdrant search API
- Query preprocessing tools
- Result ranking algorithms
- Natural language interpretation
|
{
'query': 'Latest developments in AGI and reasoning',
'results': [
{
'title': 'The CoT Encyclopedia: Analyzing, Predicting, and Controlling how a Reasoning Model will Think',
'similarity': 0.87,
'link': 'https://arxiv.org/abs/2505.10185',
'category': 'Recent Papers'
},
{
'title': 'Absolute Zero: Reinforced Self-play Reasoning with Zero Data',
'similarity': 0.82,
'link': 'https://arxiv.org/abs/2505.03335',
'category': 'Recent Papers'
},
{
'title': 'HyperTree Planning: Enhancing LLM Reasoning via Hierarchical Thinking',
'similarity': 0.79,
'link': 'https://arxiv.org/abs/2505.02322',
'category': 'Recent Papers'
},
...
]
}
|
Agent Interaction and Workflow
CrewAI orchestrates the interactions between agents. Here's how information flows through the system:
-
Task Sequence & Dependencies:
Tasks are executed in order, with each agent receiving the output from the previous task:
Data Loader → ArXiv Fetcher → Embedding Generator → Database Manager → Search Expert
-
Data Transformation:
Each agent enriches or transforms the data:
- Raw YAML → Paper metadata → Papers with abstracts → Papers with embeddings → Stored vectors → Search results
-
Error Handling:
Agents implement custom error handling. For example, if the ArXiv Fetcher can't retrieve an abstract, it will:
- Retry with exponential backoff
- Log the failure for later review
- Continue with empty abstract rather than failing the entire workflow
-
Performance Optimization:
Agents can optimize their operations:
- Batch processing papers in chunks
- Caching intermediate results
- Parallel processing when possible
Execution Output
Here's what the process looks like when executed:
CrewAI Execution Flow
The CrewAI framework provides detailed execution logs showing:
- Task assignment and completion status
- Agent thinking processes and decision-making
- Data flow between agents
- Error handling and recovery
- Performance metrics and timing
Advanced CrewAI Features for Vector Search
The CrewAI framework enables several advanced features that enhance the vector indexing workflow:
-
Tool Integration: Agents can be equipped with specialized tools like:
- Custom paper filtering functions
- Analysis tools to evaluate embedding quality
- Visualization utilities to explore the vector space
-
Agent Memory: Agents maintain memory of previous operations, enabling:
- Caching frequently used embeddings
- Learning from previous search patterns
- Adapting to specific research domains
-
Custom Callbacks: The workflow can implement callbacks for:
- Logging detailed performance metrics
- Real-time monitoring of embedding quality
- Integration with external systems
-
Human-in-the-Loop: CrewAI supports human augmentation:
- Human validation of key abstracts
- Expert refinement of search queries
- Feedback mechanisms to improve vector search quality
Performance Optimization Techniques
To ensure optimal performance when scaling to large research paper collections, consider these advanced optimization techniques:
| Technique |
Implementation |
Benefits |
| Index Training |
def train_index(embeddings, config):
# Sample subset for training
train_samples = np.random.choice(
len(embeddings),
int(0.25*len(embeddings)),
replace=False
)
# Train index on sample
index.train(embeddings[train_samples])
|
- Ensures representative cluster initialization
- Improves vector distribution
- Reduces training time on large datasets
|
| Dynamic efSearch |
# Adjust efSearch parameter based on query
index.hnsw.efSearch = 50 + 2*k
|
- Higher values increase search accuracy
- Lower values improve search speed
- Dynamic adjustment balances needs
|
| Batch Processing |
# Process in optimal batches
batch_size = 100
for i in range(0, len(data), batch_size):
batch = data[i:i + batch_size]
# Process batch
|
- Reduces memory pressure
- Improves throughput
- Enables progress tracking
|
| Memory Mapping |
# Enable memory mapping
client.update_collection(
collection_name="papers",
optimizers_config=models.OptimizersConfigDiff(
memmap_threshold=20000
)
)
|
- Handles indices larger than RAM
- Reduces memory usage
- Enables persistent storage
|
| Hybrid Search |
# Combine vector search with text filtering
results = client.search(
collection_name="papers",
query_vector=vector,
filter=models.Filter(
must=[models.FieldCondition(
key="keywords",
match=models.MatchText(text="neural")
)]
)
)
|
- Combines semantic and keyword search
- Improves precision
- Enables complex queries
|
| Evaluation Metrics |
def evaluate_search(results, ground_truth):
relevant = set(r['id'] for r in results)
relevant_truth = relevant.intersection(ground_truth)
precision = len(relevant_truth) / len(results)
recall = len(relevant_truth) / len(ground_truth)
f1 = 2 * (precision * recall) / (precision + recall)
return {'precision': precision, 'recall': recall, 'f1': f1}
|
- Quantifies search quality
- Guides parameter optimization
- Enables objective comparisons
|
Typical Performance Metrics
When properly configured, vector indexing with HNSW can achieve impressive performance even at scale:
- Recall@10: 0.92 ± 0.05 (percentage of relevant documents retrieved)
- Query latency:
- 10K papers: ~10ms
- 100K papers: ~25ms
- 1M papers: ~80ms
- 5M papers: ~200ms
- Index build time: ~30 minutes for 1M papers (8-core CPU)
- Memory usage: ~800 bytes per vector + payload overhead
Advantages of Vector Indexing for Research Paper Search
- Semantic Understanding: Find papers that are conceptually related even if they don't share exact keywords
- Relevance Over Recency: Discover important papers regardless of publication date
- Cross-Domain Insights: Connect research across different domains through conceptual similarity
- Natural Language Queries: Search using conversational language rather than specific keywords
- Scalability: Efficiently search through thousands or millions of papers
Comparing Semantic Search and CrewAI Implementations
When examining the two search implementations demonstrated in this section, we can observe significant differences between the traditional semantic search and the CrewAI-based approach:
| Feature |
Traditional Semantic Search |
CrewAI Implementation |
| Architecture |
Script-based approach with sequential function calls |
Agent-based with distributed tasks and workflow orchestration |
| Output Format |
Console logs showing processing steps and final results |
Detailed execution flow with visual task tree and status indicators |
| Workflow Detail |
Technical processing steps like "Processing query," "Converting query to embedding vector" |
Complete pipeline orchestration with specialized agents for each task |
| Task Separation |
Functions with mixed responsibilities |
Clear separation of concerns with specialized agents |
| Result Depth |
Basic results with arxiv links and similarity scores |
Comprehensive paper information with detailed metadata and author information |
| Papers Returned |
Research papers with specific arXiv identifiers |
Classic ML/AI textbooks and surveys with comprehensive details |
| Workflow Management |
Implicit process flow |
Explicit task completion statuses and agent state visualization |
| Extensibility |
Requires rewriting functions for new capabilities |
Simply add new agents with specialized capabilities |
| Error Handling |
Basic try/except patterns |
Agent-specific error handling with retry mechanisms |
The CrewAI implementation demonstrates a more modern, extensible architecture that's better suited for complex workflows with multiple processing stages. It provides clearer visibility into the execution process and naturally separates concerns into specialized agents. This architecture scales more effectively for complex tasks and allows for easier addition of new capabilities such as human-in-the-loop validation, customized search strategies, or integration with other systems.
How This Example Demonstrates Vector Indexing Principles
This implementation showcases several key vector indexing concepts:
- Embedding Generation: Converting text data (paper abstracts) into high-dimensional vectors
- Vector Storage: Using specialized databases optimized for vector data
- Similarity Search: Finding papers based on semantic similarity rather than keyword matching
- Metadata Association: Keeping original metadata alongside vectors for retrieval
- Distributed Processing: Using a multi-agent system to handle different parts of the workflow
By implementing vector indexing for research papers, you create a powerful tool that can help researchers discover relevant work that might otherwise be missed by traditional keyword-based searches. This approach aligns perfectly with the growing need for more intuitive and context-aware information retrieval systems in academic research.