En este tutorial, demostramos cómo construir un agente inteligente de manejo de consultas de varios pasos usando Langgraph y Gemini 1.5 Flash. La idea central es estructurar el razonamiento de AI como un flujo de trabajo con estado, donde se pasa una consulta entrante a través de una serie de nodos con propósito: enrutamiento, análisis, investigación, generación de respuesta y validación. Cada nodo opera como un bloque funcional con un papel bien definido, lo que hace que el agente no solo sea reactivo sino analíticamente consciente. Usando el estado de Langgraph, orquestamos estos nodos para crear un sistema de bucle que pueda volver a analizar y mejorar su salida hasta que la respuesta se valida como completa o se alcanza un umbral de iteración máxima.
!pip install langgraph langchain-google-genai python-dotenvPrimero, el comando! Pip instale Langgraph Langchain-Google-Genai Python-Dotenv instala tres paquetes de Python esenciales para construir flujos de trabajo de agentes inteligentes. Langgraph permite la orquestación basada en gráficos de agentes de IA, Langchain-Google-Genai proporciona integración con los modelos Gemini de Google, y Python-Dotenv permite una carga segura de variables de entorno de archivos .env.
import os
from typing import Dict, Any, List
from dataclasses import dataclass
from langgraph.graph import Graph, StateGraph, END
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.schema import HumanMessage, SystemMessage
import json
os.environ["GOOGLE_API_KEY"] = "Use Your API Key Here"Importamos módulos y bibliotecas esenciales para construir flujos de trabajo de agentes, incluido ChatGoogleGenerativeai para interactuar con los modelos Gemini y StateGraph para la gestión del estado de conversación. La línea OS.[“GOOGLE_API_KEY”] = “Use su clave API aquí” asigna la clave API a una variable de entorno, lo que permite que el modelo Gemini autentique y genere respuestas.
@dataclass
class AgentState:
"""State shared across all nodes in the graph"""
query: str = ""
context: str = ""
analysis: str = ""
response: str = ""
next_action: str = ""
iteration: int = 0
max_iterations: int = 3Mira el Cuaderno aquí
Este dataclass de Agentstate define el estado compartido que persiste en diferentes nodos en un flujo de trabajo Langgraph. Rastrea los campos clave, incluida la consulta del usuario, el contexto recuperado, cualquier análisis realizado, la respuesta generada y la acción siguiente recomendada. También incluye un contador de iteración y un límite de max_iteraciones para controlar cuántas veces el flujo de trabajo puede recorrer, lo que permite el razonamiento iterativo o la toma de decisiones por parte del agente.
@dataclass
class AgentState:
"""State shared across all nodes in the graph"""
query: str = ""
context: str = ""
analysis: str = ""
response: str = ""
next_action: str = ""
iteration: int = 0
max_iterations: int = 3
This AgentState dataclass defines the shared state that persists across different nodes in a LangGraph workflow. It tracks key fields, including the user's query, retrieved context, any analysis performed, the generated response, and the recommended next action. It also includes an iteration counter and a max_iterations limit to control how many times the workflow can loop, enabling iterative reasoning or decision-making by the agent.
class GraphAIAgent:
def __init__(self, api_key: str = None):
if api_key:
os.environ["GOOGLE_API_KEY"] = api_key
self.llm = ChatGoogleGenerativeAI(
model="gemini-1.5-flash",
temperature=0.7,
convert_system_message_to_human=True
)
self.analyzer = ChatGoogleGenerativeAI(
model="gemini-1.5-flash",
temperature=0.3,
convert_system_message_to_human=True
)
self.graph = self._build_graph()
def _build_graph(self) -> StateGraph:
"""Build the LangGraph workflow"""
workflow = StateGraph(AgentState)
workflow.add_node("router", self._router_node)
workflow.add_node("analyzer", self._analyzer_node)
workflow.add_node("researcher", self._researcher_node)
workflow.add_node("responder", self._responder_node)
workflow.add_node("validator", self._validator_node)
workflow.set_entry_point("router")
workflow.add_edge("router", "analyzer")
workflow.add_conditional_edges(
"analyzer",
self._decide_next_step,
{
"research": "researcher",
"respond": "responder"
}
)
workflow.add_edge("researcher", "responder")
workflow.add_edge("responder", "validator")
workflow.add_conditional_edges(
"validator",
self._should_continue,
{
"continue": "analyzer",
"end": END
}
)
return workflow.compile()
def _router_node(self, state: AgentState) -> Dict[str, Any]:
"""Route and categorize the incoming query"""
system_msg = """You are a query router. Analyze the user's query and provide context.
Determine if this is a factual question, creative request, problem-solving task, or analysis."""
messages = [
SystemMessage(content=system_msg),
HumanMessage(content=f"Query: {state.query}")
]
response = self.llm.invoke(messages)
return {
"context": response.content,
"iteration": state.iteration + 1
}
def _analyzer_node(self, state: AgentState) -> Dict[str, Any]:
"""Analyze the query and determine the approach"""
system_msg = """Analyze the query and context. Determine if additional research is needed
or if you can provide a direct response. Be thorough in your analysis."""
messages = [
SystemMessage(content=system_msg),
HumanMessage(content=f"""
Query: {state.query}
Context: {state.context}
Previous Analysis: {state.analysis}
""")
]
response = self.analyzer.invoke(messages)
analysis = response.content
if "research" in analysis.lower() or "more information" in analysis.lower():
next_action = "research"
else:
next_action = "respond"
return {
"analysis": analysis,
"next_action": next_action
}
def _researcher_node(self, state: AgentState) -> Dict[str, Any]:
"""Conduct additional research or information gathering"""
system_msg = """You are a research assistant. Based on the analysis, gather relevant
information and insights to help answer the query comprehensively."""
messages = [
SystemMessage(content=system_msg),
HumanMessage(content=f"""
Query: {state.query}
Analysis: {state.analysis}
Research focus: Provide detailed information relevant to the query.
""")
]
response = self.llm.invoke(messages)
updated_context = f"{state.context}\n\nResearch: {response.content}"
return {"context": updated_context}
def _responder_node(self, state: AgentState) -> Dict[str, Any]:
"""Generate the final response"""
system_msg = """You are a helpful AI assistant. Provide a comprehensive, accurate,
and well-structured response based on the analysis and context provided."""
messages = [
SystemMessage(content=system_msg),
HumanMessage(content=f"""
Query: {state.query}
Context: {state.context}
Analysis: {state.analysis}
Provide a complete and helpful response.
""")
]
response = self.llm.invoke(messages)
return {"response": response.content}
def _validator_node(self, state: AgentState) -> Dict[str, Any]:
"""Validate the response quality and completeness"""
system_msg = """Evaluate if the response adequately answers the query.
Return 'COMPLETE' if satisfactory, or 'NEEDS_IMPROVEMENT' if more work is needed."""
messages = [
SystemMessage(content=system_msg),
HumanMessage(content=f"""
Original Query: {state.query}
Response: {state.response}
Is this response complete and satisfactory?
""")
]
response = self.analyzer.invoke(messages)
validation = response.content
return {"context": f"{state.context}\n\nValidation: {validation}"}
def _decide_next_step(self, state: AgentState) -> str:
"""Decide whether to research or respond directly"""
return state.next_action
def _should_continue(self, state: AgentState) -> str:
"""Decide whether to continue iterating or end"""
if state.iteration >= state.max_iterations:
return "end"
if "COMPLETE" in state.context:
return "end"
if "NEEDS_IMPROVEMENT" in state.context:
return "continue"
return "end"
def run(self, query: str) -> str:
"""Run the agent with a query"""
initial_state = AgentState(query=query)
result = self.graph.invoke(initial_state)
return result["response"]Mira el Cuaderno aquí
La clase GraprayAgent define un flujo de trabajo de IA basado en Langgraph utilizando modelos Gemini para analizar, investigar, responder y validar las respuestas a las consultas de los usuarios. Utiliza nodos modulares, como enrutador, analizador, investigador, respondedor y validador, para razonar a través de tareas complejas, refinando respuestas a través de iteraciones controladas.
def main():
agent = GraphAIAgent("Use Your API Key Here")
test_queries = [
"Explain quantum computing and its applications",
"What are the best practices for machine learning model deployment?",
"Create a story about a robot learning to paint"
]
print("🤖 Graph AI Agent with LangGraph and Gemini")
print("=" * 50)
for i, query in enumerate(test_queries, 1):
print(f"\n📝 Query {i}: {query}")
print("-" * 30)
try:
response = agent.run(query)
print(f"🎯 Response: {response}")
except Exception as e:
print(f"❌ Error: {str(e)}")
print("\n" + "="*50)
if __name__ == "__main__":
main()
Finalmente, la función Main () inicializa el GraprayAgent con una tecla API de Gemini y la ejecuta en un conjunto de consultas de prueba que cubren tareas técnicas, estratégicas y creativas. Imprime cada consulta y la respuesta generada por IA, que muestra cómo el agente impulsado por Langgraph procesa diversos tipos de entrada utilizando las capacidades de razonamiento y generación de Gemini.
En conclusión, al combinar la máquina de estado estructurada de Langgraph con el poder de la inteligencia conversacional de Géminis, este agente representa un nuevo paradigma en la ingeniería de flujo de trabajo de IA, uno que refleja los ciclos de investigación, análisis y validación de razonamiento humano. El tutorial proporciona una plantilla modular y extensible para desarrollar agentes de IA avanzados que puedan manejar de manera autónoma varias tareas, desde responder consultas complejas hasta generar contenido creativo.
Mira el Cuaderno aquí. Todo el crédito por esta investigación va a los investigadores de este proyecto.
🆕 ¿Sabías? MarktechPost es la plataforma de medios AI de más rápido crecimiento, remunerada por más de 1 millón de lectores mensuales. Reserve una llamada de estrategia para discutir los objetivos de su campaña. Además, siéntete libre de seguirnos Gorjeo Y no olvides unirte a nuestro 95k+ ml de subreddit y suscribirse a Nuestro boletín.
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.