# Project export: CliniSearch This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: UC Berkeley AI Hackathon 2025 - Tagline: CliniSearch: Augmenting Medical Intelligence - Devpost: https://devpost.com/software/corpus-medica - GitHub: https://github.com/Kush614/CliniSearch - Team: 1 GitHub contributor(s) — Kush Ise (7 commits) ## Devpost submission (written by the team) ### Inspiration In today's medical landscape, the volume of new research, clinical data, and patient information is growing at an exponential rate. We were inspired by the daily challenge faced by clinicians and researchers: how to quickly and accurately access and synthesize this vast sea of data to make informed decisions. A conversation with a radiology resident highlighted a specific pain point: the time-consuming process of cross-referencing a patient's imaging findings with their clinical history and the latest peer-reviewed literature. This sparked the core idea for CliniSearch: an AI agent that could not only "read" but also "see," acting as an intelligent assistant to bridge the gap between diverse medical data types. ### What it does CliniSearch is a powerful, multimodal AI agent designed to be a "second brain" for medical professionals. Its functionality is split into two primary tools: Medical RAG Q&A: Users can ask complex medical questions and receive synthesized, evidence-based answers. The agent retrieves information in real-time from three distinct sources: General Web Search: For broad context and general knowledge. PubMed: For access to peer-reviewed scientific literature. Uploaded Documents: Users can upload their own PDFs (research papers, reports) to create a private, searchable knowledge base for highly specific queries. Medical RAG Q&A: Users can ask complex medical questions and receive synthesized, evidence-based answers. The agent retrieves information in real-time from three distinct sources: General Web Search: For broad context and general knowledge. PubMed: For access to peer-reviewed scientific literature. Uploaded Documents: Users can upload their own PDFs (research papers, reports) to create a private, searchable knowledge base for highly specific queries. Radiology Image Analysis: This tool is designed specifically for radiologists. A user can: Upload an anonymized medical image (e.g., an X-ray). Optionally upload a related context document (e.g., a patient's clinical notes). The AI then analyzes the image, providing a preliminary description of findings and potential differential diagnoses, using the context from the uploaded text to deliver a more holistic and relevant analysis. Radiology Image Analysis: This tool is designed specifically for radiologists. A user can: Upload an anonymized medical image (e.g., an X-ray). Optionally upload a related context document (e.g., a patient's clinical notes). The AI then analyzes the image, providing a preliminary description of findings and potential differential diagnoses, using the context from the uploaded text to deliver a more holistic and relevant analysis. All outputs are clearly sourced with links, ensuring transparency and allowing for easy verification. ### How we built it We architected CliniSearch as a modular, modern web application with a clear separation of concerns. Frontend: The user interface is built with Streamlit, chosen for its ability to rapidly create interactive and data-centric web apps in Python. We used tabs to separate the two main functionalities and custom styling to ensure a clean, professional UI. Frontend: The user interface is built with Streamlit, chosen for its ability to rapidly create interactive and data-centric web apps in Python. We used tabs to separate the two main functionalities and custom styling to ensure a clean, professional UI. Backend & Orchestration: The core logic resides within the Streamlit app (app.py), which orchestrates calls to various backend components. We used Python's asyncio library to handle concurrent requests to our data sources, improving efficiency. Backend & Orchestration: The core logic resides within the Streamlit app (app.py), which orchestrates calls to various backend components. We used Python's asyncio library to handle concurrent requests to our data sources, improving efficiency. Tool Servers (MCP): We built two lightweight, independent tool servers using FastAPI to handle data retrieval. This "Model Context Protocol" approach makes the system modular and scalable. One server uses the duckduckgo-search library for web searches. The other uses BioPython to interact with the NCBI Entrez API for PubMed searches. Tool Servers (MCP): We built two lightweight, independent tool servers using FastAPI to handle data retrieval. This "Model Context Protocol" approach makes the system modular and scalable. One server uses the duckduckgo-search library for web searches. The other uses BioPython to interact with the NCBI Entrez API for PubMed searches. AI & RAG Pipeline: This is the heart of our project. LLMs: We leveraged the Google Gemini API, specifically gemini-1.5-flash-latest, for its powerful text synthesis and state-of-the-art multimodal (vision) capabilities. Embeddings: For our local RAG feature, we used a sentence-transformers model to generate vector embeddings from PDF text. Vector Store: We implemented an in-memory vector database using FAISS from Meta AI, which allows for incredibly fast semantic similarity searches on the uploaded documents. AI & RAG Pipeline: This is the heart of our project. LLMs: We leveraged the Google Gemini API, specifically gemini-1.5-flash-latest, for its powerful text synthesis and state-of-the-art multimodal (vision) capabilities. Embeddings: For our local RAG feature, we used a sentence-transformers model to generate vector embeddings from PDF text. Vector Store: We implemented an in-memory vector database using FAISS from Meta AI, which allows for incredibly fast semantic similarity searches on the uploaded documents. GCP Readiness: The project was built with an eye towards production, with a structure ready for deployment on Google Cloud Platform services like Cloud Run (for the app and servers) and Vertex AI (for embeddings and vector search). GCP Readiness: The project was built with an eye towards production, with a structure ready for deployment on Google Cloud Platform services like Cloud Run (for the app and servers) and Vertex AI (for embeddings and vector search). ### Challenges we ran into Real-time API Latency: Integrating multiple real-time APIs (Web Search, PubMed, Gemini) presented a challenge. Initial user requests were slow. We mitigated this by using asyncio to run the data retrieval tasks concurrently, significantly speeding up the process. Real-time API Latency: Integrating multiple real-time APIs (Web Search, PubMed, Gemini) presented a challenge. Initial user requests were slow. We mitigated this by using asyncio to run the data retrieval tasks concurrently, significantly speeding up the process. Context Window Management: LLMs have finite context windows. Combining information from three different RAG sources could easily exceed this limit. Our solution was to process each source independently and present separate, synthesized answers, which not only solved the technical problem but also improved the clarity and traceability of the output for the user. Context Window Management: LLMs have finite context windows. Combining information from three different RAG sources could easily exceed this limit. Our solution was to process each source independently and present separate, synthesized answers, which not only solved the technical problem but also improved the clarity and traceability of the output for the user. UI State and Interactivity: Streamlit reruns the script on every interaction, which made managing the state of our vector store and file uploads tricky. We solved this by effectively using st.session_state to persist data across reruns and implementing logic to ensure PDFs were only processed once upon upload. UI State and Interactivity: Streamlit reruns the script on every interaction, which made managing the state of our vector store and file uploads tricky. We solved this by effectively using st.session_state to persist data across reruns and implementing logic to ensure PDFs were only processed once upon upload. ### Accomplishments we're proud of True Multimodality: We're incredibly proud of the Radiology Analysis tab. It's not just an image-to-text model; it's a system that fuses visual analysis with contextual text-based RAG, which we believe is a significant step towards creating truly useful clinical AI assistants. True Multimodality: We're incredibly proud of the Radiology Analysis tab. It's not just an image-to-text model; it's a system that fuses visual analysis with contextual text-based RAG, which we believe is a significant step towards creating truly useful clinical AI assistants. Modular and Scalable Design: By separating our data retrieval into microservice-like MCP servers, we've built a system that is easy to maintain and extend. Adding a new data source would be as simple as building another small FastAPI server. Modular and Scalable Design: By separating our data retrieval into microservice-like MCP servers, we've built a system that is easy to maintain and extend. Adding a new data source would be as simple as building another small FastAPI server. Delivering a Polished UX: Despite the technical complexity on the backend, we managed to create a clean, intuitive, and professional-looking user interface that is genuinely usable. Delivering a Polished UX: Despite the technical complexity on the backend, we managed to create a clean, intuitive, and professional-looking user interface that is genuinely usable. ### What we learned The Power of RAG: We learned firsthand how Retrieval-Augmented Generation can ground LLMs in factual, real-time, or private data, drastically reducing hallucinations and increasing the reliability of their outputs. The Nuances of Prompt Engineering: Crafting effective prompts is an art. We learned how to structure prompts to instruct the LLM to use only the provided context, to cite sources, and to tailor its response for a specific audience (like a medical professional). Full-Stack Python Development: This project was a deep dive into the modern Python ecosystem, from backend APIs with FastAPI to interactive web UIs with Streamlit and advanced AI/ML libraries like FAISS and Sentence Transformers. ### What's next CliniSearch is a powerful proof-of-concept with immense potential for growth. Our next steps would include: Deployment on GCP:6 -x Moving the application and its components to Google Cloud Platform to make it scalable, reliable, and accessible. This would involve using Cloud Run, Vertex AI Vector Search, and Document AI for more robust PDF parsing. Enhanced Conversational Memory: Implementing a more sophisticated chat history management system (e.g., using Firestore) to allow for meaningful follow-up questions. Deeper EMR/RIS Integration: Developing secure integrations with hospital systems (like EMRs) to automatically pull relevant patient context, further enhancing the AI's utility. Model Evaluation and Fine-Tuning: Rigorously evaluating the accuracy of different models (including the Claude family) and potentially fine-tuning a model on a specific medical domain for even higher accuracy and reliability. ## README (from the GitHub repository) # CliniSearch: A Multimodal Medical Research & Radiology Assistant **CliniSearch** is an advanced, multimodal AI agent developed for the UCB Hackathon. It is designed to be a powerful assistant for medical professionals, particularly radiologists, by streamlining clinical research and providing AI-powered preliminary image analysis. The agent integrates multiple state-of-the-art technologies, including premium LLMs (Google Gemini), a multi-source RAG pipeline, and a user-friendly web interface built with Streamlit. # Video Demo: [![CliniSearch AI Agent Demo](https://youtu.be/Q3a4GuCqoKQ/maxresdefault.jpg)](https://www.youtube.com/watch?v=Q3a4GuCqoKQ"CliniSearch AI Agent Demo") image image image image image --- ## 🚀 Application & Real-World Benefit In the fast-paced medical field, professionals face the dual challenges of information overload and time scarcity. Spectra AI is designed to address these critical issues directly. * **For Clinicians & Researchers:** It acts as an intelligent research assistant, capable of querying real-time web data, peer-reviewed PubMed articles, and user-uploaded documents (like research papers or reports). By providing synthesized, source-cited answers from these distinct domains, it dramatically accelerates literature reviews, deepens contextual understanding, and helps identify research gaps. * **For Radiologists:** The "Radiology Image Analysis" tab offers a cutting-edge tool for decision support. A radiologist can upload a medical image (e.g., an X-ray, CT scan) and receive a preliminary analysis from a multimodal AI. By combining this visual analysis with contextual information from uploaded patient reports, Spectra AI can help identify potential abnormalities, suggest differential diagnoses, and reduce cognitive load, acting as a "second pair of eyes" to enhance diagnostic confidence and efficiency. --- ## ✨ Features * **User-Friendly Web Interface:** A clean, tabbed UI built with Streamlit separates the text-based RAG Q&A from the specialized Radiology Image Analysis tool. * **Multi-Source RAG Pipeline:** * Dynamically queries **Web Search** (via DuckDuckGo) and **PubMed** (via NCBI Entrez) for real-time information. * Allows users to upload their own **PDF documents**, creating a private, searchable knowledge base for highly contextualized answers. * **Multimodal Radiology Analysis:** * Leverages **Google Gemini Pro Vision** to analyze uploaded medical images. * **Context-Aware Analysis:** Uniquely combines image analysis with a RAG search of user-uploaded PDFs (e.g., patient reports), providing a holistic preliminary assessment. * **High-Quality AI Models:** Powered by the **Google Gemini API** (`gemini-1.5-flash`) for state-of-the-art text synthesis and multimodal understanding. * **Structured & Sourced Outputs:** All answers are presented in a clean, readable format with clearly listed sources and clickable links, ensuring transparency and enabling further verification. * **GCP-Ready Architecture:** The modular design (MCP servers, API clients, RAG processing) is built to be scalable and easily deployable on Google Cloud Platform services like Cloud Run, Vertex AI, and Cloud Storage. --- ## 🛠️ Technical Complexity & Design CliniSearch demonstrates a strong command of modern AI engineering principles and technologies. * **Asynchronous Architecture:** Utilizes `asyncio` and `httpx` for efficient, non-blocking calls to the backend MCP tool servers. * **Advanced RAG Implementation:** The system implements a full RAG pipeline, including: * **Document Parsing:** Using `PyMuPDF` to extract text from PDFs. * **Text Chunking & Embedding:** Using `sentence-transformers` for local text embedding. * **Vector Search:** Using `faiss-cpu` to create an efficient, in-memory vector store that simulates the functionality of a production service like GCP Vertex AI Vector Search. * **Sophisticated Prompt Engineering:** Prompts are dynamically constructed to be context-aware, instructing the LLM to use only the provided information and to cite its sources. * **Multimodal Fusion:** The radiology tool showcases a complex workflow where insights from a text-based RAG search (on PDFs) are fused into the prompt for a visual analysis task, demonstrating a true multimodal approach. * **Modular Codebase:** The project is well-organized into a Streamlit frontend (`app.py`), backend API clients (`utils/api_clients.py`), RAG logic (`utils/rag_processing.py`), and tool servers (`mcp_servers/`), promoting maintainability and scalability. --- ## Detailed System Architecture & Data Flow ![Untitled design](https://github.com/user-attachments/assets/53db4cce-1588-4e42-9df7-bcb718204ead) The diagram above provides a comprehensive overview of the Spectra AI technology stack and the flow of data from user interaction to final output. The system is divided into four logical domains: Frontend, Backend/Orchestrator, Local Tools & Services, and External APIs. 1. Frontend (UI - Blue) Streamlit App (app.py): This is the user's single point of interaction. It's responsible for rendering the web interface, managing user inputs (text queries, file uploads), and displaying the final, formatted results. 3. Application Backend / Orchestrator (Purple) RAG & Multimodal Logic: This is the "brain" of the application, also residing within app.py. It orchestrates the entire workflow, deciding which tools to call, when to process data, which LLMs to query for synthesis, and how to format the final response. 5. Local Tools & Services (Green) This domain contains components that run locally alongside the main application. MCP Tool Servers (FastAPI): These are two independent, lightweight servers built with FastAPI. They act as modular tools that the main orchestrator can call. Web Search Server: Receives a query, uses the duckduckgo-search library to get results from the internet, and returns them in a standard JSON format. PubMed Server: Receives a query, uses the BioPython library to interact with the NCBI Entrez API, and returns formatted PubMed abstracts. Local RAG Pipeline Components: This sub-domain handles the processing of user-uploaded documents. (Step 3a) PDF Parser (PyMuPDF): When a user uploads a PDF, this library extracts the raw text. (Step 3b) Embedding Model (Sentence Transformer): This is a crucial local LLM. It takes the text chunks from the PDF and converts them into numerical vector embeddings. We use a lightweight but effective model like all-MiniLM-L6-v2. (Step 3c) Vector Store (FAISS CPU): The generated embeddings are stored in this in-memory vector database. FAISS (Facebook AI Similarity Search) allows for incredibly fast and efficient semantic searches, simulating the functionality of a production service like GCP Vertex AI Vector Search. 7. External APIs & Data Sources (Orange) This domain represents all the third-party services the system relies on. Google Gemini API: The primary engine for high-level reasoning. It's used for: Text Synthesis: Generating the final, human-readable answers based on the context provided by the RAG pipeline. Vision Analysis: Analyzing the content of uploaded medical images. .output-box { border: 1px solid #444; border-radius: 5px; padding: 10px; background-color: #1a1a1a; /* A slightly different shade for the box */ margin-bottom: 20px; /* Add space between boxes */ } .output-box p { margin-bottom: 5px; /* Adjust paragraph spacing inside the box */ } .output-box a { color: #1c83e1; /* Make links stand out */ } """, unsafe_allow_html=True) # --- Initialize Session State --- if 'vector_store' not in st.session_state: embedding_dim = EMBEDDING_MODEL.get_sentence_embedding_dimension() st.session_state.vector_store = VectorStore(dimension=embedding_dim) if "messages" not in st.session_state: st.session_state.messages = [] # This state is specific to the radiology tab's uploader if 'radiology_vector_store' not in st.session_state: embedding_dim = EMBEDDING_MODEL.get_sentence_embedding_dimension() st.session_state.radiology_vector_store = VectorStore(dimension=embedding_dim) # --- Helper Functions --- def handle_pdf_upload(uploaded_files, vector_store_key): """Processes uploaded PDF files and adds them to the specified vector store.""" if uploaded_files: with st.spinner("Processing uploaded PDFs..."): all_chunks = [] for uploaded_file in uploaded_files: bytes_data = uploaded_file.read() chunks = parse_pdf(bytes_data, uploaded_file.name) if chunks: all_chunks.extend(chunks) if all_chunks: st.session_state[vector_store_key].add_documents(all_chunks) st.success(f"Processed and indexed {len(uploaded_files)} PDF(s).") else: st.error("Could not extract text from the uploaded PDF(s).") def format_output_as_html(source_name: str, answer: str, sources: list) -> str: """ Creates an HTML string with a custom-styled box for the output. """ # Start the styled container output_html = '
' output_html += f"

✅ Answer based on {source_name}:

" output_html += "
" # Replace newlines in the answer with
for HTML display output_html += f"

{answer.replace(chr(10), '
')}

" output_html += "
" if sources: output_html += f"

Sources from {source_name}:

" for i, source in enumerate(sources): title = source.get('title', 'N/A') link = source.get('link', '#') if source.get('type') == "PubMed": pmid = link.split('/')[-2] if link.endswith('/') else link.split('/')[-1] output_html += f"

{i+1}. {title}
   PubMed ID: {pmid} (Link)

" elif source.get('type') == "Web Search": output_html += f"

{i+1}. {title}
   Link

" else: # For PDF output_html += f"

{i+1}. {title} ({link})

" else: output_html += f"

No sources were retrieved from {source_name}.

" output_html += '
' return output_html # --- UI Layout --- st.title("🧠 CliniSearch AI Agent") st.caption("A multimodal assistant for medical research and radiological image analysis.") # --- Main Content Tabs --- tab1, tab2 = st.tabs(["Medical RAG Q&A", "Radiology Image Analysis"]) # --- TAB 1: Medical RAG Q&A --- with tab1: col1, col2 = st.columns([3, 1]) # Main content area and a sidebar-like column with col2: st.subheader("Controls & Tools") st.markdown("---") use_web_search = st.toggle("Enable Web Search", value=True, help="Include real-time web search results.") use_pubmed = st.toggle("Enable PubMed", value=True, help="Include PubMed article abstracts.") use_uploaded_docs_tab1 = st.toggle("Enable Uploaded Docs", value=True, help="Include your uploaded PDFs.") st.markdown("---") st.subheader("Upload Documents for RAG") st.info("Upload research papers or reports (PDFs) to include them in your questions.") uploaded_files_tab1 = st.file_uploader("Upload PDFs for Q&A", type="pdf", accept_multiple_files=True, key="pdf_uploader_tab1") if uploaded_files_tab1: if 'processed_files_tab1' not in st.session_state or st.session_state.processed_files_tab1 != [f.name for f in uploaded_files_tab1]: handle_pdf_upload(uploaded_files_tab1, 'vector_store') st.session_state.processed_files_tab1 = [f.name for f in uploaded_files_tab1] with col1: st.header("Medical Research & Question Answering") for message in st.session_state.messages: with st.chat_message(message["role"]): st.markdown(message["content"], unsafe_allow_html=True) if prompt := st.chat_input("Ask a medical research question..."): st.session_state.messages.append({"role": "user", "content": prompt}) with st.chat_message("user"): st.markdown(prompt) with st.chat_message("assistant"): final_outputs = [] if [truncated — 6095 more characters] ``` ### mcp_servers/web_search_server.py ```python import uvicorn from fastapi import FastAPI, HTTPException from pydantic import BaseModel from duckduckgo_search import DDGS import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI( title="Web Search MCP Server", description="An MCP-compliant server for performing web searches.", version="0.1.0", ) class MCPQuery(BaseModel): query: str class MCPResultItem(BaseModel): title: str snippet: str url: str class MCPResponse(BaseModel): source: str status: str results: list[MCPResultItem] | None = None error_message: str | None = None @app.post("/execute", response_model=MCPResponse) async def execute_web_search(mcp_query: MCPQuery): query = mcp_query.query if not query: logger.warning("Web Search Server: Received empty query.") raise HTTPException(status_code=400, detail="Query cannot be empty.") try: logger.info(f"Web Search Server: Received query: '{query}'") with DDGS() as ddgs: search_results = list(ddgs.text(query, max_results=5, region='wt-wt', safesearch='moderate')) formatted_results = [] for res in search_results: formatted_results.append( MCPResultItem( title=res.get("title", "N/A"), snippet=res.get("body", "N/A"), url=res.get("href", "N/A"), ) ) logger.info(f"Web Search Server: Found {len(formatted_results)} results for query '{query}'.") return MCPResponse( source="web_search_mcp_server", # Consistent source name status="success", results=formatted_results, ) except Exception as e: logger.error(f"Web Search Server: Error processing query '{query}': {str(e)}", exc_info=True) return MCPResponse( source="web_search_mcp_server", status="error", error_message=str(e) ) if __name__ == "__main__": logger.info("Starting Web Search MCP Server on http://localhost:8001") uvicorn.run(app, host="0.0.0.0", port=8001) ``` ### mcp_servers/pubmed_search_server.py ```python import uvicorn from fastapi import FastAPI, HTTPException from pydantic import BaseModel from Bio import Entrez import os from dotenv import load_dotenv import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) load_dotenv() app = FastAPI( title="PubMed Search MCP Server", description="An MCP-compliant server for performing PubMed searches.", version="0.1.0", ) NCBI_EMAIL = os.getenv("NCBI_EMAIL") if not NCBI_EMAIL: logger.warning("NCBI_EMAIL not set in .env file. PubMed queries might be throttled or blocked.") Entrez.email = NCBI_EMAIL class MCPQuery(BaseModel): query: str class PubMedResultItem(BaseModel): title: str snippet: str id: str url: str class MCPResponse(BaseModel): source: str status: str results: list[PubMedResultItem] | None = None error_message: str | None = None MAX_RESULTS_PUBMED = 3 @app.post("/execute", response_model=MCPResponse) async def execute_pubmed_search(mcp_query: MCPQuery): query = mcp_query.query if not query: logger.warning("PubMed Server: Received empty query.") raise HTTPException(status_code=400, detail="Query cannot be empty.") if not Entrez.email: logger.warning("PubMed Server: NCBI_EMAIL not configured. Proceeding but this is not recommended.") try: logger.info(f"PubMed Server: Received query: '{query}'") handle_search = Entrez.esearch(db="pubmed", term=query, retmax=str(MAX_RESULTS_PUBMED), sort="relevance") search_record = Entrez.read(handle_search) handle_search.close() id_list = search_record["IdList"] if not id_list: logger.info(f"PubMed Server: No results found for '{query}'.") return MCPResponse(source="pubmed_mcp_server", status="success", results=[]) # Consistent source name formatted_results = [] for pmid in id_list: try: handle_fetch = Entrez.efetch(db="pubmed", id=pmid, rettype="medline", retmode="text") article_text = handle_fetch.read() handle_fetch.close() title = "N/A" abstract = "Abstract not found." current_title_lines = [] current_abstract_lines = [] in_title = False in_abstract = False for line in article_text.splitlines(): if line.startswith("TI - "): current_title_lines.append(line[6:]) in_title = True; in_abstract = False elif line.startswith("AB - "): current_abstract_lines.append(line[6:]) in_abstract = True; in_title = False elif line.startswith(" ") and in_title: current_title_lines.append(line[4:]) elif line.startswith(" ") and in_abstract: current_abstract_lines.append(line[4:]) elif not line.startswith(" "): in_title = False; in_abstract = False if current_title_lines: title = " ".join(current_title_lines) if current_abstract_lines: abstract = " ".join(current_abstract_lines) snippet = abstract[:500] + ("..." if len(abstract) > 500 else "") formatted_results.append(PubMedResultItem( title=title, snippet=snippet, id=pmid, url=f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/" )) except Exception as e_article: logger.error(f"PubMed Server: Error processing article {pmid} for query '{query}': {str(e_article)}", exc_info=True) formatted_results.append(PubMedResultItem( title=f"Error fetching article {pmid}", snippet=str(e_article), id=pmid, url=f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/" )) logger.info(f"PubMed Server: Processed {len(formatted_results)} results for query '{query}'.") return MCPResponse(source="pubmed_mcp_server", status="success", results=formatted_results) except Exception as e: logger.error(f"PubMed Server: Error processing query '{query}': {str(e)}", exc_info=True) return MCPResponse(source="pubmed_mcp_server", status="error", error_message=str(e)) if __name__ == "__main__": logger.info("Starting PubMed MCP Server on http://localhost:8002") uvicorn.run(app, host="0.0.0.0", port=8002) ``` ### utils/api_clients.py ```python # utils/api_clients.py import os import google.generativeai as genai import anthropic from dotenv import load_dotenv # Load environment variables from .env file load_dotenv() class GeminiClient: """ A client for interacting with the Google Gemini API. This client uses the 'gemini-1.5-flash-latest' model, which is multimodal and can handle both text-only and text-with-image inputs. """ def __init__(self, api_key: str): if not api_key: raise ValueError("Google API Key not provided. Please set it in your .env file.") try: genai.configure(api_key=api_key) # Use the latest fast and powerful multimodal model for all tasks. self.model = genai.GenerativeModel('gemini-1.5-flash-latest') print("Gemini Client initialized successfully with model 'gemini-1.5-flash-latest'.") except Exception as e: print(f"Failed to configure Gemini Client: {e}") raise def generate_text(self, prompt: str, temperature: float = 0.3) -> str: """Generates text using the Gemini model.""" try: response = self.model.generate_content( prompt, generation_config=genai.types.GenerationConfig(temperature=temperature) ) return response.text except Exception as e: error_message = f"Error during Gemini text generation: {e}" print(error_message) return f"Error: Could not get a response from Gemini. Details: {e}" def analyze_image(self, prompt: str, image_bytes: bytes) -> str: """Analyzes an image and text prompt using the Gemini multimodal model.""" try: # The model automatically detects image mime type, but specifying is good practice. # Assuming JPEG for this example, but could be PNG, WEBP, etc. image_parts = [{"mime_type": "image/jpeg", "data": image_bytes}] prompt_parts = [prompt, image_parts[0]] response = self.model.generate_content(prompt_parts) return response.text except Exception as e: error_message = f"Error during Gemini image analysis: {e}" print(error_message) return f"Error: Could not analyze the image with Gemini. Details: {e}" class ClaudeClient: """ An optional client for interacting with the Anthropic Claude API. It will be disabled if the API key is not found. """ def __init__(self, api_key: str): if not api_key: print("Warning: ANTHROPIC_API_KEY not found. Claude client will be unavailable.") self.client = None else: try: self.client = anthropic.Anthropic(api_key=api_key) print("Claude Client initialized successfully.") except Exception as e: print(f"Failed to initialize Claude Client: {e}") self.client = None def generate_text(self, prompt: str, model: str = "claude-3-sonnet-20240229") -> str: """ Generates text using a Claude model. Available models: - "claude-3-opus-20240229" (Most powerful) - "claude-3-sonnet-20240229" (Balanced) - "claude-3-haiku-20240307" (Fastest, most compact) """ if not self.client: return "Error: Claude client is not configured. Please provide an ANTHROPIC_API_KEY." try: message = self.client.messages.create( model=model, max_tokens=2048, # Max output tokens messages=[ {"role": "user", "content": prompt} ] ) # The response is a list of content blocks; we extract the text from the first one. return message.content[0].text except Exception as e: error_message = f"Error during Claude text generation: {e}" print(error_message) return f"Error: Could not get a response from Claude. Details: {e}" # --- Global Client Initialization --- # These clients are initialized once when the module is imported, # making them available to the rest of the application. try: gemini_client = GeminiClient(api_key=os.getenv("GOOGLE_API_KEY")) except ValueError as e: # This will still raise the error and stop the app if the Gemini key is missing, # as Gemini is a core component of this app. print(f"CRITICAL ERROR: {e}") gemini_client = None claude_client = ClaudeClient(api_key=os.getenv("ANTHROPIC_API_KEY")) ``` ### utils/rag_processing.py ```python # utils/rag_processing.py import httpx import asyncio from sentence_transformers import SentenceTransformer import numpy as np import faiss import fitz # PyMuPDF from typing import List, Dict, Tuple # --- Configuration (remains the same) --- EMBEDDING_MODEL = SentenceTransformer('all-MiniLM-L6-v2') WEB_SEARCH_MCP_URL = "http://localhost:8001/execute" PUBMED_MCP_URL = "http://localhost:8002/execute" # --- Tool Interaction (remains the same) --- async def query_mcp_server(url: str, query: str) -> Dict: """Queries an MCP server asynchronously.""" async with httpx.AsyncClient() as client: try: response = await client.post(url, json={"query": query}, timeout=30.0) response.raise_for_status() return response.json() except httpx.RequestError as e: print(f"Error querying MCP server at {url}: {e}") return {"status": "error", "error_message": f"Connection Error: {e}. Is the server running?"} except Exception as e: print(f"Unhandled error querying MCP server at {url}: {e}") return {"status": "error", "error_message": str(e)} # --- PDF Parsing, Chunking, VectorStore (all remain the same) --- def chunk_text(text: str, chunk_size=512, chunk_overlap=50) -> List[str]: if not text: return [] words = text.split() chunks = [] for i in range(0, len(words), chunk_size - chunk_overlap): chunks.append(" ".join(words[i:i + chunk_size])) return chunks def parse_pdf(file_bytes: bytes, filename: str) -> List[Dict]: document_chunks = [] try: doc = fitz.open(stream=file_bytes, filetype="pdf") full_text = "" for page_num, page in enumerate(doc): full_text += page.get_text() + "\n" text_chunks = chunk_text(full_text) for i, chunk in enumerate(text_chunks): document_chunks.append({ "source": f"PDF: {filename}", "content": chunk, "metadata": {"page_num_approx": 1 + (i * (512 - 50) // 400)} }) return document_chunks except Exception as e: print(f"Error parsing PDF '{filename}': {e}") return [] class VectorStore: def __init__(self, dimension): self.dimension = dimension self.index = faiss.IndexFlatL2(dimension) self.documents = [] def add_documents(self, docs: List[Dict]): contents = [doc['content'] for doc in docs] if not contents: return embeddings = EMBEDDING_MODEL.encode(contents, convert_to_tensor=False) self.index.add(np.array(embeddings).astype('float32')) self.documents.extend(docs) def search(self, query: str, k=3) -> List[Dict]: if self.index.ntotal == 0: return [] query_embedding = EMBEDDING_MODEL.encode([query], convert_to_tensor=False) distances, indices = self.index.search(np.array(query_embedding).astype('float32'), k=min(k, self.index.ntotal)) return [self.documents[i] for i in indices[0] if i != -1] # --- Main RAG Orchestration (MODIFIED AND CORRECTED) --- async def perform_rag(query: str, vector_store: VectorStore, use_web=True, use_pubmed=True) -> Tuple[str, List[Dict]]: """ Orchestrates the RAG process for specified sources and returns both the formatted context string and a structured list of source documents. """ print(f"Performing RAG for query: '{query}' with flags Web={use_web}, PubMed={use_pubmed}") context = "" sources = [] # --- Step 1: Query external tools if requested --- tasks = [] if use_web: tasks.append(query_mcp_server(WEB_SEARCH_MCP_URL, query)) if use_pubmed: tasks.append(query_mcp_server(PUBMED_MCP_URL, query)) if tasks: tool_results = await asyncio.gather(*tasks) # Consolidate and format context and sources from external tools for res in tool_results: if res and res.get('status') == 'success' and res.get('results'): is_pubmed = "pubmed" in res.get('source', '').lower() source_name = "PubMed" if is_pubmed else "Web Search" context += f"--- Context from {source_name} ---\n" for item in res['results'][:3]: # Take top 3 results title = item.get('title', 'N/A') snippet = item.get('snippet', 'N/A') context += f"Title: {title}\nSnippet: {snippet}\n\n" if is_pubmed: pmid = item.get('id', '') sources.append({ "type": "PubMed", "title": title, "link": f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/" if pmid else "#" }) else: sources.append({ "type": "Web Search", "title": title, "link": item.get('url', '#') }) # --- Step 2: Query internal vector store if no external tools were used --- # This logic assumes that if a user turns off both Web and PubMed, they *only* want to # query the documents they have uploaded. if not use_web and not use_pubmed: print("Querying uploaded documents only...") semantic_results = vector_store.search(query, k=5) # Get more results if it's the only source if semantic_results: context += "--- Context from Uploaded Documents ---\n" for res in semantic_results: context += f"Source: {res.get('source', 'Uploaded Document')} (Page ~{res.get('metadata', {}).get('page_num_approx', 'N/A')}) - Content: {res.get('content', 'N/A')}\n\n" sources.append({ "type": "PDF Document", "title": res.get('source', 'Uploaded Document'), "link": f"Page ~{res.get('metadata', {} [truncated — 429 more characters] ```