INTRODUCTION
The landscape of artificial intelligence has transformed dramatically with the emergence of large language models. These powerful systems have opened new possibilities for automating complex research tasks that previously required significant human effort. In this comprehensive tutorial, we will explore how to build a sophisticated multi-agent research system using LlamaIndex, a framework specifically designed for creating LLM-powered applications with advanced retrieval and reasoning capabilities.
Our research agent system represents a paradigm shift in how we approach information gathering and synthesis. Rather than manually searching through countless web pages, academic papers, and online resources, we can deploy a coordinated team of specialized agents that work together to produce comprehensive research reports. This system leverages the computational power of modern GPUs, whether you are using Apple's Metal Performance Shaders on Mac hardware or NVIDIA's CUDA platform on traditional GPU systems.
The architecture we will build consists of several specialized agents, each with distinct responsibilities. An orchestration agent serves as the conductor of this digital symphony, coordinating the activities of all other agents and ensuring smooth workflow execution. A conversation agent handles all user interactions, interpreting research requests and delivering results in a clear, accessible format. Multiple search agents work in parallel to scour different corners of the internet, from academic repositories like ArXiv to general web searches using free services. A summary agent distills lengthy documents into concise, actionable insights. Finally, a report agent assembles all gathered information into polished, professional research reports complete with citations, relevance scores, and comprehensive bibliographies.
What makes this system particularly powerful is its reliance on open source and free services. We will primarily use locally hosted language models that can run on your own hardware, eliminating ongoing API costs and ensuring complete data privacy. For search capabilities, we will leverage free APIs like ArXiv, Semantic Scholar, and DuckDuckGo search, which provide excellent coverage without requiring payment. Only when these free services prove insufficient for specific advanced capabilities will we consider commercial alternatives, and even then, we will provide clear guidance on cost-effective options.
The system intelligently utilizes available GPU resources, automatically detecting and optimizing for Apple MPS on Mac systems or NVIDIA CUDA on compatible hardware. This GPU acceleration is crucial for running local language models efficiently, making the system practical for real-world research tasks.
UNDERSTANDING THE ARCHITECTURE
Before diving into implementation details, we must understand the conceptual architecture that underpins our multi-agent research system. The design follows a hierarchical structure where agents operate at different levels of abstraction, each contributing specialized capabilities to the overall research workflow.
At the highest level sits the orchestration agent. This agent does not perform research tasks directly but instead manages the coordination and sequencing of operations across all other agents. When a user submits a research request, the orchestration agent analyzes the request, determines which agents need to be activated, and in what sequence they should operate. It maintains the state of the research process, tracking which tasks have been completed and which are still pending. The orchestration agent also handles error recovery, ensuring that if one agent fails, the system can gracefully recover and continue with alternative approaches.
The conversation agent serves as the primary interface between the human user and the machine intelligence. This agent must understand natural language requests, which can range from simple topic specifications to complex refinement instructions for existing reports. It translates user intentions into structured commands that other agents can process. Equally important, the conversation agent formats the outputs from other agents into human-readable responses, ensuring that technical details are presented in an accessible manner.
Search agents represent the information gathering backbone of our system. Unlike a single monolithic search component, we deploy multiple search agents that can operate in parallel, each potentially targeting different information sources or search strategies. One search agent might specialize in academic databases, querying ArXiv and Semantic Scholar for peer-reviewed publications. Another might use DuckDuckGo's free search API to find recent news articles and blog posts to capture emerging trends. Yet another could target specific technical documentation or industry reports available through open access repositories. This parallel architecture dramatically reduces the time required to gather comprehensive information on any given topic.
The summary agent plays a crucial role in information distillation. Modern web searches can return hundreds or thousands of potentially relevant documents. Reading and synthesizing all this information would be prohibitively time-consuming for a human researcher. The summary agent applies advanced natural language processing using local LLMs to extract key insights, identify main themes, and condense lengthy documents into concise summaries that preserve essential information while eliminating redundancy.
Finally, the report agent acts as the publishing arm of our research system. It takes the raw outputs from search and summary agents and transforms them into polished, professional research reports. These reports follow a structured format including an abstract, trend analysis, and a comprehensive bibliography with relevance scoring. The report agent ensures consistent formatting, proper citation styles, and logical organization of information.
GPU ACCELERATION AND LOCAL MODEL SUPPORT
One of the most critical aspects of our system is its ability to leverage GPU acceleration across different hardware platforms while running models locally. Modern language models are computationally intensive, and running them efficiently requires careful attention to hardware optimization. Our system must seamlessly support both Apple's Metal Performance Shaders architecture found in M1, M2, and M3 chips, as well as NVIDIA's CUDA platform that powers most dedicated GPU hardware.
The challenge lies in the fact that these platforms use fundamentally different approaches to parallel computing. Apple's MPS framework integrates tightly with the unified memory architecture of Apple Silicon, where CPU and GPU share the same memory pool. This architecture offers unique advantages in terms of memory bandwidth and efficiency but requires specific optimization strategies. NVIDIA CUDA, on the other hand, operates with discrete GPU memory and offers a more traditional but highly mature ecosystem of optimization tools and libraries.
Our system addresses this challenge through abstraction layers that detect the available hardware at runtime and configure the appropriate backend. When the system initializes, it probes for available GPU resources. On a Mac with Apple Silicon, it detects MPS availability and configures PyTorch to use the MPS backend. On systems with NVIDIA GPUs, it detects CUDA availability and configures accordingly. If no GPU is available, the system gracefully falls back to CPU execution, though with reduced performance.
The model support strategy focuses entirely on local execution using open source models. We will use models from the Hugging Face ecosystem, which provides thousands of pre-trained models that can be downloaded and run locally without any API costs. Popular options include Mistral 7B, Llama 2, and Phi-2, all of which offer strong performance for research tasks while being small enough to run on consumer hardware. For even more efficient inference, we will support GGUF format models that can be run through llama-cpp-python, which provides highly optimized inference on both CPU and GPU.
The local model approach offers several compelling advantages. First, it eliminates all per-request costs, making the system economical even for heavy usage. Second, it ensures complete data privacy since no research queries or documents are sent to external servers. Third, it provides consistent performance without dependency on network connectivity or external service availability. The tradeoff is higher upfront computational requirements and the need to download model files, which can range from a few gigabytes to tens of gigabytes depending on model size.
SETTING UP THE DEVELOPMENT ENVIRONMENT
Building our research agent system requires a properly configured development environment with several key dependencies. The foundation is Python, which should be version 3.10 or higher to ensure compatibility with modern libraries. We will use LlamaIndex as our primary framework, which provides high-level abstractions for building LLM applications while still offering the flexibility to customize behavior at a granular level.
The installation process begins with creating a virtual environment to isolate our project dependencies. This practice prevents conflicts with other Python projects and makes dependency management more straightforward. After activating the virtual environment, we install LlamaIndex along with its optional dependencies for specific integrations. We also need PyTorch, which provides the underlying tensor operations and neural network functionality. The PyTorch installation must be configured for the appropriate hardware backend, either MPS for Apple Silicon or CUDA for NVIDIA GPUs.
Additional dependencies include libraries for web scraping and HTTP requests, such as requests and BeautifulSoup for parsing HTML content. We need aiohttp for asynchronous HTTP operations, which allows our search agents to query multiple sources concurrently. For working with academic databases, we include specialized libraries like arxiv for querying the ArXiv API and semanticscholar for accessing Semantic Scholar data. For web search, we will use the duckduckgo-search library, which provides free access to DuckDuckGo search results without requiring an API key.
For local LLM inference, we need either the transformers library from Hugging Face for running standard model formats, or llama-cpp-python for running GGUF format models with optimized performance. The llama-cpp-python library is particularly valuable because it provides excellent performance on both CPU and GPU while supporting a wide range of model formats.
The configuration of GPU support requires careful attention to PyTorch installation. For Apple Silicon systems, we install PyTorch with MPS support, which is included in recent PyTorch releases. For NVIDIA systems, we install the CUDA-enabled version of PyTorch, ensuring that the CUDA version matches the installed NVIDIA drivers. The system should verify GPU availability at runtime and provide clear feedback about which hardware acceleration is being used.
Here is how we initialize the environment and detect available GPU resources:
import torch
import platform
import sys
from typing import Optional, Literal
import os
class HardwareDetector:
"""Detects and configures available GPU hardware for LLM inference."""
def __init__(self):
self.device_type = self._detect_device()
self.device = torch.device(self.device_type)
self._print_device_info()
def _detect_device(self) -> str:
"""Determine the best available compute device."""
if torch.cuda.is_available():
return "cuda"
elif hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
return "mps"
else:
return "cpu"
def _print_device_info(self):
"""Print detailed information about the detected device."""
print("=" * 70)
print("HARDWARE DETECTION REPORT")
print("=" * 70)
if self.device_type == "cuda":
gpu_name = torch.cuda.get_device_name(0)
gpu_count = torch.cuda.device_count()
cuda_version = torch.version.cuda
print(f"GPU Platform: NVIDIA CUDA")
print(f"CUDA Version: {cuda_version}")
print(f"GPU Count: {gpu_count}")
print(f"Primary GPU: {gpu_name}")
total_memory = torch.cuda.get_device_properties(0).total_memory
print(f"GPU Memory: {total_memory / 1024**3:.2f} GB")
elif self.device_type == "mps":
print(f"GPU Platform: Apple Metal Performance Shaders")
print(f"System: {platform.platform()}")
print(f"Processor: {platform.processor()}")
print(f"Architecture: {platform.machine()}")
else:
print(f"GPU Platform: None (CPU only)")
print(f"System: {platform.platform()}")
print(f"Processor: {platform.processor()}")
print(f"Warning: No GPU acceleration available. Performance will be limited.")
print("=" * 70)
print()
def get_device_info(self) -> dict:
"""Return detailed information about the compute device."""
info = {
"device_type": self.device_type,
"device": str(self.device),
"platform": platform.platform(),
"processor": platform.processor()
}
if self.device_type == "cuda":
info["gpu_count"] = torch.cuda.device_count()
info["gpu_name"] = torch.cuda.get_device_name(0)
info["cuda_version"] = torch.version.cuda
info["total_memory_gb"] = torch.cuda.get_device_properties(0).total_memory / 1024**3
return info
def optimize_for_inference(self):
"""Apply device-specific optimizations for inference."""
if self.device_type == "cuda":
torch.backends.cudnn.benchmark = True
torch.backends.cuda.matmul.allow_tf32 = True
print("Applied CUDA optimizations for inference")
elif self.device_type == "mps":
print("MPS device ready for inference")
else:
print("CPU device ready for inference")
This hardware detection code forms the foundation of our GPU acceleration strategy. The HardwareDetector class encapsulates all logic for identifying available compute resources. It first checks for NVIDIA CUDA availability, which is the most common GPU platform for deep learning. If CUDA is not available, it checks for Apple MPS support. If neither GPU platform is available, it falls back to CPU execution.
The detection logic provides detailed feedback about the discovered hardware. For CUDA systems, it reports the GPU name, CUDA version, GPU count, and total memory available. This information is invaluable for understanding whether the system has sufficient resources to run larger models. For MPS systems, it reports platform and processor information. For CPU-only systems, it provides a warning that performance will be limited, helping users understand what to expect.
The optimize_for_inference method applies platform-specific optimizations that can significantly improve inference performance. For CUDA, it enables cuDNN benchmarking, which automatically selects the fastest convolution algorithms for the specific hardware. It also enables TF32 tensor cores on newer NVIDIA GPUs, which provide faster matrix multiplication with minimal accuracy impact.
IMPLEMENTING LOCAL LLM SUPPORT
The core of our research system is the ability to run language models locally without relying on commercial API services. This section covers the implementation of local LLM support using both Hugging Face transformers and llama-cpp-python for maximum flexibility and performance.
We will create an abstraction layer that supports multiple local model backends. This design allows users to choose the model format and inference engine that best suits their hardware and performance requirements. The abstraction provides a consistent interface regardless of which backend is being used, making it easy to switch between different models or inference engines.
For Hugging Face models, we will use the transformers library with PyTorch backend. This combination provides access to thousands of pre-trained models and supports both CPU and GPU inference with automatic device placement. For GGUF models, we will use llama-cpp-python, which provides highly optimized inference through the llama.cpp engine. This engine is particularly efficient for running quantized models on both CPU and GPU.
Here is the implementation of our local LLM manager:
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
from llama_index.llms.llama_cpp import LlamaCPP
from llama_index.llms.huggingface import HuggingFaceLLM
@dataclass
class ModelConfig:
"""Configuration for a local language model."""
model_name: str
model_type: str # 'huggingface' or 'gguf'
model_path: Optional[str] = None
context_window: int = 4096
max_new_tokens: int = 512
temperature: float = 0.7
top_p: float = 0.9
class LocalLLMManager:
"""Manages local language model loading and inference."""
def __init__(self, hardware_detector: HardwareDetector):
self.hardware = hardware_detector
self.loaded_models: Dict[str, Any] = {}
def load_huggingface_model(self, config: ModelConfig):
"""Load a Hugging Face model for local inference."""
print(f"Loading Hugging Face model: {config.model_name}")
model_path = config.model_path or config.model_name
try:
tokenizer = AutoTokenizer.from_pretrained(model_path)
device_map = None
if self.hardware.device_type == "cuda":
device_map = "auto"
elif self.hardware.device_type == "mps":
device_map = {"": "mps"}
else:
device_map = {"": "cpu"}
model = AutoModelForCausalLM.from_pretrained(
model_path,
device_map=device_map,
torch_dtype=torch.float16 if self.hardware.device_type != "cpu" else torch.float32,
low_cpu_mem_usage=True
)
llm = HuggingFaceLLM(
model=model,
tokenizer=tokenizer,
context_window=config.context_window,
max_new_tokens=config.max_new_tokens,
generate_kwargs={
"temperature": config.temperature,
"top_p": config.top_p,
"do_sample": True
},
device_map=device_map
)
self.loaded_models[config.model_name] = llm
print(f"Successfully loaded {config.model_name}")
return llm
except Exception as e:
print(f"Failed to load Hugging Face model: {e}")
raise
def load_gguf_model(self, config: ModelConfig):
"""Load a GGUF format model using llama-cpp-python."""
print(f"Loading GGUF model: {config.model_name}")
if not config.model_path:
raise ValueError("GGUF models require an explicit model_path")
try:
n_gpu_layers = 0
if self.hardware.device_type == "cuda":
n_gpu_layers = -1
elif self.hardware.device_type == "mps":
n_gpu_layers = 1
llm = LlamaCPP(
model_path=config.model_path,
temperature=config.temperature,
max_new_tokens=config.max_new_tokens,
context_window=config.context_window,
generate_kwargs={"top_p": config.top_p},
model_kwargs={"n_gpu_layers": n_gpu_layers},
verbose=True
)
self.loaded_models[config.model_name] = llm
print(f"Successfully loaded {config.model_name}")
return llm
except Exception as e:
print(f"Failed to load GGUF model: {e}")
raise
def load_model(self, config: ModelConfig):
"""Load a model based on its configuration."""
if config.model_name in self.loaded_models:
print(f"Model {config.model_name} already loaded")
return self.loaded_models[config.model_name]
if config.model_type == "huggingface":
return self.load_huggingface_model(config)
elif config.model_type == "gguf":
return self.load_gguf_model(config)
else:
raise ValueError(f"Unknown model type: {config.model_type}")
def get_model(self, model_name: str) -> Optional[Any]:
"""Retrieve a loaded model by name."""
return self.loaded_models.get(model_name)
def unload_model(self, model_name: str):
"""Unload a model to free memory."""
if model_name in self.loaded_models:
del self.loaded_models[model_name]
if self.hardware.device_type == "cuda":
torch.cuda.empty_cache()
print(f"Unloaded model: {model_name}")
The LocalLLMManager class provides a unified interface for loading and managing different types of local language models. The load_huggingface_model method handles models in the standard Hugging Face format. It automatically configures device placement based on the detected hardware, using GPU acceleration when available and falling back to CPU when necessary. The method also selects the appropriate data type, using float16 for GPU inference to reduce memory usage and improve performance, while using float32 for CPU inference to maintain numerical stability.
The load_gguf_model method handles GGUF format models through llama-cpp-python. GGUF is a file format designed for efficient storage and loading of quantized language models. These models are typically much smaller than their full-precision counterparts and can run efficiently even on systems with limited GPU memory. The method configures GPU layer offloading based on available hardware. On CUDA systems, it offloads all layers to the GPU by setting n_gpu_layers to negative one. On MPS systems, it uses partial offloading. On CPU-only systems, it runs entirely on CPU.
The ModelConfig dataclass encapsulates all configuration parameters for a model. This design makes it easy to define and manage multiple model configurations. Users can specify the model name, type, path, context window size, maximum token generation limit, and sampling parameters like temperature and top_p. These parameters control the behavior of the model during text generation, affecting factors like creativity and randomness in the output.
IMPLEMENTING THE ORCHESTRATION AGENT
The orchestration agent represents the brain of our multi-agent system. It receives high-level research requests from the conversation agent and translates them into a coordinated sequence of operations across specialized agents. The orchestration logic must be sophisticated enough to handle complex workflows while remaining flexible enough to adapt to different types of research requests.
At its core, the orchestration agent maintains a state machine that tracks the progress of each research task. When a new research request arrives, the orchestrator creates a new task context that includes the research topic, any specific requirements or constraints, and metadata about the request. This context is passed through various stages of processing, with each stage potentially involving one or more specialized agents.
The orchestration workflow typically follows this pattern: First, the conversation agent parses the user request and extracts key parameters such as the research topic, desired depth of analysis, and any specific sources to prioritize. The orchestrator receives this structured request and creates a research plan. This plan identifies which search agents should be activated, what queries they should execute, and how results should be aggregated.
Next, the orchestrator dispatches search tasks to multiple search agents in parallel. Each search agent operates independently, querying its designated sources and returning raw results. The orchestrator collects these results as they arrive, using asynchronous programming patterns to avoid blocking on slow network requests. Once all search results are collected, or a timeout is reached, the orchestrator moves to the summarization phase.
During summarization, the orchestrator batches documents and sends them to summary agents. If the volume of documents is large, multiple summary agents may work in parallel, each processing a subset of the total document collection. The orchestrator aggregates the summaries and identifies key themes and trends across all documents.
Finally, the orchestrator invokes the report agent with all collected and processed information. The report agent generates the final research report, which the orchestrator then passes back to the conversation agent for presentation to the user.
Here is the implementation of the core orchestration logic:
from typing import List, Dict, Any, Optional
import asyncio
from datetime import datetime
from dataclasses import dataclass, field
from enum import Enum
import uuid
class TaskStatus(Enum):
"""Enumeration of possible task states."""
PENDING = "pending"
SEARCHING = "searching"
SUMMARIZING = "summarizing"
GENERATING_REPORT = "generating_report"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class ResearchTask:
"""Represents a single research task with its state and results."""
task_id: str
topic: str
requirements: Optional[str] = None
status: TaskStatus = TaskStatus.PENDING
created_at: datetime = field(default_factory=datetime.now)
search_results: List[Dict[str, Any]] = field(default_factory=list)
summaries: List[Dict[str, Any]] = field(default_factory=list)
final_report: Optional[str] = None
error_message: Optional[str] = None
progress_log: List[str] = field(default_factory=list)
def log_progress(self, message: str):
"""Add a progress message to the task log."""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
log_entry = f"[{timestamp}] {message}"
self.progress_log.append(log_entry)
print(log_entry)
class OrchestrationAgent:
"""Coordinates the activities of all specialized agents."""
def __init__(self, search_agents: List, summary_agent, report_agent):
self.search_agents = search_agents
self.summary_agent = summary_agent
self.report_agent = report_agent
self.active_tasks: Dict[str, ResearchTask] = {}
self.completed_tasks: Dict[str, ResearchTask] = {}
async def execute_research(self, topic: str, requirements: Optional[str] = None) -> ResearchTask:
"""Execute a complete research workflow for the given topic."""
task_id = str(uuid.uuid4())
task = ResearchTask(task_id=task_id, topic=topic, requirements=requirements)
self.active_tasks[task_id] = task
task.log_progress(f"Starting research on topic: {topic}")
try:
await self._search_phase(task)
await self._summarization_phase(task)
await self._report_generation_phase(task)
task.status = TaskStatus.COMPLETED
task.log_progress("Research task completed successfully")
self.completed_tasks[task_id] = task
del self.active_tasks[task_id]
except Exception as e:
task.status = TaskStatus.FAILED
task.error_message = str(e)
task.log_progress(f"Research task failed: {e}")
self.completed_tasks[task_id] = task
if task_id in self.active_tasks:
del self.active_tasks[task_id]
return task
async def _search_phase(self, task: ResearchTask):
"""Execute parallel search across multiple agents."""
task.status = TaskStatus.SEARCHING
task.log_progress(f"Starting parallel search with {len(self.search_agents)} agents")
search_tasks = [
agent.search(task.topic)
for agent in self.search_agents
]
search_results = await asyncio.gather(*search_tasks, return_exceptions=True)
valid_results = []
for idx, result in enumerate(search_results):
if isinstance(result, Exception):
task.log_progress(f"Search agent {idx} failed: {result}")
else:
valid_results.extend(result)
task.log_progress(f"Search agent {idx} returned {len(result)} results")
task.search_results = valid_results
task.log_progress(f"Search phase completed with {len(valid_results)} total results")
if len(valid_results) == 0:
raise ValueError("No search results found from any agent")
async def _summarization_phase(self, task: ResearchTask):
"""Summarize all collected documents."""
task.status = TaskStatus.SUMMARIZING
task.log_progress(f"Starting summarization of {len(task.search_results)} documents")
summary_tasks = [
self.summary_agent.summarize(doc)
for doc in task.search_results
]
summaries = await asyncio.gather(*summary_tasks, return_exceptions=True)
valid_summaries = []
for idx, summary in enumerate(summaries):
if isinstance(summary, Exception):
task.log_progress(f"Summarization failed for document {idx}: {summary}")
else:
valid_summaries.append(summary)
task.summaries = valid_summaries
task.log_progress(f"Summarization phase completed with {len(valid_summaries)} summaries")
if len(valid_summaries) == 0:
raise ValueError("No summaries generated successfully")
async def _report_generation_phase(self, task: ResearchTask):
"""Generate the final research report."""
task.status = TaskStatus.GENERATING_REPORT
task.log_progress("Starting report generation")
report = await self.report_agent.generate_report(
topic=task.topic,
requirements=task.requirements,
search_results=task.search_results,
summaries=task.summaries
)
task.final_report = report
task.log_progress(f"Report generation completed ({len(report)} characters)")
def get_task_status(self, task_id: str) -> Optional[ResearchTask]:
"""Retrieve the current status of a research task."""
if task_id in self.active_tasks:
return self.active_tasks[task_id]
return self.completed_tasks.get(task_id)
def get_all_tasks(self) -> Dict[str, ResearchTask]:
"""Retrieve all tasks (active and completed)."""
all_tasks = {}
all_tasks.update(self.active_tasks)
all_tasks.update(self.completed_tasks)
return all_tasks
The orchestration implementation demonstrates several important design patterns. The use of asynchronous programming with asyncio allows multiple search agents to operate concurrently, dramatically reducing the total time required to gather information. The gather function collects results from all agents, using return_exceptions to ensure that a failure in one agent does not crash the entire workflow.
The ResearchTask dataclass serves as a comprehensive container for all information related to a single research request. It tracks the task status through its lifecycle using the TaskStatus enumeration, which provides clear visibility into what phase of processing is currently active. The task stores intermediate results from each processing phase, including raw search results, generated summaries, and the final report. The progress_log field maintains a detailed audit trail of all operations, which is invaluable for debugging and monitoring.
Error handling is built into every phase of the orchestration workflow. When a search agent fails, the orchestrator logs the error but continues processing results from successful agents. This resilience ensures that partial failures do not prevent the system from producing useful output. The final report generation only proceeds if at least some valid search results and summaries are available. If all agents fail in a particular phase, the orchestrator raises an exception and marks the task as failed.
The separation of concerns between different phases makes the orchestration logic easy to understand and maintain. Each phase has a dedicated method that encapsulates all logic for that phase. This design makes it straightforward to add new phases or modify existing ones without affecting the overall workflow structure.
BUILDING THE CONVERSATION AGENT
The conversation agent serves as the primary interface between human users and our research system. This agent must understand natural language requests, maintain conversation context across multiple interactions, and present results in a clear, accessible format. Unlike the other agents in our system, which operate on structured data, the conversation agent works primarily with unstructured natural language.
The conversation agent leverages the language understanding capabilities of our local language models to interpret user intentions. When a user submits a request, the conversation agent analyzes the text to extract key information such as the research topic, any specific requirements or constraints, and whether this is a new research request or a refinement of existing work. This analysis is performed using carefully crafted prompts that instruct the LLM on what information to extract and how to format the response.
Context management is crucial for the conversation agent. Users often engage in multi-turn conversations where later requests reference earlier ones. For example, a user might first request research on quantum computing, then ask to refine the report to focus specifically on quantum error correction. The conversation agent must maintain a history of the conversation and use this context to correctly interpret follow-up requests.
The conversation agent also handles the presentation of results. When the orchestration agent completes a research task, it returns a structured ResearchTask object. The conversation agent transforms this structured data into natural language responses that are easy for users to understand. It might highlight key findings, summarize the main trends discovered, and provide guidance on how to access the full report.
Here is the implementation of the conversation agent:
from typing import List, Dict, Optional, Tuple, Any
from dataclasses import dataclass
import re
from datetime import datetime
@dataclass
class ConversationTurn:
"""Represents a single turn in the conversation."""
user_message: str
agent_response: str
timestamp: datetime
extracted_intent: Optional[Dict[str, Any]] = None
task_id: Optional[str] = None
class ConversationAgent:
"""Manages natural language interactions with users."""
def __init__(self, llm):
self.llm = llm
self.conversation_history: List[ConversationTurn] = []
def parse_user_request(self, user_message: str) -> Dict[str, Any]:
"""Extract structured intent from natural language user input."""
prompt = f"""Analyze the following user request and extract structured information about what research they want.
User Request: {user_message}
Extract the following information:
- Research Topic: The main subject the user wants to research
- Request Type: Either 'new_research', 'refine_research', or 'question'
- Specific Requirements: Any particular constraints or focus areas mentioned
- Reference to Previous Work: Whether this refers to earlier research (yes/no)
Provide your analysis in exactly this format: TOPIC: [extracted topic or 'none'] TYPE: [request type] REQUIREMENTS: [specific requirements or 'none'] PREVIOUS_REFERENCE: [yes or no]
Example: TOPIC: machine learning in healthcare TYPE: new_research REQUIREMENTS: focus on diagnostic applications PREVIOUS_REFERENCE: no """
response = self.llm.complete(prompt).text
intent = self._parse_llm_response(response)
if intent.get('previous_reference') and self.conversation_history:
previous_topic = self._get_previous_topic()
if previous_topic:
intent['previous_topic'] = previous_topic
if intent.get('topic') == 'none':
intent['topic'] = previous_topic
return intent
def _parse_llm_response(self, response: str) -> Dict[str, Any]:
"""Parse the structured response from the LLM."""
intent = {}
topic_match = re.search(r'TOPIC:\s*(.+?)(?:\n|$)', response, re.IGNORECASE)
if topic_match:
topic = topic_match.group(1).strip()
intent['topic'] = None if topic.lower() == 'none' else topic
type_match = re.search(r'TYPE:\s*(.+?)(?:\n|$)', response, re.IGNORECASE)
if type_match:
intent['request_type'] = type_match.group(1).strip()
req_match = re.search(r'REQUIREMENTS:\s*(.+?)(?:\n|$)', response, re.IGNORECASE)
if req_match:
reqs = req_match.group(1).strip()
intent['requirements'] = None if reqs.lower() == 'none' else reqs
prev_match = re.search(r'PREVIOUS_REFERENCE:\s*(.+?)(?:\n|$)', response, re.IGNORECASE)
if prev_match:
intent['previous_reference'] = prev_match.group(1).strip().lower() == 'yes'
return intent
def _get_previous_topic(self) -> Optional[str]:
"""Retrieve the topic from the most recent research request."""
for turn in reversed(self.conversation_history):
if turn.extracted_intent and turn.extracted_intent.get('topic'):
return turn.extracted_intent['topic']
return None
def format_research_results(self, task: ResearchTask) -> str:
"""Format research results into a user-friendly response."""
if task.status == TaskStatus.FAILED:
return f"""I apologize, but the research task encountered an error and could not be completed.
Error Details: {task.error_message}
Progress Log: {chr(10).join(task.progress_log[-5:])}
Please try again with a different topic or contact support if the issue persists."""
if task.status != TaskStatus.COMPLETED:
return f"""Your research task is currently in progress.
Status: {task.status.value} Topic: {task.topic} Started: {task.created_at.strftime('%Y-%m-%d %H:%M:%S')}
Recent Progress: {chr(10).join(task.progress_log[-3:])}
Please check back in a few moments for the completed report."""
lines = []
lines.append("=" * 70)
lines.append(f"RESEARCH REPORT: {task.topic.upper()}")
lines.append("=" * 70)
lines.append("")
lines.append(f"Research completed: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
lines.append(f"Sources analyzed: {len(task.search_results)}")
lines.append(f"Summaries generated: {len(task.summaries)}")
lines.append("")
lines.append("EXECUTIVE SUMMARY")
lines.append("-" * 70)
if task.final_report:
report_lines = task.final_report.split('\n')
abstract_start = -1
abstract_end = -1
for i, line in enumerate(report_lines):
if 'ABSTRACT' in line.upper():
abstract_start = i
elif abstract_start >= 0 and (line.startswith('##') or line.startswith('TRENDS') or line.startswith('REFERENCES')):
abstract_end = i
break
if abstract_start >= 0:
if abstract_end < 0:
abstract_end = min(abstract_start + 15, len(report_lines))
for line in report_lines[abstract_start:abstract_end]:
if line.strip():
lines.append(line)
lines.append("")
lines.append("The full research report is available below.")
lines.append("It includes detailed analysis, trends, and comprehensive references.")
lines.append("")
lines.append("=" * 70)
return '\n'.join(lines)
def generate_welcome_message(self) -> str:
"""Generate a welcome message for new users."""
return """Welcome to the AI Research Agent System!
I am your research assistant, powered by local language models and open source tools. I can help you conduct comprehensive research on any topic by:
- Searching academic databases (ArXiv, Semantic Scholar)
- Finding relevant web resources using DuckDuckGo
- Summarizing research papers and articles
- Generating comprehensive research reports with citations
To get started, simply tell me what topic you would like to research. For example:
- "Research recent developments in quantum computing"
- "Find information about renewable energy technologies"
- "Investigate advances in natural language processing"
You can also refine previous research or ask follow-up questions. What would you like to research today?"""
def add_to_history(self, user_message: str, agent_response: str,
intent: Dict[str, Any], task_id: Optional[str] = None):
"""Record a conversation turn in the history."""
turn = ConversationTurn(
user_message=user_message,
agent_response=agent_response,
timestamp=datetime.now(),
extracted_intent=intent,
task_id=task_id
)
self.conversation_history.append(turn)
def get_conversation_summary(self) -> str:
"""Generate a summary of the conversation history."""
if not self.conversation_history:
return "No conversation history available."
lines = []
lines.append("CONVERSATION HISTORY")
lines.append("=" * 70)
for idx, turn in enumerate(self.conversation_history, 1):
lines.append(f"\nTurn {idx} - {turn.timestamp.strftime('%Y-%m-%d %H:%M:%S')}")
lines.append(f"User: {turn.user_message[:100]}...")
if turn.extracted_intent and turn.extracted_intent.get('topic'):
lines.append(f"Topic: {turn.extracted_intent['topic']}")
if turn.task_id:
lines.append(f"Task ID: {turn.task_id}")
return '\n'.join(lines)
The conversation agent implementation demonstrates sophisticated natural language processing capabilities. The parse_user_request method uses the local LLM to analyze user input and extract structured information. This approach is more flexible than rule-based parsing because it can handle the natural variability in how users express their requests. The prompt engineering is crucial here, providing clear instructions to the LLM about what information to extract and how to format the response.
The use of regular expressions to parse the LLM response provides robustness against minor variations in formatting. Even if the LLM does not follow the exact format specified in the prompt, the regex patterns can often still extract the relevant information. The method handles missing or malformed fields gracefully, ensuring that the system can continue operating even if the parsing is not perfect.
Context management through conversation history enables the agent to handle multi-turn interactions naturally. The _get_previous_topic method searches backward through the conversation history to find the most recent research topic. This allows users to make references like "refine that report" or "tell me more about the previous topic" without having to repeat the full topic specification.
The format_research_results method transforms the structured ResearchTask object into natural language that is easy for users to read. It provides different responses based on the task status, giving appropriate feedback whether the task is still in progress, has completed successfully, or has failed. For completed tasks, it extracts and highlights key information like the abstract and provides a summary of what was accomplished.
The generate_welcome_message method creates a friendly introduction that helps new users understand what the system can do and how to interact with it. This kind of user guidance is essential for making the system accessible to people who may not be familiar with AI research tools.
IMPLEMENTING FREE SEARCH AGENTS
Search agents form the information gathering backbone of our research system. Each search agent is responsible for querying specific information sources and returning relevant results. The power of our multi-agent architecture comes from deploying multiple search agents in parallel, each targeting different sources or using different search strategies. Critically, we will implement these agents using entirely free and open source services to eliminate ongoing costs.
A search agent must handle several key responsibilities. First, it must translate the research topic into appropriate search queries for its target source. Different sources require different query formats and search strategies. Academic databases like ArXiv respond well to precise technical queries, while general web search benefits from broader, more natural language queries. Second, the search agent must execute the search and handle any network-related issues such as timeouts or rate limiting. Third, it must parse the results from the source into a standardized format that other agents can process.
We will implement several specialized search agents using free services. An ArXiv search agent queries the ArXiv API for academic papers related to the research topic. This agent is particularly valuable for technical and scientific research topics where peer-reviewed publications are essential. A Semantic Scholar agent searches for academic publications across a broader range of sources than ArXiv alone, providing excellent coverage of computer science and related fields. A DuckDuckGo search agent uses the free DuckDuckGo search API to find relevant web pages, blog posts, and news articles without requiring any API keys or authentication.
Each search agent returns results in a standardized format that includes the document title, authors, publication date, abstract or summary, and source URL. This standardization is crucial for the downstream summarization and report generation phases.
Here is the implementation of an ArXiv search agent:
import arxiv
from typing import List, Dict, Any
from datetime import datetime
import asyncio
class ArxivSearchAgent:
"""Searches ArXiv for academic papers related to the research topic."""
def __init__(self, max_results: int = 20):
self.max_results = max_results
self.source_name = "ArXiv"
async def search(self, topic: str) -> List[Dict[str, Any]]:
"""Search ArXiv for papers related to the given topic."""
print(f"[ArXiv Agent] Searching for: {topic}")
try:
search = arxiv.Search(
query=topic,
max_results=self.max_results,
sort_by=arxiv.SortCriterion.Relevance
)
results = []
for paper in search.results():
result = {
'title': paper.title,
'authors': [author.name for author in paper.authors],
'published': paper.published.strftime('%Y-%m-%d'),
'updated': paper.updated.strftime('%Y-%m-%d') if paper.updated else None,
'abstract': paper.summary.replace('\n', ' ').strip(),
'url': paper.entry_id,
'pdf_url': paper.pdf_url,
'source': self.source_name,
'categories': paper.categories,
'primary_category': paper.primary_category,
'doi': paper.doi,
'journal_ref': paper.journal_ref,
'comment': paper.comment
}
results.append(result)
print(f"[ArXiv Agent] Found {len(results)} papers")
return results
except Exception as e:
print(f"[ArXiv Agent] Search failed: {e}")
return []
def calculate_relevance(self, result: Dict[str, Any], topic: str) -> float:
"""Calculate a relevance score for a search result."""
score = 0.0
topic_lower = topic.lower()
topic_words = set(topic_lower.split())
title_lower = result['title'].lower()
if topic_lower in title_lower:
score += 5.0
else:
title_words = set(title_lower.split())
common_words = topic_words.intersection(title_words)
score += len(common_words) * 0.5
abstract_lower = result['abstract'].lower()
if topic_lower in abstract_lower:
score += 3.0
else:
abstract_words = set(abstract_lower.split())
common_words = topic_words.intersection(abstract_words)
score += len(common_words) * 0.2
return min(score, 10.0)
The ArXiv search agent demonstrates a clean, focused implementation using the free ArXiv API. The arxiv Python library handles the low-level details of query formatting and result parsing, making the agent code straightforward and maintainable. The agent configures the search to return results sorted by relevance, ensuring that the most pertinent papers appear first.
The result formatting is comprehensive, capturing all relevant metadata about each paper. This includes not just the title and abstract, but also detailed author information, publication dates, multiple URL formats for accessing the paper, category classifications, DOI identifiers, and journal references. This rich metadata is valuable for generating comprehensive research reports with proper citations.
The calculate_relevance method implements a simple but effective relevance scoring algorithm. It analyzes how well the search result matches the research topic by looking for exact matches and word overlap in both the title and abstract. This scoring can be used later to filter or rank results, ensuring that the most relevant papers receive priority in summarization and reporting.
Now let us implement a Semantic Scholar search agent:
import aiohttp
from typing import List, Dict, Any, Optional
import asyncio
class SemanticScholarAgent:
"""Searches Semantic Scholar for academic publications."""
def __init__(self, max_results: int = 20, api_key: Optional[str] = None):
self.max_results = max_results
self.api_key = api_key
self.base_url = "https://api.semanticscholar.org/graph/v1/paper/search"
self.source_name = "Semantic Scholar"
async def search(self, topic: str) -> List[Dict[str, Any]]:
"""Search Semantic Scholar for papers related to the given topic."""
print(f"[Semantic Scholar Agent] Searching for: {topic}")
try:
params = {
'query': topic,
'limit': self.max_results,
'fields': 'title,authors,year,abstract,url,citationCount,publicationDate,venue,externalIds'
}
headers = {}
if self.api_key:
headers['x-api-key'] = self.api_key
async with aiohttp.ClientSession() as session:
async with session.get(self.base_url, params=params, headers=headers) as response:
if response.status != 200:
print(f"[Semantic Scholar Agent] API returned status {response.status}")
return []
data = await response.json()
results = []
for paper in data.get('data', []):
authors = [author.get('name', 'Unknown') for author in paper.get('authors', [])]
abstract = paper.get('abstract', '')
if not abstract:
abstract = f"Citation count: {paper.get('citationCount', 0)}. Published in {paper.get('venue', 'Unknown venue')}."
result = {
'title': paper.get('title', 'Untitled'),
'authors': authors,
'published': paper.get('publicationDate', paper.get('year', 'Unknown')),
'abstract': abstract,
'url': paper.get('url', ''),
'source': self.source_name,
'citation_count': paper.get('citationCount', 0),
'venue': paper.get('venue', ''),
'external_ids': paper.get('externalIds', {})
}
results.append(result)
print(f"[Semantic Scholar Agent] Found {len(results)} papers")
return results
except Exception as e:
print(f"[Semantic Scholar Agent] Search failed: {e}")
return []
The Semantic Scholar agent uses the free Semantic Scholar API, which provides excellent coverage of computer science and related academic fields. The API does not require authentication for basic usage, though an API key can be provided for higher rate limits. The agent uses aiohttp for asynchronous HTTP requests, allowing it to operate efficiently within our async orchestration framework.
The result formatting normalizes Semantic Scholar results to match the same structure as ArXiv results. This normalization is crucial for ensuring that downstream agents can process results from different sources uniformly. The agent includes citation count information, which can be valuable for assessing the impact and credibility of papers. When abstracts are not available, the agent creates a synthetic summary using available metadata like citation count and venue.
Now let us implement a DuckDuckGo search agent:
from duckduckgo_search import DDGS
from typing import List, Dict, Any
import asyncio
from datetime import datetime
class DuckDuckGoSearchAgent:
"""Searches the web using DuckDuckGo's free search service."""
def __init__(self, max_results: int = 15):
self.max_results = max_results
self.source_name = "Web Search (DuckDuckGo)"
async def search(self, topic: str) -> List[Dict[str, Any]]:
"""Search the web for content related to the given topic."""
print(f"[DuckDuckGo Agent] Searching for: {topic}")
try:
loop = asyncio.get_event_loop()
search_results = await loop.run_in_executor(
None,
self._perform_search,
topic
)
results = []
for item in search_results[:self.max_results]:
result = {
'title': item.get('title', 'Untitled'),
'authors': ['Web Source'],
'published': datetime.now().strftime('%Y-%m-%d'),
'abstract': item.get('body', ''),
'url': item.get('href', ''),
'source': self.source_name,
'domain': self._extract_domain(item.get('href', ''))
}
results.append(result)
print(f"[DuckDuckGo Agent] Found {len(results)} results")
return results
except Exception as e:
print(f"[DuckDuckGo Agent] Search failed: {e}")
return []
def _perform_search(self, topic: str) -> List[Dict[str, Any]]:
"""Perform the actual search using DDGS."""
with DDGS() as ddgs:
results = list(ddgs.text(topic, max_results=self.max_results))
return results
def _extract_domain(self, url: str) -> str:
"""Extract the domain name from a URL."""
try:
from urllib.parse import urlparse
parsed = urlparse(url)
return parsed.netloc
except:
return "unknown"
The DuckDuckGo search agent uses the duckduckgo-search library, which provides free access to DuckDuckGo search results without requiring any API keys or authentication. This makes it an excellent choice for web search in a cost-free research system. The library does not natively support async operations, so the agent uses run_in_executor to run the synchronous search in a thread pool, allowing it to integrate smoothly with our async orchestration framework.
The result formatting normalizes web search results to match the same structure as academic search results. Since web pages typically do not have traditional authors or publication dates in the same sense as academic papers, the agent uses placeholder values. The domain extraction provides useful information about the source of each result, which can help users assess credibility.
IMPLEMENTING THE SUMMARY AGENT
The summary agent plays a critical role in transforming raw search results into digestible insights. Modern web searches can return dozens or hundreds of documents, each potentially containing thousands of words. Reading and synthesizing all this information manually would be extremely time-consuming. The summary agent automates this process using the language understanding and generation capabilities of our local language models.
The summary agent receives individual documents from the orchestration agent and produces concise summaries that capture the essential information. These summaries must be detailed enough to be useful but concise enough to allow a human reader to quickly grasp the key points. The agent must also identify and extract specific information that will be needed for the final report, such as key findings, methodologies, and conclusions.
The summarization process uses carefully crafted prompts that instruct the LLM on what information to extract and how to format the summary. Different types of documents may require different summarization strategies. Academic papers benefit from summaries that highlight the research question, methodology, key findings, and implications. News articles and blog posts might be better summarized by extracting the main narrative and key facts.
Here is the implementation of the summary agent:
from typing import Dict, Any, Optional
import asyncio
class SummaryAgent:
"""Generates concise summaries of documents using local LLM capabilities."""
def __init__(self, llm, max_summary_length: int = 300):
self.llm = llm
self.max_summary_length = max_summary_length
async def summarize(self, document: Dict[str, Any]) -> Dict[str, Any]:
"""Generate a summary of the given document."""
title = document.get('title', 'Untitled')
print(f"[Summary Agent] Summarizing: {title[:60]}...")
try:
loop = asyncio.get_event_loop()
summary_text = await loop.run_in_executor(
None,
self._generate_summary,
document
)
relevance_score = await loop.run_in_executor(
None,
self._calculate_relevance,
document,
summary_text
)
summary = {
'original_document': document,
'summary_text': summary_text,
'relevance_score': relevance_score,
'title': title,
'source': document.get('source', 'Unknown'),
'url': document.get('url', ''),
'authors': document.get('authors', []),
'published': document.get('published', 'Unknown')
}
return summary
except Exception as e:
print(f"[Summary Agent] Summarization failed for '{title}': {e}")
return {
'original_document': document,
'summary_text': document.get('abstract', '')[:self.max_summary_length],
'relevance_score': 5.0,
'title': title,
'source': document.get('source', 'Unknown'),
'url': document.get('url', ''),
'authors': document.get('authors', []),
'published': document.get('published', 'Unknown'),
'error': str(e)
}
def _generate_summary(self, document: Dict[str, Any]) -> str:
"""Generate a summary using the local LLM."""
title = document.get('title', 'Untitled')
abstract = document.get('abstract', '')
source = document.get('source', 'Unknown')
if source == "ArXiv" or source == "Semantic Scholar":
prompt = f"""Summarize the following academic paper in approximately {self.max_summary_length} words. Focus on:
- The main research question or problem addressed
- The methodology or approach used
- Key findings or results
- Significance and implications
Title: {title}
Abstract: {abstract}
Provide a concise summary:""" else: prompt = f"""Summarize the following web content in approximately {self.max_summary_length} words. Focus on:
- The main topic or subject
- Key points or arguments
- Important facts or data
- Practical implications or applications
Title: {title}
Content: {abstract}
Provide a concise summary:"""
response = self.llm.complete(prompt)
summary = response.text.strip()
if len(summary.split()) > self.max_summary_length * 1.5:
words = summary.split()
summary = ' '.join(words[:self.max_summary_length]) + '...'
return summary
def _calculate_relevance(self, document: Dict[str, Any], summary: str) -> float:
"""Calculate a relevance score for the document."""
prompt = f"""Rate the relevance and quality of this research document on a scale from 1 to 10, where:
1-3: Low relevance or quality (tangentially related, poor quality, or outdated) 4-6: Moderate relevance or quality (somewhat related, acceptable quality) 7-9: High relevance or quality (directly related, high quality, authoritative) 10: Exceptional relevance and quality (highly relevant, seminal work, authoritative source)
Title: {document.get('title', 'Untitled')} Source: {document.get('source', 'Unknown')} Summary: {summary}
Provide only a single number from 1 to 10 as your rating:"""
try:
response = self.llm.complete(prompt)
rating_text = response.text.strip()
import re
numbers = re.findall(r'\d+', rating_text)
if numbers:
rating = int(numbers[0])
rating = max(1, min(10, rating))
return float(rating)
else:
return 5.0
except:
return 5.0
The summary agent implementation demonstrates sophisticated use of local LLMs for document understanding and synthesis. The summarize method orchestrates the entire summarization process, generating both a textual summary and a relevance score for each document. The use of run_in_executor allows the synchronous LLM calls to integrate with our async architecture without blocking the event loop.
The _generate_summary method uses different prompting strategies based on the document source. Academic papers receive prompts that focus on research methodology, findings, and implications. Web content receives prompts that focus on main topics, key points, and practical applications. This adaptive prompting ensures that summaries capture the most relevant information for each document type.
The prompt engineering is crucial for generating high-quality summaries. The prompts provide clear instructions about what information to extract and approximately how long the summary should be. By specifying a target length in words rather than characters, we give the LLM more flexibility to produce natural-sounding summaries. The post-processing step truncates summaries that exceed the target length, ensuring consistency across all summaries.
The _calculate_relevance method uses the LLM to assess the quality and relevance of each document. This assessment is valuable for ranking results and helping users identify the most important sources. The prompt provides a clear rubric for scoring, with specific criteria for different score ranges. The method extracts the numerical rating from the LLM response using regular expressions, providing robustness against variations in response format.
Error handling ensures that the summarization process continues even if individual documents fail. When summarization fails, the agent falls back to using the document's abstract directly, truncated to the maximum length. This graceful degradation ensures that the system can still produce useful output even when some summaries cannot be generated.
IMPLEMENTING THE REPORT AGENT
The report agent represents the final stage of our research pipeline, transforming all collected and processed information into a polished, comprehensive research report. This agent must synthesize information from multiple sources, organize it logically, and present it in a format that is both professional and easy to read. The report should include all essential components: an abstract, trend analysis, and a comprehensive bibliography with relevance scores.
The report generation process is complex because it must handle variable amounts of input data and produce consistent output regardless of the specific research topic. The agent must identify common themes across multiple summaries, extract key trends and developments, and organize references in a meaningful way. All of this must be accomplished using our local LLM without relying on external services.
The report structure follows academic conventions while remaining accessible to general readers. It begins with the research topic and an abstract that provides a high-level overview of the findings. The trends and developments section synthesizes information from all sources to identify major themes, recent advances, and future directions. The references section provides detailed citations with relevance scores, publication information, and brief summaries.
Here is the implementation of the report agent:
from typing import List, Dict, Any, Optional
from datetime import datetime
import asyncio
class ReportAgent:
"""Generates comprehensive research reports from collected information."""
def __init__(self, llm):
self.llm = llm
async def generate_report(self, topic: str, requirements: Optional[str],
search_results: List[Dict[str, Any]],
summaries: List[Dict[str, Any]]) -> str:
"""Generate a complete research report."""
print(f"[Report Agent] Generating report for: {topic}")
loop = asyncio.get_event_loop()
abstract = await loop.run_in_executor(
None,
self._generate_abstract,
topic,
requirements,
summaries
)
trends = await loop.run_in_executor(
None,
self._generate_trends_analysis,
topic,
summaries
)
references = self._format_references(summaries)
report = self._assemble_report(topic, abstract, trends, references)
print(f"[Report Agent] Report generated ({len(report)} characters)")
return report
def _generate_abstract(self, topic: str, requirements: Optional[str],
summaries: List[Dict[str, Any]]) -> str:
"""Generate an abstract for the research report."""
summary_texts = [s['summary_text'] for s in summaries[:10]]
combined_summaries = '\n\n'.join(summary_texts)
req_text = f"\nSpecific Requirements: {requirements}" if requirements else ""
prompt = f"""Based on the following research summaries, write a comprehensive abstract for a research report on "{topic}".{req_text}
The abstract should:
- Provide an overview of the current state of research on this topic
- Highlight the most significant findings and developments
- Indicate the scope and coverage of the research
- Be approximately 200-300 words
Research Summaries: {combined_summaries}
Write the abstract:"""
response = self.llm.complete(prompt)
return response.text.strip()
def _generate_trends_analysis(self, topic: str,
summaries: List[Dict[str, Any]]) -> str:
"""Generate an analysis of trends and developments."""
summary_texts = [s['summary_text'] for s in summaries]
combined_summaries = '\n\n'.join(summary_texts)
prompt = f"""Based on the following research summaries about "{topic}", identify and describe the major trends, developments, and themes.
Organize your analysis into clear sections covering:
- Current State: What is the current state of this field or topic?
- Recent Developments: What are the most significant recent advances or changes?
- Key Themes: What are the recurring themes or focus areas across the research?
- Future Directions: What appear to be emerging trends or future research directions?
- Challenges and Opportunities: What challenges and opportunities are identified?
Provide a comprehensive analysis of approximately 500-800 words.
Research Summaries: {combined_summaries}
Write the trends and developments analysis:"""
response = self.llm.complete(prompt)
return response.text.strip()
def _format_references(self, summaries: List[Dict[str, Any]]) -> str:
"""Format the references section with detailed citations."""
sorted_summaries = sorted(summaries, key=lambda x: x.get('relevance_score', 0), reverse=True)
lines = []
for idx, summary in enumerate(sorted_summaries, 1):
doc = summary['original_document']
lines.append(f"\n[{idx}] {summary['title']}")
lines.append(f" Relevance: {summary['relevance_score']:.1f}/10")
authors = doc.get('authors', [])
if authors:
author_str = ', '.join(authors[:3])
if len(authors) > 3:
author_str += f" et al. ({len(authors)} authors total)"
lines.append(f" Authors: {author_str}")
published = doc.get('published', 'Unknown')
lines.append(f" Published: {published}")
source = doc.get('source', 'Unknown')
lines.append(f" Source: {source}")
if 'citation_count' in doc:
lines.append(f" Citations: {doc['citation_count']}")
url = doc.get('url', '')
if url:
lines.append(f" URL: {url}")
pdf_url = doc.get('pdf_url', '')
if pdf_url and pdf_url != url:
lines.append(f" PDF: {pdf_url}")
summary_text = summary['summary_text']
if len(summary_text) > 400:
summary_text = summary_text[:397] + '...'
lines.append(f" Summary: {summary_text}")
lines.append("")
return '\n'.join(lines)
def _assemble_report(self, topic: str, abstract: str, trends: str, references: str) -> str:
"""Assemble all components into the final report."""
lines = []
lines.append("=" * 80)
lines.append(f"RESEARCH REPORT: {topic.upper()}")
lines.append("=" * 80)
lines.append("")
lines.append(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
lines.append(f"Research Agent System v1.0")
lines.append("")
lines.append("-" * 80)
lines.append("ABSTRACT")
lines.append("-" * 80)
lines.append("")
lines.append(abstract)
lines.append("")
lines.append("-" * 80)
lines.append("TRENDS AND DEVELOPMENTS")
lines.append("-" * 80)
lines.append("")
lines.append(trends)
lines.append("")
lines.append("-" * 80)
lines.append("REFERENCES")
lines.append("-" * 80)
lines.append(references)
lines.append("=" * 80)
lines.append("END OF REPORT")
lines.append("=" * 80)
return '\n'.join(lines)
The report agent implementation demonstrates sophisticated document generation capabilities using local LLMs. The generate_report method orchestrates the entire report generation process, creating each section independently and then assembling them into the final document. The use of run_in_executor allows the synchronous LLM calls to integrate with our async architecture.
The _generate_abstract method creates a high-level overview of the research findings. It combines summaries from multiple sources and instructs the LLM to synthesize them into a coherent abstract. The prompt provides clear guidelines about what the abstract should include and approximately how long it should be. This structured prompting helps ensure consistent quality across different research topics.
The _generate_trends_analysis method performs the most complex synthesis task. It must identify patterns and themes across multiple summaries, extract significant developments, and organize them into a logical narrative. The prompt breaks down this complex task into specific sections, making it easier for the LLM to produce well-structured output. By requesting specific sections like current state, recent developments, and future directions, we ensure comprehensive coverage of the topic.
The _format_references method creates a detailed bibliography with all collected sources. It sorts summaries by relevance score to prioritize the most important sources. Each reference includes comprehensive metadata: title, relevance score, authors, publication date, source, citation count when available, URLs for accessing the document, and a summary of the content. This rich citation format provides users with all the information they need to evaluate and access each source.
The _assemble_report method brings all components together into a professional-looking document. It uses ASCII formatting with clear section headers and dividers to create visual structure. The report includes metadata about when it was generated, making it easy to track different versions. The consistent formatting ensures that reports are easy to read and navigate regardless of the specific research topic.
PUTTING IT ALL TOGETHER: THE COMPLETE SYSTEM
Now that we have implemented all the individual components, we need to integrate them into a cohesive system that users can actually run. This integration involves initializing all agents with appropriate configurations, setting up the local LLM, and providing a user interface for submitting research requests and viewing results.
The system initialization process must handle several important tasks. It must detect available hardware and configure GPU acceleration appropriately. It must load the local language model, which can take significant time and memory for larger models. It must instantiate all agents and wire them together so they can communicate properly. Finally, it must provide a simple interface that allows users to submit research requests and monitor progress.
We will create a main system class that encapsulates all these components and provides a clean API for conducting research. This class will handle initialization, coordinate the agents, and manage the overall workflow. We will also create a simple command-line interface that demonstrates how to use the system.
Here is the complete system implementation:
import asyncio
from typing import Optional, List
import sys
class ResearchAgentSystem:
"""Main system class that coordinates all agents and provides the user interface."""
def __init__(self, model_config: ModelConfig):
print("Initializing Research Agent System...")
print("=" * 70)
self.hardware = HardwareDetector()
self.hardware.optimize_for_inference()
print("\nLoading language model...")
self.llm_manager = LocalLLMManager(self.hardware)
self.llm = self.llm_manager.load_model(model_config)
print("Language model loaded successfully\n")
print("Initializing agents...")
self._initialize_agents()
print("All agents initialized\n")
print("=" * 70)
print("Research Agent System ready!")
print("=" * 70)
print()
def _initialize_agents(self):
"""Initialize all specialized agents."""
self.search_agents = [
ArxivSearchAgent(max_results=15),
SemanticScholarAgent(max_results=15),
DuckDuckGoSearchAgent(max_results=10)
]
self.summary_agent = SummaryAgent(self.llm, max_summary_length=300)
self.report_agent = ReportAgent(self.llm)
self.orchestrator = OrchestrationAgent(
search_agents=self.search_agents,
summary_agent=self.summary_agent,
report_agent=self.report_agent
)
self.conversation_agent = ConversationAgent(self.llm)
async def conduct_research(self, user_message: str) -> str:
"""Conduct research based on a user message."""
print("\n" + "=" * 70)
print("PROCESSING USER REQUEST")
print("=" * 70)
print(f"User: {user_message}\n")
intent = self.conversation_agent.parse_user_request(user_message)
print(f"Extracted Intent:")
print(f" Topic: {intent.get('topic')}")
print(f" Type: {intent.get('request_type')}")
print(f" Requirements: {intent.get('requirements')}")
print()
if intent.get('request_type') == 'question':
response = "I can help you conduct research on any topic. Please specify what you would like me to research."
self.conversation_agent.add_to_history(user_message, response, intent)
return response
topic = intent.get('topic')
if not topic:
response = "I could not identify a clear research topic. Please specify what you would like me to research."
self.conversation_agent.add_to_history(user_message, response, intent)
return response
task = await self.orchestrator.execute_research(
topic=topic,
requirements=intent.get('requirements')
)
response = self.conversation_agent.format_research_results(task)
self.conversation_agent.add_to_history(user_message, response, intent, task.task_id)
if task.status == TaskStatus.COMPLETED and task.final_report:
print("\n" + "=" * 70)
print("FULL RESEARCH REPORT")
print("=" * 70)
print(task.final_report)
return response
def get_conversation_history(self) -> str:
"""Get the conversation history."""
return self.conversation_agent.get_conversation_summary()
def get_welcome_message(self) -> str:
"""Get the welcome message."""
return self.conversation_agent.generate_welcome_message()
The ResearchAgentSystem class provides a clean, high-level interface for the entire research system. The initialization process is carefully orchestrated to ensure all components are properly configured. It starts by detecting hardware capabilities, then loads the language model, and finally initializes all agents. The progress messages keep users informed during the potentially lengthy initialization process.
The _initialize_agents method creates instances of all specialized agents and wires them together. It creates multiple search agents targeting different sources, ensuring comprehensive information gathering. The orchestrator is initialized with references to all other agents, giving it the ability to coordinate their activities. The conversation agent is initialized with the LLM, enabling it to understand and respond to user requests.
The conduct_research method provides the main entry point for research requests. It uses the conversation agent to parse the user message and extract structured intent. Based on the intent type, it either provides a clarifying response or initiates a full research workflow through the orchestrator. The method handles the entire process asynchronously, allowing multiple research tasks to run concurrently if needed.
The method provides detailed progress information throughout the research process, helping users understand what the system is doing. When research completes successfully, it prints the full report to the console, making it easy for users to access the results. The conversation history is maintained across multiple interactions, enabling multi-turn conversations.
Now let us create a command-line interface for the system:
async def main():
"""Main entry point for the research agent system."""
print("\n" * 2)
print("*" * 70)
print("*" + " " * 68 + "*")
print("*" + " AI RESEARCH AGENT SYSTEM".center(68) + "*")
print("*" + " Powered by Local LLMs and Open Source Tools".center(68) + "*")
print("*" + " " * 68 + "*")
print("*" * 70)
print("\n")
print("Recommended Models:")
print(" - Mistral-7B-Instruct (Hugging Face): mistralai/Mistral-7B-Instruct-v0.2")
print(" - Phi-2 (Hugging Face): microsoft/phi-2")
print(" - Llama-2-7B (GGUF): path/to/llama-2-7b.Q4_K_M.gguf")
print()
model_choice = input("Enter model type (1=Mistral HF, 2=Phi-2 HF, 3=GGUF path): ").strip()
if model_choice == "1":
config = ModelConfig(
model_name="mistral-7b-instruct",
model_type="huggingface",
model_path="mistralai/Mistral-7B-Instruct-v0.2",
context_window=4096,
max_new_tokens=512
)
elif model_choice == "2":
config = ModelConfig(
model_name="phi-2",
model_type="huggingface",
model_path="microsoft/phi-2",
context_window=2048,
max_new_tokens=512
)
elif model_choice == "3":
model_path = input("Enter full path to GGUF model file: ").strip()
config = ModelConfig(
model_name="local-gguf",
model_type="gguf",
model_path=model_path,
context_window=4096,
max_new_tokens=512
)
else:
print("Invalid choice. Using Phi-2 as default.")
config = ModelConfig(
model_name="phi-2",
model_type="huggingface",
model_path="microsoft/phi-2",
context_window=2048,
max_new_tokens=512
)
print("\n")
system = ResearchAgentSystem(config)
print(system.get_welcome_message())
print("\n")
while True:
try:
user_input = input("\nYou: ").strip()
if not user_input:
continue
if user_input.lower() in ['exit', 'quit', 'bye']:
print("\nThank you for using the Research Agent System. Goodbye!")
break
if user_input.lower() == 'history':
print("\n" + system.get_conversation_history())
continue
response = await system.conduct_research(user_input)
print(f"\nAssistant: {response}\n")
except KeyboardInterrupt:
print("\n\nInterrupted by user. Exiting...")
break
except Exception as e:
print(f"\nError: {e}")
print("Please try again or type 'exit' to quit.\n")
if __name__ == "__main__":
asyncio.run(main())
The command-line interface provides a simple but functional way to interact with the research system. It starts by presenting a welcome screen with branding and information about the system. It then prompts the user to select a language model, offering several pre-configured options for popular open source models.
The main loop handles user input and dispatches research requests to the system. It supports several special commands: exit/quit/bye to terminate the program, and history to view the conversation history. For all other inputs, it treats them as research requests and processes them through the full research pipeline.
The interface provides clear feedback throughout the process, showing user messages, extracted intents, progress updates, and final results. Error handling ensures that the system continues running even if individual requests fail, allowing users to try different queries without restarting the entire system.
FULL RUNNING EXAMPLE
Here is the complete, production-ready implementation of the AI Research Agent System. This code includes all components integrated into a single, executable program:
import torch
import platform
import sys
from typing import Optional, Dict, Any, List, Literal
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import asyncio
import uuid
import re
import os
try:
from transformers import AutoModelForCausalLM, AutoTokenizer
from llama_index.llms.llama_cpp import LlamaCPP
from llama_index.llms.huggingface import HuggingFaceLLM
import arxiv
import aiohttp
from duckduckgo_search import DDGS
except ImportError as e:
print(f"Missing required dependency: {e}")
print("Please install required packages:")
print("pip install torch transformers llama-index llama-cpp-python arxiv aiohttp duckduckgo-search")
sys.exit(1)
class HardwareDetector:
"""Detects and configures available GPU hardware for LLM inference."""
def __init__(self):
self.device_type = self._detect_device()
self.device = torch.device(self.device_type)
self._print_device_info()
def _detect_device(self) -> str:
"""Determine the best available compute device."""
if torch.cuda.is_available():
return "cuda"
elif hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
return "mps"
else:
return "cpu"
def _print_device_info(self):
"""Print detailed information about the detected device."""
print("=" * 70)
print("HARDWARE DETECTION REPORT")
print("=" * 70)
if self.device_type == "cuda":
gpu_name = torch.cuda.get_device_name(0)
gpu_count = torch.cuda.device_count()
cuda_version = torch.version.cuda
print(f"GPU Platform: NVIDIA CUDA")
print(f"CUDA Version: {cuda_version}")
print(f"GPU Count: {gpu_count}")
print(f"Primary GPU: {gpu_name}")
total_memory = torch.cuda.get_device_properties(0).total_memory
print(f"GPU Memory: {total_memory / 1024**3:.2f} GB")
elif self.device_type == "mps":
print(f"GPU Platform: Apple Metal Performance Shaders")
print(f"System: {platform.platform()}")
print(f"Processor: {platform.processor()}")
print(f"Architecture: {platform.machine()}")
else:
print(f"GPU Platform: None (CPU only)")
print(f"System: {platform.platform()}")
print(f"Processor: {platform.processor()}")
print(f"Warning: No GPU acceleration available. Performance will be limited.")
print("=" * 70)
print()
def get_device_info(self) -> dict:
"""Return detailed information about the compute device."""
info = {
"device_type": self.device_type,
"device": str(self.device),
"platform": platform.platform(),
"processor": platform.processor()
}
if self.device_type == "cuda":
info["gpu_count"] = torch.cuda.device_count()
info["gpu_name"] = torch.cuda.get_device_name(0)
info["cuda_version"] = torch.version.cuda
info["total_memory_gb"] = torch.cuda.get_device_properties(0).total_memory / 1024**3
return info
def optimize_for_inference(self):
"""Apply device-specific optimizations for inference."""
if self.device_type == "cuda":
torch.backends.cudnn.benchmark = True
torch.backends.cuda.matmul.allow_tf32 = True
print("Applied CUDA optimizations for inference")
elif self.device_type == "mps":
print("MPS device ready for inference")
else:
print("CPU device ready for inference")
@dataclass
class ModelConfig:
"""Configuration for a local language model."""
model_name: str
model_type: str
model_path: Optional[str] = None
context_window: int = 4096
max_new_tokens: int = 512
temperature: float = 0.7
top_p: float = 0.9
class LocalLLMManager:
"""Manages local language model loading and inference."""
def __init__(self, hardware_detector: HardwareDetector):
self.hardware = hardware_detector
self.loaded_models: Dict[str, Any] = {}
def load_huggingface_model(self, config: ModelConfig):
"""Load a Hugging Face model for local inference."""
print(f"Loading Hugging Face model: {config.model_name}")
model_path = config.model_path or config.model_name
try:
tokenizer = AutoTokenizer.from_pretrained(model_path)
device_map = None
if self.hardware.device_type == "cuda":
device_map = "auto"
elif self.hardware.device_type == "mps":
device_map = {"": "mps"}
else:
device_map = {"": "cpu"}
model = AutoModelForCausalLM.from_pretrained(
model_path,
device_map=device_map,
torch_dtype=torch.float16 if self.hardware.device_type != "cpu" else torch.float32,
low_cpu_mem_usage=True
)
llm = HuggingFaceLLM(
model=model,
tokenizer=tokenizer,
context_window=config.context_window,
max_new_tokens=config.max_new_tokens,
generate_kwargs={
"temperature": config.temperature,
"top_p": config.top_p,
"do_sample": True
},
device_map=device_map
)
self.loaded_models[config.model_name] = llm
print(f"Successfully loaded {config.model_name}")
return llm
except Exception as e:
print(f"Failed to load Hugging Face model: {e}")
raise
def load_gguf_model(self, config: ModelConfig):
"""Load a GGUF format model using llama-cpp-python."""
print(f"Loading GGUF model: {config.model_name}")
if not config.model_path:
raise ValueError("GGUF models require an explicit model_path")
try:
n_gpu_layers = 0
if self.hardware.device_type == "cuda":
n_gpu_layers = -1
elif self.hardware.device_type == "mps":
n_gpu_layers = 1
llm = LlamaCPP(
model_path=config.model_path,
temperature=config.temperature,
max_new_tokens=config.max_new_tokens,
context_window=config.context_window,
generate_kwargs={"top_p": config.top_p},
model_kwargs={"n_gpu_layers": n_gpu_layers},
verbose=True
)
self.loaded_models[config.model_name] = llm
print(f"Successfully loaded {config.model_name}")
return llm
except Exception as e:
print(f"Failed to load GGUF model: {e}")
raise
def load_model(self, config: ModelConfig):
"""Load a model based on its configuration."""
if config.model_name in self.loaded_models:
print(f"Model {config.model_name} already loaded")
return self.loaded_models[config.model_name]
if config.model_type == "huggingface":
return self.load_huggingface_model(config)
elif config.model_type == "gguf":
return self.load_gguf_model(config)
else:
raise ValueError(f"Unknown model type: {config.model_type}")
def get_model(self, model_name: str) -> Optional[Any]:
"""Retrieve a loaded model by name."""
return self.loaded_models.get(model_name)
def unload_model(self, model_name: str):
"""Unload a model to free memory."""
if model_name in self.loaded_models:
del self.loaded_models[model_name]
if self.hardware.device_type == "cuda":
torch.cuda.empty_cache()
print(f"Unloaded model: {model_name}")
class ArxivSearchAgent:
"""Searches ArXiv for academic papers related to the research topic."""
def __init__(self, max_results: int = 20):
self.max_results = max_results
self.source_name = "ArXiv"
async def search(self, topic: str) -> List[Dict[str, Any]]:
"""Search ArXiv for papers related to the given topic."""
print(f"[ArXiv Agent] Searching for: {topic}")
try:
search = arxiv.Search(
query=topic,
max_results=self.max_results,
sort_by=arxiv.SortCriterion.Relevance
)
results = []
for paper in search.results():
result = {
'title': paper.title,
'authors': [author.name for author in paper.authors],
'published': paper.published.strftime('%Y-%m-%d'),
'updated': paper.updated.strftime('%Y-%m-%d') if paper.updated else None,
'abstract': paper.summary.replace('\n', ' ').strip(),
'url': paper.entry_id,
'pdf_url': paper.pdf_url,
'source': self.source_name,
'categories': paper.categories,
'primary_category': paper.primary_category,
'doi': paper.doi,
'journal_ref': paper.journal_ref,
'comment': paper.comment
}
results.append(result)
print(f"[ArXiv Agent] Found {len(results)} papers")
return results
except Exception as e:
print(f"[ArXiv Agent] Search failed: {e}")
return []
class SemanticScholarAgent:
"""Searches Semantic Scholar for academic publications."""
def __init__(self, max_results: int = 20, api_key: Optional[str] = None):
self.max_results = max_results
self.api_key = api_key
self.base_url = "https://api.semanticscholar.org/graph/v1/paper/search"
self.source_name = "Semantic Scholar"
async def search(self, topic: str) -> List[Dict[str, Any]]:
"""Search Semantic Scholar for papers related to the given topic."""
print(f"[Semantic Scholar Agent] Searching for: {topic}")
try:
params = {
'query': topic,
'limit': self.max_results,
'fields': 'title,authors,year,abstract,url,citationCount,publicationDate,venue,externalIds'
}
headers = {}
if self.api_key:
headers['x-api-key'] = self.api_key
async with aiohttp.ClientSession() as session:
async with session.get(self.base_url, params=params, headers=headers) as response:
if response.status != 200:
print(f"[Semantic Scholar Agent] API returned status {response.status}")
return []
data = await response.json()
results = []
for paper in data.get('data', []):
authors = [author.get('name', 'Unknown') for author in paper.get('authors', [])]
abstract = paper.get('abstract', '')
if not abstract:
abstract = f"Citation count: {paper.get('citationCount', 0)}. Published in {paper.get('venue', 'Unknown venue')}."
result = {
'title': paper.get('title', 'Untitled'),
'authors': authors,
'published': paper.get('publicationDate', paper.get('year', 'Unknown')),
'abstract': abstract,
'url': paper.get('url', ''),
'source': self.source_name,
'citation_count': paper.get('citationCount', 0),
'venue': paper.get('venue', ''),
'external_ids': paper.get('externalIds', {})
}
results.append(result)
print(f"[Semantic Scholar Agent] Found {len(results)} papers")
return results
except Exception as e:
print(f"[Semantic Scholar Agent] Search failed: {e}")
return []
class DuckDuckGoSearchAgent:
"""Searches the web using DuckDuckGo's free search service."""
def __init__(self, max_results: int = 15):
self.max_results = max_results
self.source_name = "Web Search (DuckDuckGo)"
async def search(self, topic: str) -> List[Dict[str, Any]]:
"""Search the web for content related to the given topic."""
print(f"[DuckDuckGo Agent] Searching for: {topic}")
try:
loop = asyncio.get_event_loop()
search_results = await loop.run_in_executor(None, self._perform_search, topic)
results = []
for item in search_results[:self.max_results]:
result = {
'title': item.get('title', 'Untitled'),
'authors': ['Web Source'],
'published': datetime.now().strftime('%Y-%m-%d'),
'abstract': item.get('body', ''),
'url': item.get('href', ''),
'source': self.source_name,
'domain': self._extract_domain(item.get('href', ''))
}
results.append(result)
print(f"[DuckDuckGo Agent] Found {len(results)} results")
return results
except Exception as e:
print(f"[DuckDuckGo Agent] Search failed: {e}")
return []
def _perform_search(self, topic: str) -> List[Dict[str, Any]]:
"""Perform the actual search using DDGS."""
with DDGS() as ddgs:
results = list(ddgs.text(topic, max_results=self.max_results))
return results
def _extract_domain(self, url: str) -> str:
"""Extract the domain name from a URL."""
try:
from urllib.parse import urlparse
parsed = urlparse(url)
return parsed.netloc
except:
return "unknown"
class SummaryAgent:
"""Generates concise summaries of documents using local LLM capabilities."""
def __init__(self, llm, max_summary_length: int = 300):
self.llm = llm
self.max_summary_length = max_summary_length
async def summarize(self, document: Dict[str, Any]) -> Dict[str, Any]:
"""Generate a summary of the given document."""
title = document.get('title', 'Untitled')
print(f"[Summary Agent] Summarizing: {title[:60]}...")
try:
loop = asyncio.get_event_loop()
summary_text = await loop.run_in_executor(None, self._generate_summary, document)
relevance_score = await loop.run_in_executor(None, self._calculate_relevance, document, summary_text)
summary = {
'original_document': document,
'summary_text': summary_text,
'relevance_score': relevance_score,
'title': title,
'source': document.get('source', 'Unknown'),
'url': document.get('url', ''),
'authors': document.get('authors', []),
'published': document.get('published', 'Unknown')
}
return summary
except Exception as e:
print(f"[Summary Agent] Summarization failed for '{title}': {e}")
return {
'original_document': document,
'summary_text': document.get('abstract', '')[:self.max_summary_length],
'relevance_score': 5.0,
'title': title,
'source': document.get('source', 'Unknown'),
'url': document.get('url', ''),
'authors': document.get('authors', []),
'published': document.get('published', 'Unknown'),
'error': str(e)
}
def _generate_summary(self, document: Dict[str, Any]) -> str:
"""Generate a summary using the local LLM."""
title = document.get('title', 'Untitled')
abstract = document.get('abstract', '')
source = document.get('source', 'Unknown')
if source == "ArXiv" or source == "Semantic Scholar":
prompt = f"""Summarize the following academic paper in approximately {self.max_summary_length} words. Focus on:
- The main research question or problem addressed
- The methodology or approach used
- Key findings or results
- Significance and implications
Title: {title}
Abstract: {abstract}
Provide a concise summary:""" else: prompt = f"""Summarize the following web content in approximately {self.max_summary_length} words. Focus on:
- The main topic or subject
- Key points or arguments
- Important facts or data
- Practical implications or applications
Title: {title}
Content: {abstract}
Provide a concise summary:"""
response = self.llm.complete(prompt)
summary = response.text.strip()
if len(summary.split()) > self.max_summary_length * 1.5:
words = summary.split()
summary = ' '.join(words[:self.max_summary_length]) + '...'
return summary
def _calculate_relevance(self, document: Dict[str, Any], summary: str) -> float:
"""Calculate a relevance score for the document."""
prompt = f"""Rate the relevance and quality of this research document on a scale from 1 to 10, where:
1-3: Low relevance or quality (tangentially related, poor quality, or outdated) 4-6: Moderate relevance or quality (somewhat related, acceptable quality) 7-9: High relevance or quality (directly related, high quality, authoritative) 10: Exceptional relevance and quality (highly relevant, seminal work, authoritative source)
Title: {document.get('title', 'Untitled')} Source: {document.get('source', 'Unknown')} Summary: {summary}
Provide only a single number from 1 to 10 as your rating:"""
try:
response = self.llm.complete(prompt)
rating_text = response.text.strip()
numbers = re.findall(r'\d+', rating_text)
if numbers:
rating = int(numbers[0])
rating = max(1, min(10, rating))
return float(rating)
else:
return 5.0
except:
return 5.0
class ReportAgent:
"""Generates comprehensive research reports from collected information."""
def __init__(self, llm):
self.llm = llm
async def generate_report(self, topic: str, requirements: Optional[str],
search_results: List[Dict[str, Any]],
summaries: List[Dict[str, Any]]) -> str:
"""Generate a complete research report."""
print(f"[Report Agent] Generating report for: {topic}")
loop = asyncio.get_event_loop()
abstract = await loop.run_in_executor(None, self._generate_abstract, topic, requirements, summaries)
trends = await loop.run_in_executor(None, self._generate_trends_analysis, topic, summaries)
references = self._format_references(summaries)
report = self._assemble_report(topic, abstract, trends, references)
print(f"[Report Agent] Report generated ({len(report)} characters)")
return report
def _generate_abstract(self, topic: str, requirements: Optional[str],
summaries: List[Dict[str, Any]]) -> str:
"""Generate an abstract for the research report."""
summary_texts = [s['summary_text'] for s in summaries[:10]]
combined_summaries = '\n\n'.join(summary_texts)
req_text = f"\nSpecific Requirements: {requirements}" if requirements else ""
prompt = f"""Based on the following research summaries, write a comprehensive abstract for a research report on "{topic}".{req_text}
The abstract should:
- Provide an overview of the current state of research on this topic
- Highlight the most significant findings and developments
- Indicate the scope and coverage of the research
- Be approximately 200-300 words
Research Summaries: {combined_summaries}
Write the abstract:"""
response = self.llm.complete(prompt)
return response.text.strip()
def _generate_trends_analysis(self, topic: str, summaries: List[Dict[str, Any]]) -> str:
"""Generate an analysis of trends and developments."""
summary_texts = [s['summary_text'] for s in summaries]
combined_summaries = '\n\n'.join(summary_texts)
prompt = f"""Based on the following research summaries about "{topic}", identify and describe the major trends, developments, and themes.
Organize your analysis into clear sections covering:
- Current State: What is the current state of this field or topic?
- Recent Developments: What are the most significant recent advances or changes?
- Key Themes: What are the recurring themes or focus areas across the research?
- Future Directions: What appear to be emerging trends or future research directions?
- Challenges and Opportunities: What challenges and opportunities are identified?
Provide a comprehensive analysis of approximately 500-800 words.
Research Summaries: {combined_summaries}
Write the trends and developments analysis:"""
response = self.llm.complete(prompt)
return response.text.strip()
def _format_references(self, summaries: List[Dict[str, Any]]) -> str:
"""Format the references section with detailed citations."""
sorted_summaries = sorted(summaries, key=lambda x: x.get('relevance_score', 0), reverse=True)
lines = []
for idx, summary in enumerate(sorted_summaries, 1):
doc = summary['original_document']
lines.append(f"\n[{idx}] {summary['title']}")
lines.append(f" Relevance: {summary['relevance_score']:.1f}/10")
authors = doc.get('authors', [])
if authors:
author_str = ', '.join(authors[:3])
if len(authors) > 3:
author_str += f" et al. ({len(authors)} authors total)"
lines.append(f" Authors: {author_str}")
published = doc.get('published', 'Unknown')
lines.append(f" Published: {published}")
source = doc.get('source', 'Unknown')
lines.append(f" Source: {source}")
if 'citation_count' in doc:
lines.append(f" Citations: {doc['citation_count']}")
url = doc.get('url', '')
if url:
lines.append(f" URL: {url}")
pdf_url = doc.get('pdf_url', '')
if pdf_url and pdf_url != url:
lines.append(f" PDF: {pdf_url}")
summary_text = summary['summary_text']
if len(summary_text) > 400:
summary_text = summary_text[:397] + '...'
lines.append(f" Summary: {summary_text}")
lines.append("")
return '\n'.join(lines)
def _assemble_report(self, topic: str, abstract: str, trends: str, references: str) -> str:
"""Assemble all components into the final report."""
lines = []
lines.append("=" * 80)
lines.append(f"RESEARCH REPORT: {topic.upper()}")
lines.append("=" * 80)
lines.append("")
lines.append(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
lines.append(f"Research Agent System v1.0")
lines.append("")
lines.append("-" * 80)
lines.append("ABSTRACT")
lines.append("-" * 80)
lines.append("")
lines.append(abstract)
lines.append("")
lines.append("-" * 80)
lines.append("TRENDS AND DEVELOPMENTS")
lines.append("-" * 80)
lines.append("")
lines.append(trends)
lines.append("")
lines.append("-" * 80)
lines.append("REFERENCES")
lines.append("-" * 80)
lines.append(references)
lines.append("=" * 80)
lines.append("END OF REPORT")
lines.append("=" * 80)
return '\n'.join(lines)
class TaskStatus(Enum):
"""Enumeration of possible task states."""
PENDING = "pending"
SEARCHING = "searching"
SUMMARIZING = "summarizing"
GENERATING_REPORT = "generating_report"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class ResearchTask:
"""Represents a single research task with its state and results."""
task_id: str
topic: str
requirements: Optional[str] = None
status: TaskStatus = TaskStatus.PENDING
created_at: datetime = field(default_factory=datetime.now)
search_results: List[Dict[str, Any]] = field(default_factory=list)
summaries: List[Dict[str, Any]] = field(default_factory=list)
final_report: Optional[str] = None
error_message: Optional[str] = None
progress_log: List[str] = field(default_factory=list)
def log_progress(self, message: str):
"""Add a progress message to the task log."""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
log_entry = f"[{timestamp}] {message}"
self.progress_log.append(log_entry)
print(log_entry)
class OrchestrationAgent:
"""Coordinates the activities of all specialized agents."""
def __init__(self, search_agents: List, summary_agent, report_agent):
self.search_agents = search_agents
self.summary_agent = summary_agent
self.report_agent = report_agent
self.active_tasks: Dict[str, ResearchTask] = {}
self.completed_tasks: Dict[str, ResearchTask] = {}
async def execute_research(self, topic: str, requirements: Optional[str] = None) -> ResearchTask:
"""Execute a complete research workflow for the given topic."""
task_id = str(uuid.uuid4())
task = ResearchTask(task_id=task_id, topic=topic, requirements=requirements)
self.active_tasks[task_id] = task
task.log_progress(f"Starting research on topic: {topic}")
try:
await self._search_phase(task)
await self._summarization_phase(task)
await self._report_generation_phase(task)
task.status = TaskStatus.COMPLETED
task.log_progress("Research task completed successfully")
self.completed_tasks[task_id] = task
del self.active_tasks[task_id]
except Exception as e:
task.status = TaskStatus.FAILED
task.error_message = str(e)
task.log_progress(f"Research task failed: {e}")
self.completed_tasks[task_id] = task
if task_id in self.active_tasks:
del self.active_tasks[task_id]
return task
async def _search_phase(self, task: ResearchTask):
"""Execute parallel search across multiple agents."""
task.status = TaskStatus.SEARCHING
task.log_progress(f"Starting parallel search with {len(self.search_agents)} agents")
search_tasks = [agent.search(task.topic) for agent in self.search_agents]
search_results = await asyncio.gather(*search_tasks, return_exceptions=True)
valid_results = []
for idx, result in enumerate(search_results):
if isinstance(result, Exception):
task.log_progress(f"Search agent {idx} failed: {result}")
else:
valid_results.extend(result)
task.log_progress(f"Search agent {idx} returned {len(result)} results")
task.search_results = valid_results
task.log_progress(f"Search phase completed with {len(valid_results)} total results")
if len(valid_results) == 0:
raise ValueError("No search results found from any agent")
async def _summarization_phase(self, task: ResearchTask):
"""Summarize all collected documents."""
task.status = TaskStatus.SUMMARIZING
task.log_progress(f"Starting summarization of {len(task.search_results)} documents")
summary_tasks = [self.summary_agent.summarize(doc) for doc in task.search_results]
summaries = await asyncio.gather(*summary_tasks, return_exceptions=True)
valid_summaries = [s for s in summaries if not isinstance(s, Exception)]
task.summaries = valid_summaries
task.log_progress(f"Summarization phase completed with {len(valid_summaries)} summaries")
if len(valid_summaries) == 0:
raise ValueError("No summaries generated successfully")
async def _report_generation_phase(self, task: ResearchTask):
"""Generate the final research report."""
task.status = TaskStatus.GENERATING_REPORT
task.log_progress("Starting report generation")
report = await self.report_agent.generate_report(
topic=task.topic,
requirements=task.requirements,
search_results=task.search_results,
summaries=task.summaries
)
task.final_report = report
task.log_progress(f"Report generation completed ({len(report)} characters)")
def get_task_status(self, task_id: str) -> Optional[ResearchTask]:
"""Retrieve the current status of a research task."""
if task_id in self.active_tasks:
return self.active_tasks[task_id]
return self.completed_tasks.get(task_id)
def get_all_tasks(self) -> Dict[str, ResearchTask]:
"""Retrieve all tasks (active and completed)."""
all_tasks = {}
all_tasks.update(self.active_tasks)
all_tasks.update(self.completed_tasks)
return all_tasks
@dataclass
class ConversationTurn:
"""Represents a single turn in the conversation."""
user_message: str
agent_response: str
timestamp: datetime
extracted_intent: Optional[Dict[str, Any]] = None
task_id: Optional[str] = None
class ConversationAgent:
"""Manages natural language interactions with users."""
def __init__(self, llm):
self.llm = llm
self.conversation_history: List[ConversationTurn] = []
def parse_user_request(self, user_message: str) -> Dict[str, Any]:
"""Extract structured intent from natural language user input."""
prompt = f"""Analyze the following user request and extract structured information about what research they want.
User Request: {user_message}
Extract the following information:
- Research Topic: The main subject the user wants to research
- Request Type: Either 'new_research', 'refine_research', or 'question'
- Specific Requirements: Any particular constraints or focus areas mentioned
- Reference to Previous Work: Whether this refers to earlier research (yes/no)
Provide your analysis in exactly this format: TOPIC: [extracted topic or 'none'] TYPE: [request type] REQUIREMENTS: [specific requirements or 'none'] PREVIOUS_REFERENCE: [yes or no]
Example: TOPIC: machine learning in healthcare TYPE: new_research REQUIREMENTS: focus on diagnostic applications PREVIOUS_REFERENCE: no """
response = self.llm.complete(prompt).text
intent = self._parse_llm_response(response)
if intent.get('previous_reference') and self.conversation_history:
previous_topic = self._get_previous_topic()
if previous_topic:
intent['previous_topic'] = previous_topic
if intent.get('topic') == 'none':
intent['topic'] = previous_topic
return intent
def _parse_llm_response(self, response: str) -> Dict[str, Any]:
"""Parse the structured response from the LLM."""
intent = {}
topic_match = re.search(r'TOPIC:\s*(.+?)(?:\n|$)', response, re.IGNORECASE)
if topic_match:
topic = topic_match.group(1).strip()
intent['topic'] = None if topic.lower() == 'none' else topic
type_match = re.search(r'TYPE:\s*(.+?)(?:\n|$)', response, re.IGNORECASE)
if type_match:
intent['request_type'] = type_match.group(1).strip()
req_match = re.search(r'REQUIREMENTS:\s*(.+?)(?:\n|$)', response, re.IGNORECASE)
if req_match:
reqs = req_match.group(1).strip()
intent['requirements'] = None if reqs.lower() == 'none' else reqs
prev_match = re.search(r'PREVIOUS_REFERENCE:\s*(.+?)(?:\n|$)', response, re.IGNORECASE)
if prev_match:
intent['previous_reference'] = prev_match.group(1).strip().lower() == 'yes'
return intent
def _get_previous_topic(self) -> Optional[str]:
"""Retrieve the topic from the most recent research request."""
for turn in reversed(self.conversation_history):
if turn.extracted_intent and turn.extracted_intent.get('topic'):
return turn.extracted_intent['topic']
return None
def format_research_results(self, task: ResearchTask) -> str:
"""Format research results into a user-friendly response."""
if task.status == TaskStatus.FAILED:
return f"""I apologize, but the research task encountered an error and could not be completed.
Error Details: {task.error_message}
Progress Log: {chr(10).join(task.progress_log[-5:])}
Please try again with a different topic or contact support if the issue persists."""
if task.status != TaskStatus.COMPLETED:
return f"""Your research task is currently in progress.
Status: {task.status.value} Topic: {task.topic} Started: {task.created_at.strftime('%Y-%m-%d %H:%M:%S')}
Recent Progress: {chr(10).join(task.progress_log[-3:])}
Please check back in a few moments for the completed report."""
lines = []
lines.append("=" * 70)
lines.append(f"RESEARCH REPORT: {task.topic.upper()}")
lines.append("=" * 70)
lines.append("")
lines.append(f"Research completed: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
lines.append(f"Sources analyzed: {len(task.search_results)}")
lines.append(f"Summaries generated: {len(task.summaries)}")
lines.append("")
lines.append("EXECUTIVE SUMMARY")
lines.append("-" * 70)
if task.final_report:
report_lines = task.final_report.split('\n')
abstract_start = -1
abstract_end = -1
for i, line in enumerate(report_lines):
if 'ABSTRACT' in line.upper():
abstract_start = i
elif abstract_start >= 0 and (line.startswith('##') or line.startswith('TRENDS') or line.startswith('REFERENCES')):
abstract_end = i
break
if abstract_start >= 0:
if abstract_end < 0:
abstract_end = min(abstract_start + 15, len(report_lines))
for line in report_lines[abstract_start:abstract_end]:
if line.strip():
lines.append(line)
lines.append("")
lines.append("The full research report is available below.")
lines.append("It includes detailed analysis, trends, and comprehensive references.")
lines.append("")
lines.append("=" * 70)
return '\n'.join(lines)
def generate_welcome_message(self) -> str:
"""Generate a welcome message for new users."""
return """Welcome to the AI Research Agent System!
I am your research assistant, powered by local language models and open source tools. I can help you conduct comprehensive research on any topic by:
- Searching academic databases (ArXiv, Semantic Scholar)
- Finding relevant web resources using DuckDuckGo
- Summarizing research papers and articles
- Generating comprehensive research reports with citations
To get started, simply tell me what topic you would like to research. For example:
- "Research recent developments in quantum computing"
- "Find information about renewable energy technologies"
- "Investigate advances in natural language processing"
You can also refine previous research or ask follow-up questions. What would you like to research today?"""
def add_to_history(self, user_message: str, agent_response: str,
intent: Dict[str, Any], task_id: Optional[str] = None):
"""Record a conversation turn in the history."""
turn = ConversationTurn(
user_message=user_message,
agent_response=agent_response,
timestamp=datetime.now(),
extracted_intent=intent,
task_id=task_id
)
self.conversation_history.append(turn)
def get_conversation_summary(self) -> str:
"""Generate a summary of the conversation history."""
if not self.conversation_history:
return "No conversation history available."
lines = []
lines.append("CONVERSATION HISTORY")
lines.append("=" * 70)
for idx, turn in enumerate(self.conversation_history, 1):
lines.append(f"\nTurn {idx} - {turn.timestamp.strftime('%Y-%m-%d %H:%M:%S')}")
lines.append(f"User: {turn.user_message[:100]}...")
if turn.extracted_intent and turn.extracted_intent.get('topic'):
lines.append(f"Topic: {turn.extracted_intent['topic']}")
if turn.task_id:
lines.append(f"Task ID: {turn.task_id}")
return '\n'.join(lines)
class ResearchAgentSystem:
"""Main system class that coordinates all agents and provides the user interface."""
def __init__(self, model_config: ModelConfig):
print("Initializing Research Agent System...")
print("=" * 70)
self.hardware = HardwareDetector()
self.hardware.optimize_for_inference()
print("\nLoading language model...")
self.llm_manager = LocalLLMManager(self.hardware)
self.llm = self.llm_manager.load_model(model_config)
print("Language model loaded successfully\n")
print("Initializing agents...")
self._initialize_agents()
print("All agents initialized\n")
print("=" * 70)
print("Research Agent System ready!")
print("=" * 70)
print()
def _initialize_agents(self):
"""Initialize all specialized agents."""
self.search_agents = [
ArxivSearchAgent(max_results=15),
SemanticScholarAgent(max_results=15),
DuckDuckGoSearchAgent(max_results=10)
]
self.summary_agent = SummaryAgent(self.llm, max_summary_length=300)
self.report_agent = ReportAgent(self.llm)
self.orchestrator = OrchestrationAgent(
search_agents=self.search_agents,
summary_agent=self.summary_agent,
report_agent=self.report_agent
)
self.conversation_agent = ConversationAgent(self.llm)
async def conduct_research(self, user_message: str) -> str:
"""Conduct research based on a user message."""
print("\n" + "=" * 70)
print("PROCESSING USER REQUEST")
print("=" * 70)
print(f"User: {user_message}\n")
intent = self.conversation_agent.parse_user_request(user_message)
print(f"Extracted Intent:")
print(f" Topic: {intent.get('topic')}")
print(f" Type: {intent.get('request_type')}")
print(f" Requirements: {intent.get('requirements')}")
print()
if intent.get('request_type') == 'question':
response = "I can help you conduct research on any topic. Please specify what you would like me to research."
self.conversation_agent.add_to_history(user_message, response, intent)
return response
topic = intent.get('topic')
if not topic:
response = "I could not identify a clear research topic. Please specify what you would like me to research."
self.conversation_agent.add_to_history(user_message, response, intent)
return response
task = await self.orchestrator.execute_research(topic=topic, requirements=intent.get('requirements'))
response = self.conversation_agent.format_research_results(task)
self.conversation_agent.add_to_history(user_message, response, intent, task.task_id)
if task.status == TaskStatus.COMPLETED and task.final_report:
print("\n" + "=" * 70)
print("FULL RESEARCH REPORT")
print("=" * 70)
print(task.final_report)
return response
def get_conversation_history(self) -> str:
"""Get the conversation history."""
return self.conversation_agent.get_conversation_summary()
def get_welcome_message(self) -> str:
"""Get the welcome message."""
return self.conversation_agent.generate_welcome_message()
async def main():
"""Main entry point for the research agent system."""
print("\n" * 2)
print("*" * 70)
print("*" + " " * 68 + "*")
print("*" + " AI RESEARCH AGENT SYSTEM".center(68) + "*")
print("*" + " Powered by Local LLMs and Open Source Tools".center(68) + "*")
print("*" + " " * 68 + "*")
print("*" * 70)
print("\n")
print("Recommended Models:")
print(" - Mistral-7B-Instruct (Hugging Face): mistralai/Mistral-7B-Instruct-v0.2")
print(" - Phi-2 (Hugging Face): microsoft/phi-2")
print(" - Llama-2-7B (GGUF): path/to/llama-2-7b.Q4_K_M.gguf")
print()
model_choice = input("Enter model type (1=Mistral HF, 2=Phi-2 HF, 3=GGUF path): ").strip()
if model_choice == "1":
config = ModelConfig(
model_name="mistral-7b-instruct",
model_type="huggingface",
model_path="mistralai/Mistral-7B-Instruct-v0.2",
context_window=4096,
max_new_tokens=512
)
elif model_choice == "2":
config = ModelConfig(
model_name="phi-2",
model_type="huggingface",
model_path="microsoft/phi-2",
context_window=2048,
max_new_tokens=512
)
elif model_choice == "3":
model_path = input("Enter full path to GGUF model file: ").strip()
config = ModelConfig(
model_name="local-gguf",
model_type="gguf",
model_path=model_path,
context_window=4096,
max_new_tokens=512
)
else:
print("Invalid choice. Using Phi-2 as default.")
config = ModelConfig(
model_name="phi-2",
model_type="huggingface",
model_path="microsoft/phi-2",
context_window=2048,
max_new_tokens=512
)
print("\n")
system = ResearchAgentSystem(config)
print(system.get_welcome_message())
print("\n")
while True:
try:
user_input = input("\nYou: ").strip()
if not user_input:
continue
if user_input.lower() in ['exit', 'quit', 'bye']:
print("\nThank you for using the Research Agent System. Goodbye!")
break
if user_input.lower() == 'history':
print("\n" + system.get_conversation_history())
continue
response = await system.conduct_research(user_input)
print(f"\nAssistant: {response}\n")
except KeyboardInterrupt:
print("\n\nInterrupted by user. Exiting...")
break
except Exception as e:
print(f"\nError: {e}")
print("Please try again or type 'exit' to quit.\n")
if __name__ == "__main__":
asyncio.run(main())
CONCLUSION AND FUTURE ENHANCEMENTS
We have successfully built a comprehensive multi-agent research system using LlamaIndex and entirely free, open source tools. This system demonstrates the power of coordinating specialized agents to accomplish complex research tasks that would be time-consuming and difficult for humans to perform manually.
The system architecture provides several key advantages. The multi-agent design allows for parallel processing of search queries across different sources, dramatically reducing the time required to gather comprehensive information. The use of local language models ensures complete data privacy and eliminates ongoing API costs, making the system economical for heavy usage. The GPU acceleration support for both Apple MPS and NVIDIA CUDA ensures optimal performance across different hardware platforms.
The implementation demonstrates production-ready code with comprehensive error handling, progress logging, and graceful degradation when individual components fail. The system can handle a wide variety of research topics and produces professional-quality reports with proper citations and relevance scoring.
Future enhancements could include support for additional search sources such as PubMed for medical research or IEEE Xplore for engineering topics. The system could be extended to support multiple languages, allowing research in non-English sources. Advanced features like automatic fact-checking, citation verification, and plagiarism detection could further improve the quality of generated reports. The conversation agent could be enhanced with more sophisticated dialogue management to handle complex multi-turn refinement conversations.
The modular architecture makes it straightforward to add these enhancements without disrupting existing functionality. Each agent operates independently, so new agents can be added or existing ones can be replaced with minimal impact on the overall system. This extensibility ensures that the research agent system can evolve to meet changing requirements and take advantage of new technologies as they become available.
