En este tutorial, construimos un sistema de equipo de investigación de múltiples agentes de múltiples agentes utilizando Langgraph y la API Gemini de Google. Utilizamos agentes, investigadores, analistas, escritor y supervisor específicos de roles, cada uno responsable de una parte distinta de la tubería de investigación. Juntos, estos agentes recopilan datos, analizan ideas, sintetizan un informe y coordinan el flujo de trabajo. También incorporamos características como persistencia de memoria, coordinación de agentes, agentes personalizados y monitoreo del rendimiento. Al final de la configuración, podemos ejecutar sesiones de investigación automatizadas e inteligentes que generan informes estructurados sobre cualquier tema dado.
!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_KEYComenzamos instalando las bibliotecas necesarias, incluidas Langgraph y la integración de Google Gemini de Langchain. Luego, importamos los módulos esenciales y configuramos nuestro entorno ingresando de forma segura la tecla API de Google utilizando el módulo GetPass. Esto asegura que podamos autenticar nuestro Gemini LLM sin exponer la clave en el código.
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"]
)
Definimos dos clases de typeddict para mantener el estado estructurado y las respuestas compartidas en todos los agentes del langgraph. AgentState rastrea los mensajes, el estado de flujo de trabajo, el tema y los hallazgos recopilados, mientras que AgentResponse estandariza la salida de cada agente. También creamos una función de ayuda para iniciar el Gemini LLM con un modelo y temperatura especificados, asegurando un comportamiento constante en todos los agentes.
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_agentAhora creamos nuestro primer agente especializado, el especialista en investigación IA. Se solicita a este agente que analice profundamente un tema dado, extraiga áreas clave de interés y sugiera instrucciones para una mayor exploración. Usando una CHATPromptTemplate, definimos su comportamiento y lo conectamos con nuestro Gemini LLM. La función Research_Agent ejecuta esta lógica, actualiza el estado compartido con hallazgos y mensajes, y pasa el control al siguiente agente en línea, el analista.
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_agentAhora definimos la IA del analista de datos, que se sumerge más profundamente en los resultados de la investigación generados por el agente anterior. Este agente identifica patrones clave, tendencias y métricas, que ofrece ideas procesables respaldadas por evidencia. Utilizando un mensaje de sistema personalizado y el Gemini LLM, la función Analyst_Agent enriquece al estado con análisis estructurado, preparando las bases para que el escritor de informes sintetice todo en un documento final.
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
Ahora creamos la IA de informes de informes, que es responsable de transformar la investigación y el análisis recopilados en un documento pulido y estructurado. Este agente sintetiza todas las ideas anteriores sobre un informe claro y profesional con un resumen ejecutivo, hallazgos detallados y conclusiones. Al invocar el modelo Gemini con un aviso estructurado, el agente del escritor actualiza el informe final en el estado compartido y el control de las manos al agente del supervisor para su revisión.
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_agentAhora traemos la IA del supervisor, que supervisa y orquesta todo el flujo de trabajo de múltiples agentes. Este agente evalúa el progreso actual, sabiendo qué miembro del equipo acaba de terminar su tarea, y decide inteligentemente el siguiente paso: si continuar con la investigación, proceder a análisis, iniciar la redacción de informes o marcar el proyecto como completo. Al analizar el contexto de la conversación y utilizar Gemini para el razonamiento, el agente del supervisor garantiza transiciones suaves y control de calidad a lo largo de la tubería de investigación.
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 NoneMira el completo Codos
Ahora ensamblamos y ejecutamos todo el flujo de trabajo de múltiples agentes usando Langgraph. Primero, definimos el gráfico del equipo de investigación, que consiste en nodos para cada agente, investigador, analista, escritor y supervisor, conectado por transiciones lógicas. Luego, compilamos este gráfico con memoria usando MemorySaver para persistir en el historial de conversación. Finalmente, la función run_research_team () inicializa el proceso con un tema y transmite la ejecución paso a paso, lo que nos permite rastrear la contribución de cada agente en tiempo real. Esta orquestación garantiza una tubería de investigación colaborativa totalmente automatizada.
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_agentMira el completo Codos
Envolvemos nuestro sistema con tiempo de ejecución y capacidades de personalización. El bloque principal nos permite activar una ejecución de investigación directamente, lo que la hace perfecta para probar la tubería con un tema del mundo real, como la inteligencia artificial en la atención médica. Para un uso más dinámico, el Interactive_Research_Session () habilita múltiples consultas de temas en un bucle, simulando la exploración en tiempo real. Por último, la función create_custom_agent () nos permite integrar nuevos agentes con roles e instrucciones únicas, haciendo que el marco sea flexible y extensible para flujos de trabajo especializados.
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()
Finalizamos el sistema agregando utilidades potentes para la visualización de gráficos, el monitoreo del rendimiento y una demostración de inicio rápido. La función Visualize_Graph () proporciona una visión general estructural de las conexiones de agentes, ideal para fines de depuración o presentación. El monitor_research_performance () rastrea tiempo de ejecución, volumen de mensajes y tamaño del informe, ayudándonos a evaluar la eficiencia del sistema. Finalmente, Quick_Start_Demo () ejecuta múltiples temas de investigación de muestra en secuencia, mostrando cuán sin problemas los agentes colaboran para generar informes perspicaces.
En conclusión, hemos construido y probado con éxito un marco de asistente de investigación de IA modular totalmente funcional y modular utilizando Langgraph. Con los roles de agentes claros y el enrutamiento de tareas automatizadas, simplificamos la investigación de la entrada de temas sin procesar hasta un informe final bien estructurado. Ya sea que usemos la demostración de inicio rápido, ejecutemos sesiones interactivas o monitoreamos el rendimiento, este sistema nos permite manejar tareas de investigación complejas con una intervención mínima. Ahora estamos equipados para adaptar o extender esta configuración aún más integrando agentes personalizados, visualizando flujos de trabajo o incluso implementándola en aplicaciones del mundo real.
Mira el completo Codos | Oportunidad de patrocinio: ¿Quieres llegar a los desarrolladores de IA más influyentes en los Estados Unidos y Europa? Únase a nuestro ecosistema de lectores mensuales de 1M+ y 500k+ miembros de la comunidad comprometidos. [Explore Sponsorship]
Asif Razzaq es el CEO de MarktechPost Media Inc .. Como empresario e ingeniero visionario, ASIF se compromete a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, MarktechPost, que se destaca por su cobertura profunda de noticias de aprendizaje automático y de aprendizaje profundo que es técnicamente sólido y fácilmente comprensible por una audiencia amplia. La plataforma cuenta con más de 2 millones de vistas mensuales, ilustrando su popularidad entre el público.