Introduction
Many real-world questions cannot be answered with a single document retrieval. They require multi-hop reasoning—finding information from multiple sources and connecting the dots to arrive at a comprehensive answer. Multi-hop search systems excel at answering complex questions that involve relationships, comparisons, and synthesizing information across different documents.
Key Concept: Multi-hop reasoning mimics how humans research complex topics: we find one piece of information, which leads us to search for another, and finally we combine them to form an answer.
Understanding Multi-hop Search
What is Multi-hop Reasoning?
Multi-hop reasoning generally involves three stages:
- First hop: Initial information retrieval based on the original question.
- Intermediate hops: Finding related information based on the results of previous steps.
- Final hop: Synthesizing all gathered information to answer the original question comprehensively.
Example Scenarios
- Comparative Questions: "How does the cost of living in San Francisco compare to Austin?" (Requires finding data for SF, then Austin, then comparing).
- Chain Questions: "Which companies did the founders of Google work for before starting Google?" (Find founders, then find their previous employment history).
- Aggregation Questions: "What is the total market cap of all tech companies founded after 2010?" (Find companies founded >2010, find their market caps, sum them up).
- Causal Questions: "What factors led to the 2008 financial crisis and how did it affect the housing market?" (Find causes, then trace effects on housing).
Building Multi-hop Systems
Basic Multi-hop Architecture
Here is a classic implementation of a multi-hop search module in DSPy using a loop to iteratively retrieve information:
import dspy
class MultiHopSearch(dspy.Module):
def __init__(self, max_hops=3):
super().__init__()
self.max_hops = max_hops
self.retrieve = dspy.Retrieve(k=5)
self.generate_query = dspy.Predict("question, context -> next_query")
self.generate_answer = dspy.ChainOfThought("question, all_contexts -> answer")
def forward(self, question):
all_contexts = []
current_question = question
for hop in range(self.max_hops):
# Retrieve documents for current query
retrieved = self.retrieve(question=current_question)
contexts = retrieved.passages
all_contexts.extend(contexts)
# Generate next query based on retrieved information
next_query_result = self.generate_query(
question=question,
context="\n".join(contexts)
)
# Check if we have enough information (heuristic)
if "sufficient" in next_query_result.next_query.lower():
break
current_question = next_query_result.next_query
# Generate final answer using all retrieved contexts
final_answer = self.generate_answer(
question=question,
all_contexts="\n\n".join(all_contexts)
)
return dspy.Prediction(
answer=final_answer.answer,
contexts=all_contexts,
hops=hop + 1,
reasoning=final_answer.rationale
)Advanced Multi-hop with Question Decomposition
Instead of sequential hopping, you can decompose a complex question into simpler sub-questions first:
class DecomposedMultiHop(dspy.Module):
def __init__(self):
super().__init__()
self.decompose = dspy.Predict("question -> subquestions")
self.retrieve = dspy.Retrieve(k=5)
self.answer_subquestion = dspy.Predict("subquestion, context -> subanswer")
self.synthesize = dspy.ChainOfThought("question, subanswers -> final_answer")
def forward(self, question):
# Decompose the complex question
decomposition = self.decompose(question=question)
subquestions = decomposition.subquestions.split(";")
subanswers = []
# Answer each subquestion
for subq in subquestions:
subq = subq.strip()
if subq:
# Retrieve relevant context
context = self.retrieve(question=subq).passages
# Answer subquestion
subanswer = self.answer_subquestion(
subquestion=subq,
context="\n".join(context)
)
subanswers.append({
"subquestion": subq,
"answer": subanswer.subanswer,
"context": context
})
# Synthesize final answer
subanswers_text = "\n".join([
f"Q: {sa['subquestion']}\nA: {sa['answer']}"
for sa in subanswers
])
synthesis = self.synthesize(
question=question,
subanswers=subanswers_text
)
return dspy.Prediction(
answer=synthesis.answer,
subquestions=subquestions,
subanswers=subanswers,
reasoning=synthesis.rationale
)Graph-based Multi-hop Search
For knowledge-intensive tasks, treating information as a graph of connected entities can be powerful.
class GraphMultiHop(dspy.Module):
def __init__(self):
super().__init__()
self.retrieve = dspy.Retrieve(k=5)
self.extract_entities = dspy.Predict("text -> entities")
self.find_connections = dspy.Predict("entities, question -> search_queries")
self.traverse_graph = dspy.ChainOfThought("question, entities, paths -> answer")
def forward(self, question):
visited_entities = set()
all_paths = []
# Initial retrieval
initial_docs = self.retrieve(question=question).passages
# Extract entities from initial documents
for doc in initial_docs:
entities_result = self.extract_entities(text=doc)
entities = [e.strip() for e in entities_result.entities.split(",")]
for entity in entities:
if entity not in visited_entities and len(visited_entities) < 20:
visited_entities.add(entity)
# Find connections to other entities
connections = self.find_connections(
entities=", ".join(visited_entities),
question=question
)
# Search for each connection
for query in connections.search_queries.split(";"):
query = query.strip()
if query:
related_docs = self.retrieve(question=query).passages
all_paths.extend(related_docs)
# Traverse the graph of connections
traversal = self.traverse_graph(
question=question,
entities=", ".join(list(visited_entities)),
paths="\n".join(all_paths[:20])
)
return dspy.Prediction(
answer=traversal.answer,
entities=list(visited_entities),
paths=all_paths,
reasoning=traversal.rationale
)Specialized Multi-hop Applications
Multi-hop search is essential for various domain-specific applications:
Fact Verification System
Verifying a claim often requires checking multiple sub-claims against evidence.
class FactChecker(dspy.Module):
def __init__(self):
super().__init__()
self.retrieve = dspy.Retrieve(k=5)
self.extract_claims = dspy.Predict("statement -> claims")
self.verify_claim = dspy.Predict("claim, evidence -> verification, confidence")
self.find_contradictions = dspy.Predict("verified_claims -> contradictions")
self.final_judgment = dspy.ChainOfThought(
"statement, verifications, contradictions -> final_verdict, explanation"
)
def forward(self, statement):
# Extract individual claims from the statement
claims_result = self.extract_claims(statement=statement)
claims = [c.strip() for c in claims_result.claims.split(".")]
verifications = []
# Verify each claim
for claim in claims:
if claim:
# Search for evidence
evidence = self.retrieve(question=claim).passages
# Verify the claim with evidence
verification = self.verify_claim(
claim=claim,
evidence="\n".join(evidence)
)
verifications.append({
"claim": claim,
"verification": verification.verification,
"confidence": verification.confidence
})
# Check for contradictions and make judgment
contradictions = self.find_contradictions(
verified_claims="\n".join([f"{v['claim']}: {v['verification']}" for v in verifications])
)
judgment = self.final_judgment(
statement=statement,
verifications="\n".join([str(v) for v in verifications]),
contradictions=contradictions.contradictions
)
return dspy.Prediction(
verdict=judgment.final_verdict,
explanation=judgment.explanation,
reasoning=judgment.rationale
)Optimizing Multi-hop Systems
You can use DSPy optimizers like MIPRO (Multi-prompt Instruction Proposal Optimizer) to automatically improve the prompts and search strategies of your multi-hop system.
# Example training data for optimization
multi_hop_trainset = [
dspy.Example(
question="What is the relationship between quantum computing and cryptography?",
strategy="trace_technology_relationships",
hops_needed=3,
answer="Quantum computing threatens current cryptographic systems while also enabling quantum-resistant cryptography solutions."
),
# ... more complex examples
]
# Optimize with MIPRO
mipro_optimizer = MIPRO(
metric=multi_hop_metric,
num_candidates=10
)
optimized_multihop = mipro_optimizer.compile(OptimizedMultiHop(), trainset=multi_hop_trainset)By defining a custom metric that rewards correct answers, reasoning depth, and efficient hops, you can train your DSPy module to become a more effective researcher.