In this tutorial, we built a complete multi-agent research team system using Langgraph and Google’s Gemini API. We utilize agents, researchers, analysts, writers and supervisors in specific roles, each responsible for studying different parts of the pipeline. Together, these agents collect data, analyze insights, synthesize reports, and coordinate workflows. We also combine features such as memory persistence, proxy coordination, custom proxy and performance monitoring. By the end of the setup, we can run automated, intelligent research courses to generate structured reports on any given topic.
!pip install langgraph langchain-google-genai langchain-community langchain-core python-dotenv
import os
from typing import Annotated, List, Tuple, Union
from typing_extensions import TypedDict
import operator
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_google_genai import ChatGoogleGenerativeAI
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver
import functools
import getpass
GOOGLE_API_KEY = getpass.getpass("Enter your Google API Key: ")
os.environ["GOOGLE_API_KEY"] = GOOGLE_API_KEYWe first install the necessary libraries, including the Google Gemini integration of Langgraph and Langchain. We then import the basic module and set up the environment by safely entering the Google API key using the GetPass module. This ensures that we can authenticate our Gemini LLM without exposing the keys 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: str
class AgentResponse(TypedDict):
"""Standard response format for all agents"""
content: str
next_agent: str
findings: dict
def 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 type categories to maintain the structured state and response of all agents in langgraph. AgentState tracks messages, workflow states, topics, and collections discovery, while AgentResponse standardizes the output of each agent. We also created an accessibility feature to start Gemini LLM with the specified model and temperature, ensuring that all reagents behave consistently.
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_agentNow, we have created the first professional broker research expert AI. This agent prompts the agent to analyze the given topic in depth, extract key key areas, and propose directions for further exploration. Using ChatPromptTemplate, we define its behavior and associate it with Gemini LLM. The Research_agent function performs this logic, updates the shared state through discovery and message, and then passes control to the analyst’s next agent.
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_agentNow, we define Data Analyst AI, which delves into research findings generated by previous agents. The agency identified key models, trends and metrics, providing actionable insights supported by evidence. Using tailored system prompts and Gemini LLM, the Analyst_agent function enriches the state with structured analysis, preparing for the basis of the report author to synthesize everything into the 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_agent
Now, we have created the report author AI, which is responsible for converting collected research and analysis into polished structured documents. The agent combines all previous insights into clear, professional reports and provides executive summary, detailed findings and conclusions. By calling the Gemini model with structured prompts, the author agent updated the final report in the shared state and controlled control power 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_agentNow, we have introduced the Supervisor AI, who oversees and carefully plans the entire multi-agent workflow. The agent evaluates the current progress, knows which team member has just completed the task, and wisely decides the next step: whether to continue the research, continue the analysis, initiate report writing, or mark the project as completed. By analyzing the dialogue environment and using Gemini for reasoning, the supervisor agent ensures smooth transition 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 workflow
def compile_research_team():
"""Compile the research team graph with memory"""
workflow = create_research_team_graph()
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
return app
def 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 NoneView full Code
Now we assemble and execute the entire multi-agent workflow using langgraph. First, we define a research team graph, which consists of each agent, researcher, analyst, writer, and supervisor, and is connected by logical transitions. We then compile it into memory using Memory Saver to continue conversation history. Finally, the run_research_team() function initializes the process with a topic and executes it step by step, allowing us to track the contributions of each agent 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 exitn")
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_agentView full Code
We organize the system through runtime and custom functions. The main blocks allow us to get direct contact with research, which is ideal for testing pipelines using real-world topics such as artificial intelligence in healthcare. For more dynamic use, Interactive_research_session() enables multiple topic queries in a loop, thus simulating real-time exploration. Finally, the create_custom_agent() function allows us to integrate new agents into new agents with unique roles and descriptions, making the framework flexible and scalable to dedicated 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 time
from datetime import datetime
def 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 metrics
def 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 to visualize, performance monitoring, and quick start demonstrations. The Visualize_graph() function provides an overview of the structure of the proxy connection, which is ideal for debugging or demonstration purposes. Monitor_Research_Performance() tracks the runtime, message volume and report size, helping us evaluate the efficiency of the system. Finally, quick_start_demo() runs multiple sample research topics in sequence, showing how agents work seamlessly to generate insightful reports.
In short, we successfully built and tested a fully functional modular AI research assistant framework using Langgraph. With clear proxy roles and automatic task routing, we simplify the study from the original topic input to a well-structured final report. Whether we use fast-start demonstrations, for interactive sessions, or for monitoring performance, the system enables us to handle complex research tasks with minimal intervention. Now we are able to further tweak or extend this setting by integrating custom agents, visualizing workflows, and even deploying them into real-world applications.
View full Code | Sponsorship Opportunities: Want to attract the most influential AI developers in the United States and Europe? Join our 1M+ monthly readers and 500K+ participating community members ecosystem. [Explore Sponsorship]
Asif Razzaq is CEO of Marktechpost Media Inc. As a visionary entrepreneur and engineer, ASIF is committed to harnessing the potential of artificial intelligence to achieve social benefits. His recent effort is to launch Marktechpost, an artificial intelligence media platform that has an in-depth coverage of machine learning and deep learning news that can sound both technically, both through technical voices and be understood by a wide audience. The platform has over 2 million views per month, demonstrating its popularity among its audience.
