In this tutorial, we build a complete multi-agent research team system using LangGraph and Google’s Gemini API. We utilize role-specific agents, Researcher, Analyst, Writer, and Supervisor, each responsible for a distinct part of the research pipeline. Together, these agents collaboratively gather data, analyze insights, synthesize a report, and coordinate the workflow. We also incorporate features like memory persistence, agent coordination, custom agents, and performance monitoring. By the end of the setup, we can run automated, intelligent research sessions that generate structured reports on any given topic.
!pip install langgraph langchain-google-genai langchain-community langchain-core python-dotenvimport osfrom typing import Annotated, List, Tuple, Unionfrom typing_extensions import TypedDictimport operatorfrom langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessagefrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderfrom langchain_google_genai import ChatGoogleGenerativeAIfrom langgraph.graph import StateGraph, ENDfrom langgraph.prebuilt import ToolNodefrom langgraph.checkpoint.memory import MemorySaverimport functoolsimport getpassGOOGLE_API_KEY = getpass.getpass("Enter your Google API Key: ")os.environ["GOOGLE_API_KEY"] = GOOGLE_API_KEYWe begin by installing the necessary libraries, including LangGraph and LangChain’s Google Gemini integration. Then, we import the essential modules and set up our environment by securely entering the Google API key using the getpass module. This ensures we can authenticate our Gemini LLM without exposing the key in the code.
class AgentState(TypedDict): """State shared between all agents in the graph""" messages: Annotated[list, operator.add] next: str current_agent: str research_topic: str findings: dict final_report: strclass AgentResponse(TypedDict): """Standard response format for all agents""" content: str next_agent: str findings: dictdef create_llm(temperature: float = 0.1, model: str = "gemini-1.5-flash") -> ChatGoogleGenerativeAI: """Create a configured Gemini LLM instance""" return ChatGoogleGenerativeAI( model=model, temperature=temperature, google_api_key=os.environ["GOOGLE_API_KEY"] )We define two TypedDict classes to maintain structured state and responses shared across all agents in the LangGraph. AgentState tracks messages, workflow status, topic, and collected findings, while AgentResponse standardizes each agent’s output. We also create a helper function to start the Gemini LLM with a specified model and temperature, ensuring consistent behavior across all agents.
def create_research_agent(llm: ChatGoogleGenerativeAI) -> callable: """Creates a research specialist agent for initial data gathering""" research_prompt = ChatPromptTemplate.from_messages([ ("system", """You are a Research Specialist AI. Your role is to: 1. Analyze the research topic thoroughly 2. Identify key areas that need investigation 3. Provide initial research findings and insights 4. Suggest specific angles for deeper analysis Focus on providing comprehensive, accurate information and clear research directions. Always structure your response with clear sections and bullet points. """), MessagesPlaceholder(variable_name="messages"), ("human", "Research Topic: {research_topic}") ]) research_chain = research_prompt | llm def research_agent(state: AgentState) -> AgentState: """Execute research analysis""" try: response = research_chain.invoke({ "messages": state["messages"], "research_topic": state["research_topic"] }) findings = { "research_overview": response.content, "key_areas": ["area1", "area2", "area3"], "initial_insights": response.content[:500] + "..." } return { "messages": state["messages"] + [AIMessage(content=response.content)], "next": "analyst", "current_agent": "researcher", "research_topic": state["research_topic"], "findings": {**state.get("findings", {}), "research": findings}, "final_report": state.get("final_report", "") } except Exception as e: error_msg = f"Research agent error: {str(e)}" return { "messages": state["messages"] + [AIMessage(content=error_msg)], "next": "analyst", "current_agent": "researcher", "research_topic": state["research_topic"], "findings": state.get("findings", {}), "final_report": state.get("final_report", "") } return research_agentWe now create our first specialized agent, the Research Specialist AI. This agent is prompted to deeply analyze a given topic, extract key areas of interest, and suggest directions for further exploration. Using a ChatPromptTemplate, we define its behavior and connect it with our Gemini LLM. The research_agent function executes this logic, updates the shared state with findings and messages, and passes control to the next agent in line, the Analyst.
def create_analyst_agent(llm: ChatGoogleGenerativeAI) -> callable: """Creates a data analyst agent for deep analysis""" analyst_prompt = ChatPromptTemplate.from_messages([ ("system", """You are a Data Analyst AI. Your role is to: 1. Analyze data and information provided by the research team 2. Identify patterns, trends, and correlations 3. Provide statistical insights and data-driven conclusions 4. Suggest actionable recommendations based on analysis Focus on quantitative analysis, data interpretation, and evidence-based insights. Use clear metrics and concrete examples in your analysis. """), MessagesPlaceholder(variable_name="messages"), ("human", "Analyze the research findings for: {research_topic}") ]) analyst_chain = analyst_prompt | llm def analyst_agent(state: AgentState) -> AgentState: """Execute data analysis""" try: response = analyst_chain.invoke({ "messages": state["messages"], "research_topic": state["research_topic"] }) analysis_findings = { "analysis_summary": response.content, "key_metrics": ["metric1", "metric2", "metric3"], "recommendations": response.content.split("recommendations:")[-1] if "recommendations:" in response.content.lower() else "No specific recommendations found" } return { "messages": state["messages"] + [AIMessage(content=response.content)], "next": "writer", "current_agent": "analyst", "research_topic": state["research_topic"], "findings": {**state.get("findings", {}), "analysis": analysis_findings}, "final_report": state.get("final_report", "") } except Exception as e: error_msg = f"Analyst agent error: {str(e)}" return { "messages": state["messages"] + [AIMessage(content=error_msg)], "next": "writer", "current_agent": "analyst", "research_topic": state["research_topic"], "findings": state.get("findings", {}), "final_report": state.get("final_report", "") } return analyst_agentWe now define the Data Analyst AI, which dives deeper into the research findings generated by the previous agent. This agent identifies key patterns, trends, and metrics, offering actionable insights backed by evidence. Using a tailored system prompt and the Gemini LLM, the analyst_agent function enriches the state with structured analysis, preparing the groundwork for the report writer to synthesize everything into a final document.
def create_writer_agent(llm: ChatGoogleGenerativeAI) -> callable: """Creates a report writer agent for final documentation""" writer_prompt = ChatPromptTemplate.from_messages([ ("system", """You are a Report Writer AI. Your role is to: 1. Synthesize all research and analysis into a comprehensive report 2. Create clear, professional documentation 3. Ensure proper structure with executive summary, findings, and conclusions 4. Make complex information accessible to various audiences Focus on clarity, completeness, and professional presentation. Include specific examples and actionable insights. """), MessagesPlaceholder(variable_name="messages"), ("human", "Create a comprehensive report for: {research_topic}") ]) writer_chain = writer_prompt | llm def writer_agent(state: AgentState) -> AgentState: """Execute report writing""" try: response = writer_chain.invoke({ "messages": state["messages"], "research_topic": state["research_topic"] }) return { "messages": state["messages"] + [AIMessage(content=response.content)], "next": "supervisor", "current_agent": "writer", "research_topic": state["research_topic"], "findings": state.get("findings", {}), "final_report": response.content } except Exception as e: error_msg = f"Writer agent error: {str(e)}" return { "messages": state["messages"] + [AIMessage(content=error_msg)], "next": "supervisor", "current_agent": "writer", "research_topic": state["research_topic"], "findings": state.get("findings", {}), "final_report": f"Error generating report: {str(e)}" } return writer_agentWe now create the Report Writer AI, which is responsible for transforming the collected research and analysis into a polished, structured document. This agent synthesizes all previous insights into a clear, professional report with an executive summary, detailed findings, and conclusions. By invoking the Gemini model with a structured prompt, the writer agent updates the final report in the shared state and hands control over to the Supervisor agent for review.
def create_supervisor_agent(llm: ChatGoogleGenerativeAI, members: List[str]) -> callable: """Creates a supervisor agent to coordinate the team""" options = ["FINISH"] + members supervisor_prompt = ChatPromptTemplate.from_messages([ ("system", f"""You are a Supervisor AI managing a research team. Your team members are: {', '.join(members)} Your responsibilities: 1. Coordinate the workflow between team members 2. Ensure each agent completes their specialized tasks 3. Determine when the research is complete 4. Maintain quality standards throughout the process Given the conversation, determine the next step: - If research is needed: route to "researcher" - If analysis is needed: route to "analyst" - If report writing is needed: route to "writer" - If work is complete: route to "FINISH" Available options: {options} Respond with just the name of the next agent or "FINISH". """), MessagesPlaceholder(variable_name="messages"), ("human", "Current status: {current_agent} just completed their task for topic: {research_topic}") ]) supervisor_chain = supervisor_prompt | llm def supervisor_agent(state: AgentState) -> AgentState: """Execute supervisor coordination""" try: response = supervisor_chain.invoke({ "messages": state["messages"], "current_agent": state.get("current_agent", "none"), "research_topic": state["research_topic"] }) next_agent = response.content.strip().lower() if "finish" in next_agent or "complete" in next_agent: next_step = "FINISH" elif "research" in next_agent: next_step = "researcher" elif "analy" in next_agent: next_step = "analyst" elif "writ" in next_agent: next_step = "writer" else: current = state.get("current_agent", "") if current == "researcher": next_step = "analyst" elif current == "analyst": next_step = "writer" elif current == "writer": next_step = "FINISH" else: next_step = "researcher" return { "messages": state["messages"] + [AIMessage(content=f"Supervisor decision: Next agent is {next_step}")], "next": next_step, "current_agent": "supervisor", "research_topic": state["research_topic"], "findings": state.get("findings", {}), "final_report": state.get("final_report", "") } except Exception as e: error_msg = f"Supervisor error: {str(e)}" return { "messages": state["messages"] + [AIMessage(content=error_msg)], "next": "FINISH", "current_agent": "supervisor", "research_topic": state["research_topic"], "findings": state.get("findings", {}), "final_report": state.get("final_report", "") } return supervisor_agentWe now bring in the Supervisor AI, which oversees and orchestrates the entire multi-agent workflow. This agent evaluates the current progress, knowing which team member just finished their task, and intelligently decides the next step: whether to continue with research, proceed to analysis, initiate report writing, or mark the project as complete. By parsing the conversation context and utilizing Gemini for reasoning, the supervisor agent ensures smooth transitions and quality control throughout the research pipeline.
def create_research_team_graph() -> StateGraph: """Creates the complete research team workflow graph""" llm = create_llm() members = ["researcher", "analyst", "writer"] researcher = create_research_agent(llm) analyst = create_analyst_agent(llm) writer = create_writer_agent(llm) supervisor = create_supervisor_agent(llm, members) workflow = StateGraph(AgentState) workflow.add_node("researcher", researcher) workflow.add_node("analyst", analyst) workflow.add_node("writer", writer) workflow.add_node("supervisor", supervisor) workflow.add_edge("researcher", "supervisor") workflow.add_edge("analyst", "supervisor") workflow.add_edge("writer", "supervisor") workflow.add_conditional_edges( "supervisor", lambda x: x["next"], { "researcher": "researcher", "analyst": "analyst", "writer": "writer", "FINISH": END } ) workflow.set_entry_point("supervisor") return workflowdef compile_research_team(): """Compile the research team graph with memory""" workflow = create_research_team_graph() memory = MemorySaver() app = workflow.compile(checkpointer=memory) return appdef run_research_team(topic: str, thread_id: str = "research_session_1"): """Run the complete research team workflow""" app = compile_research_team() initial_state = { "messages": [HumanMessage(content=f"Research the topic: {topic}")], "research_topic": topic, "next": "researcher", "current_agent": "start", "findings": {}, "final_report": "" } config = {"configurable": {"thread_id": thread_id}} print(f"
Starting research on: {topic}") print("=" * 50) try: final_state = None for step, state in enumerate(app.stream(initial_state, config=config)): print(f"\n
Step {step + 1}: {list(state.keys())[0]}") current_state = list(state.values())[0] if current_state["messages"]: last_message = current_state["messages"][-1] if isinstance(last_message, AIMessage): print(f"
{last_message.content[:200]}...") final_state = current_state if step > 10: print("
Maximum steps reached. Stopping execution.") break return final_state except Exception as e: print(f"
Error during execution: {str(e)}") return NoneCheck out the full Codes
We now assemble and execute the entire multi-agent workflow using LangGraph. First, we define the research team graph, which consists of nodes for each agent, Researcher, Analyst, Writer, and Supervisor, connected by logical transitions. Then, we compile this graph with memory using MemorySaver to persist conversation history. Finally, the run_research_team() function initializes the process with a topic and streams execution step by step, allowing us to track each agent’s contribution in real-time. This orchestration ensures a fully automated, collaborative research pipeline.
if __name__ == "__main__": result = run_research_team("Artificial Intelligence in Healthcare") if result: print("\n" + "=" * 50) print("
FINAL RESULTS") print("=" * 50) print(f"
Final Agent: {result['current_agent']}") print(f"
Findings: {len(result['findings'])} sections") print(f"
Report Length: {len(result['final_report'])} characters") if result['final_report']: print("\n FINAL REPORT:") print("-" * 30) print(result['final_report'])def interactive_research_session(): """Run an interactive research session""" app = compile_research_team() print("
Interactive Research Team Session") print("Enter 'quit' to exit\n") session_count = 0 while True: topic = input(" Enter research topic: ").strip() if topic.lower() in ['quit', 'exit', 'q']: print("
Goodbye!") break if not topic: print(" Please enter a valid topic.") continue session_count += 1 thread_id = f"interactive_session_{session_count}" result = run_research_team(topic, thread_id) if result and result['final_report']: print(f"\n
Research completed for: {topic}") print(f" Report preview: {result['final_report'][:300]}...") show_full = input("\n
Show full report? (y/n): ").lower() if show_full.startswith('y'): print("\n" + "=" * 60) print(" COMPLETE RESEARCH REPORT") print("=" * 60) print(result['final_report']) print("\n" + "-" * 50)def create_custom_agent(role: str, instructions: str, llm: ChatGoogleGenerativeAI) -> callable: """Create a custom agent with specific role and instructions""" custom_prompt = ChatPromptTemplate.from_messages([ ("system", f"""You are a {role} AI. Your specific instructions: {instructions} Always provide detailed, professional responses relevant to your role. """), MessagesPlaceholder(variable_name="messages"), ("human", "Task: {task}") ]) custom_chain = custom_prompt | llm def custom_agent(state: AgentState) -> AgentState: """Execute custom agent task""" try: response = custom_chain.invoke({ "messages": state["messages"], "task": state["research_topic"] }) return { "messages": state["messages"] + [AIMessage(content=response.content)], "next": "supervisor", "current_agent": role.lower().replace(" ", "_"), "research_topic": state["research_topic"], "findings": state.get("findings", {}), "final_report": state.get("final_report", "") } except Exception as e: error_msg = f"{role} agent error: {str(e)}" return { "messages": state["messages"] + [AIMessage(content=error_msg)], "next": "supervisor", "current_agent": role.lower().replace(" ", "_"), "research_topic": state["research_topic"], "findings": state.get("findings", {}), "final_report": state.get("final_report", "") } return custom_agentCheck out the full Codes
We wrap up our system with runtime and customization capabilities. The main block allows us to trigger a research run directly, making it perfect for testing the pipeline with a real-world topic, such as Artificial Intelligence in Healthcare. For more dynamic use, the interactive_research_session() enables multiple topic queries in a loop, simulating real-time exploration. Lastly, the create_custom_agent() function allows us to integrate new agents with unique roles and instructions, making the framework flexible and extensible for specialized workflows.
def visualize_graph(): """Visualize the research team graph structure""" try: app = compile_research_team() graph_repr = app.get_graph() print("
Research Team Graph Structure") print("=" * 40) print(f"Nodes: {list(graph_repr.nodes.keys())}") print(f"Edges: {[(edge.source, edge.target) for edge in graph_repr.edges]}") try: graph_repr.draw_mermaid() except: print(" Visual graph requires mermaid-py package") print("Install with: !pip install mermaid-py") except Exception as e: print(f" Error visualizing graph: {str(e)}")import timefrom datetime import datetimedef monitor_research_performance(topic: str): """Monitor and report performance metrics""" start_time = time.time() print(f"
Starting performance monitoring for: {topic}") result = run_research_team(topic, f"perf_test_{int(time.time())}") end_time = time.time() duration = end_time - start_time metrics = { "duration": duration, "total_messages": len(result["messages"]) if result else 0, "findings_sections": len(result["findings"]) if result else 0, "report_length": len(result["final_report"]) if result and result["final_report"] else 0, "success": result is not None } print("\n PERFORMANCE METRICS") print("=" * 30) print(f" Duration: {duration:.2f} seconds") print(f" Total Messages: {metrics['total_messages']}") print(f" Findings Sections: {metrics['findings_sections']}") print(f" Report Length: {metrics['report_length']} chars") print(f" Success: {metrics['success']}") return metricsdef quick_start_demo(): """Complete demo of the research team system""" print("
LangGraph Research Team - Quick Start Demo") print("=" * 50) topics = [ "Climate Change Impact on Agriculture", "Quantum Computing Applications", "Digital Privacy in the Modern Age" ] for i, topic in enumerate(topics, 1): print(f"\n Demo {i}: {topic}") print("-" * 40) try: result = run_research_team(topic, f"demo_{i}") if result and result['final_report']: print(f" Research completed successfully!") print(f" Report preview: {result['final_report'][:150]}...") else: print(" Research failed") except Exception as e: print(f" Error in demo {i}: {str(e)}") print("\n" + "="*30) print("
Demo completed!")quick_start_demo()We finalize the system by adding powerful utilities for graph visualization, performance monitoring, and a quick start demo. The visualize_graph() function provides a structural overview of agent connections, ideal for debugging or presentation purposes. The monitor_research_performance() tracks runtime, message volume, and report size, helping us evaluate the system’s efficiency. Finally, quick_start_demo() runs multiple sample research topics in sequence, showing how seamlessly the agents collaborate to generate insightful reports.
In conclusion, we’ve successfully built and tested a fully functional, modular AI research assistant framework using LangGraph. With clear agent roles and automated task routing, we streamline research from raw topic input to a well-structured final report. Whether we use the quick start demo, run interactive sessions, or monitor performance, this system empowers us to handle complex research tasks with minimal intervention. We’re now equipped to adapt or extend this setup further by integrating custom agents, visualizing workflows, or even deploying it into real-world applications.
Check out the full Codes | Sponsorship Opportunity: Want to reach the most influential AI developers across the US and Europe? Join our ecosystem of 1M+ monthly readers and 500K+ engaged community members. [Explore Sponsorship]
The post Building a Multi-Agent AI Research Team with LangGraph and Gemini for Automated Reporting appeared first on MarkTechPost.
