Building an Advanced Multi-Agent Research Pipeline with LangGraph and Gemini AI

Leveraging LangGraph and Google's Gemini Model for Automated Research Workflows

This tutorial demonstrates how to construct a sophisticated multi-agent system using LangGraph combined with Google’s free-tier Gemini model to automate end-to-end research workflows. The system orchestrates three specialized agents: Research, Analysis, and Report, to generate comprehensive insights and executive reports.

Setting Up the Environment and Core Components

First, we install essential libraries such as LangGraph, LangChain-Google-GenAI, and LangChain-Core, and import necessary modules. The workflow state is structured using a TypedDict called AgentState that manages messages, current agent, research data, analysis status, and the final report.

We initialize the Gemini-1.5-Flash model with a temperature of 0.7 to balance creativity and precision.

!pip install -q langgraph langchain-google-genai langchain-core
 
import os
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
import operator
 
os.environ["GOOGLE_API_KEY"] = "Use Your Own API Key"
 
class AgentState(TypedDict):
   messages: Annotated[List[BaseMessage], operator.add]
   current_agent: str
   research_data: dict
   analysis_complete: bool
   final_report: str
 
llm = ChatGoogleGenerativeAI(model="gemini-1.5-flash", temperature=0.7)

Simulating Web Search and Data Analysis Tools

To mimic real-world tools, we define placeholder functions:

simulate_web_search(query: str): Returns mock search results for the query.
simulate_data_analysis(data: str): Returns mock analysis insights.

These simulate the retrieval and analysis phases within the workflow.

Research Agent: Gathering Structured Information

The Research agent uses the last user query to perform a simulated web search and prompts the Gemini model to produce a structured research summary, including key facts, trends, expert opinions, and statistics.

def research_agent(state: AgentState) -> AgentState:
   messages = state["messages"]
   last_message = messages[-1].content
  
   search_results = simulate_web_search(last_message)
  
   prompt = f"""You are a research agent. Based on the query: \"{last_message}\"\n\n   Here are the search results: {search_results}\n\n   Conduct thorough research and gather relevant information. Provide structured findings with:\n   1. Key facts and data points\n   2. Current trends and developments \n   3. Expert opinions and insights\n   4. Relevant statistics\n\n   Be comprehensive and analytical in your research summary."""
  
   response = llm.invoke([HumanMessage(content=prompt)])
  
   research_data = {
       "topic": last_message,
       "findings": response.content,
       "search_results": search_results,
       "sources": ["academic_papers", "industry_reports", "expert_analyses"],
       "confidence": 0.88,
       "timestamp": "2024-research-session"
   }
  
   return {
       "messages": state["messages"] + [AIMessage(content=f"Research completed on '{last_message}': {response.content}")],
       "current_agent": "analysis",
       "research_data": research_data,
       "analysis_complete": False,
       "final_report": ""
   }

Analysis Agent: Extracting Deep Insights

The Analysis agent receives the research findings, simulates data analysis, and prompts Gemini to generate in-depth analytical insights, including pattern identification, risk assessment, and strategic recommendations.

def analysis_agent(state: AgentState) -> AgentState:
   research_data = state["research_data"]
  
   analysis_results = simulate_data_analysis(research_data.get('findings', ''))
  
   prompt = f"""You are an analysis agent. Analyze this research data in depth:\n\n   Topic: {research_data.get('topic', 'Unknown')}\n   Research Findings: {research_data.get('findings', 'No findings')}\n   Analysis Results: {analysis_results}\n\n   Provide deep insights including:\n   1. Pattern identification and trend analysis\n   2. Comparative analysis with industry standards\n   3. Risk assessment and opportunities \n   4. Strategic implications\n   5. Actionable recommendations with priority levels\n\n   Be analytical and provide evidence-based insights."""
  
   response = llm.invoke([HumanMessage(content=prompt)])
  
   return {
       "messages": state["messages"] + [AIMessage(content=f"Analysis completed: {response.content}")],
       "current_agent": "report",
       "research_data": state["research_data"],
       "analysis_complete": True,
       "final_report": ""
   }

Report Agent: Producing a Professional Executive Report

The Report agent compiles the research topic, findings, and analysis into a well-structured, professional executive report with sections such as executive summary, key findings, analytical insights, strategic recommendations, risk assessment, and next steps.

def report_agent(state: AgentState) -> AgentState:
   research_data = state["research_data"]
  
   analysis_message = None
   for msg in reversed(state["messages"]):
       if isinstance(msg, AIMessage) and "Analysis completed:" in msg.content:
           analysis_message = msg.content.replace("Analysis completed: ", "")
           break
  
   prompt = f"""You are a professional report generation agent. Create a comprehensive executive report based on:\n\n    Research Topic: {research_data.get('topic')}\n    Research Findings: {research_data.get('findings')}\n    Analysis Results: {analysis_message or 'Analysis pending'}\n\n   Generate a well-structured, professional report with these sections:\n\n   ## EXECUTIVE SUMMARY  \n   ## KEY RESEARCH FINDINGS \n   [Detail the most important discoveries and data points]\n\n   ## ANALYTICAL INSIGHTS\n   [Present deep analysis, patterns, and trends identified]\n\n   ## STRATEGIC RECOMMENDATIONS\n   [Provide actionable recommendations with priority levels]\n\n   ## RISK ASSESSMENT & OPPORTUNITIES\n   [Identify potential risks and opportunities]\n\n   ## CONCLUSION & NEXT STEPS\n   [Summarize and suggest follow-up actions]\n\n   Make the report professional, data-driven, and actionable."""
  
   response = llm.invoke([HumanMessage(content=prompt)])
  
   return {
       "messages": state["messages"] + [AIMessage(content=f" FINAL REPORT GENERATED:\n\n{response.content}")],
       "current_agent": "complete",
       "research_data": state["research_data"],
       "analysis_complete": True,
       "final_report": response.content
   }

Orchestrating the Multi-Agent Workflow

A StateGraph is constructed to manage the workflow, linking the agents and transitioning based on the current agent state. The entry point is set to the research agent. The run_research_assistant function executes the full pipeline sequentially, printing progress and returning the final report.

def should_continue(state: AgentState) -> str:
   current_agent = state.get("current_agent", "research")
  
   if current_agent == "research":
       return "analysis"
   elif current_agent == "analysis":
       return "report"
   elif current_agent == "report":
       return END
   else:
       return END
 
 
workflow = StateGraph(AgentState)
 
workflow.add_node("research", research_agent)
workflow.add_node("analysis", analysis_agent)
workflow.add_node("report", report_agent)
 
workflow.add_conditional_edges(
   "research",
   should_continue,
   {"analysis": "analysis", END: END}
)
 
workflow.add_conditional_edges(
   "analysis",
   should_continue,
   {"report": "report", END: END}
)
 
workflow.add_conditional_edges(
   "report",
   should_continue,
   {END: END}
)
 
workflow.set_entry_point("research")
 
app = workflow.compile()
 
 
def run_research_assistant(query: str):
   initial_state = {
       "messages": [HumanMessage(content=query)],
       "current_agent": "research",
       "research_data": {},
       "analysis_complete": False,
       "final_report": ""
   }
  
   print(f" Starting Multi-Agent Research on: '{query}'")
   print("=" * 60)
  
   current_state = initial_state
  
   print(" Research Agent: Gathering information...")
   current_state = research_agent(current_state)
   print(" Research phase completed!\n")
  
   print(" Analysis Agent: Analyzing findings...")
   current_state = analysis_agent(current_state)
   print(" Analysis phase completed!\n")
  
   print(" Report Agent: Generating comprehensive report...")
   final_state = report_agent(current_state)
   print(" Report generation completed!\n")
  
   print("=" * 60)
   print(" MULTI-AGENT WORKFLOW COMPLETED SUCCESSFULLY!")
   print("=" * 60)
  
   final_report = final_state['final_report']
   print(f"\n COMPREHENSIVE RESEARCH REPORT:\n")
   print(final_report)
  
   return final_state

Usage and Extensibility

The system prints readiness messages, example queries, and guides users to run the research assistant with their own questions. This modular setup allows easy extension or integration with real APIs, alternate models, or additional tools to tailor the research pipeline to specific needs.

if __name__ == "__main__":
   print(" Advanced LangGraph Multi-Agent System Ready!")
   print(" Remember to set your GOOGLE_API_KEY!")
  
   example_queries = [
       "Impact of renewable energy on global markets",
       "Future of remote work post-pandemic"
   ]
  
   print(f"\n Example queries you can try:")
   for i, query in enumerate(example_queries, 1):
       print(f"  {i}. {query}")
  
   print(f"\n Usage: run_research_assistant('Your research question here')")
  
   result = run_research_assistant("What are emerging trends in sustainable technology?")

This framework empowers rapid prototyping and refinement of multi-agent research systems, making it adaptable for diverse domains and evolving product goals.