En este tutorial, demostramos cómo construir una tubería de gráfico de conocimiento automatizado (kg) utilizando Langgraph y Networkx. La tubería simula una secuencia de agentes inteligentes que realizan tareas colaborativamente como recopilación de datos, extracción de entidades, identificación de relaciones, resolución de entidades y validación de gráficos. A partir de un tema proporcionado por el usuario, como la “inteligencia artificial”, el sistema extrae metódicamente entidades y relaciones relevantes, resuelve duplicados e integra la información en una estructura gráfica cohesiva. Al visualizar el gráfico de conocimiento final, los desarrolladores y los científicos de datos obtienen información clara sobre las interrelaciones complejas entre los conceptos, lo que hace que este enfoque sea altamente beneficioso para las aplicaciones en el análisis semántico, el procesamiento del lenguaje natural y la gestión del conocimiento.
!pip install langgraph langchain_coreInstalamos dos bibliotecas esenciales de Python: Langgraph, que se utiliza para crear y orquestar flujos de trabajo computacionales basados en agentes, y Langchain Core, que proporciona clases y utilidades fundamentales para construir aplicaciones con modelos de idiomas. Estas bibliotecas permiten una integración perfecta de los agentes en tuberías de datos inteligentes.
import re
import networkx as nx
import matplotlib.pyplot as plt
from typing import TypedDict, List, Tuple, Dict, Any
from langchain_core.messages import HumanMessage, AIMessage
from langgraph.graph import StateGraph, ENDImportamos bibliotecas esenciales para construir una tubería de gráficos de conocimiento automatizado. Incluye RE para procesamiento de texto regular basado en la expresión, NetworkX y Matplotlib para crear y visualizar gráficos, tipEddict y anotaciones de tipificación para el manejo de datos estructurados, y Langgraph junto con Langchain_core para orquestar la interacción entre los agentes de IA dentro del flujo de trabajo.
class KGState(TypedDict):
topic: str
raw_text: str
entities: List[str]
relations: List[Tuple[str, str, str]]
resolved_relations: List[Tuple[str, str, str]]
graph: Any
validation: Dict[str, Any]
messages: List[Any]
current_agent: strDefinimos un tipo de datos estructurado, KgState, utilizando el Typeddict de Python. Describe el esquema para administrar el estado en diferentes pasos de la tubería de gráficos de conocimiento. Incluye detalles como el tema elegido, el texto reunido, las entidades y las relaciones identificadas, los duplicados resueltos, el objeto de gráfico construido, los resultados de validación, los mensajes de interacción y el seguimiento del agente actualmente activo.
def data_gatherer(state: KGState) -> KGState:
topic = state["topic"]
print(f"📚 Data Gatherer: Searching for information about '{topic}'")
collected_text = f"{topic} is an important concept. It relates to various entities like EntityA, EntityB, and EntityC. EntityA influences EntityB. EntityC is a type of EntityB."
state["messages"].append(AIMessage(content=f"Collected raw text about {topic}"))
state["raw_text"] = collected_text
state["current_agent"] = "entity_extractor"
return stateEsta función, data_gatherer, actúa como el primer paso en la tubería. Simula la recopilación de datos de texto sin procesar sobre un tema proporcionado (almacenado en el estado[“topic”]). Luego almacena estos datos simulados en estado[“raw_text”]agrega un mensaje que indica la finalización de la recopilación de datos y actualiza el estado de la tubería configurando el siguiente agente (entity_extractor) como activo.
def entity_extractor(state: KGState) -> KGState:
print("🔍 Entity Extractor: Identifying entities in the text")
text = state["raw_text"]
entities = re.findall(r"Entity[A-Z]", text)
entities = [state["topic"]] + entities
state["entities"] = list(set(entities))
state["messages"].append(AIMessage(content=f"Extracted entities: {state['entities']}"))
print(f" Found entities: {state['entities']}")
state["current_agent"] = "relation_extractor"
return stateLa función Entity_Extractor identifica las entidades del texto sin procesar recopilado utilizando un patrón de expresión regular simple que coincide con términos como “entidad”, “entityb”, etc. También incluye el tema principal como entidad y garantiza la singularidad al convertir la lista en un conjunto. Las entidades extraídas se almacenan en el estado, un mensaje AI registra el resultado y la tubería avanza al agente Relation_Extractor.
def relation_extractor(state: KGState) -> KGState:
print("🔗 Relation Extractor: Identifying relationships between entities")
text = state["raw_text"]
entities = state["entities"]
relations = []
relation_patterns = [
(r"([A-Za-z]+) relates to ([A-Za-z]+)", "relates_to"),
(r"([A-Za-z]+) influences ([A-Za-z]+)", "influences"),
(r"([A-Za-z]+) is a type of ([A-Za-z]+)", "is_type_of")
]
for e1 in entities:
for e2 in entities:
if e1 != e2:
for pattern, rel_type in relation_patterns:
if re.search(f"{e1}.*{rel_type}.*{e2}", text.replace("_", " "), re.IGNORECASE) or \
re.search(f"{e1}.*{e2}", text, re.IGNORECASE):
relations.append((e1, rel_type, e2))
state["relations"] = relations
state["messages"].append(AIMessage(content=f"Extracted relations: {relations}"))
print(f" Found relations: {relations}")
state["current_agent"] = "entity_resolver"
return stateLa función Relation_Extractor detecta relaciones semánticas entre entidades dentro del texto sin procesar. Utiliza patrones regex predefinidos para identificar frases como “influencias” o “es un tipo de” entre los pares de entidades. Cuando se encuentra una coincidencia, agrega la relación correspondiente como un triple (sujeto, predicado, objeto) a la lista de relaciones. Estas relaciones extraídas se almacenan en el estado, se registra un mensaje para la comunicación del agente y el control se mueve al siguiente agente: Entity_Resolver.
def entity_resolver(state: KGState) -> KGState:
print("🔄 Entity Resolver: Resolving duplicate entities")
entity_map = {}
for entity in state["entities"]:
canonical_name = entity.lower().replace(" ", "_")
entity_map[entity] = canonical_name
resolved_relations = []
for s, p, o in state["relations"]:
s_resolved = entity_map.get(s, s)
o_resolved = entity_map.get(o, o)
resolved_relations.append((s_resolved, p, o_resolved))
state["resolved_relations"] = resolved_relations
state["messages"].append(AIMessage(content=f"Resolved relations: {resolved_relations}"))
state["current_agent"] = "graph_integrator"
return stateLa función Entity_Resolver estandariza los nombres de las entidades para evitar la duplicación e inconsistencias. Crea un mapeo (entity_map) al convertir cada entidad en minúsculas y reemplazar espacios con subrayos. Luego, este mapeo se aplica a todos los sujetos y objetos en las relaciones extraídas para producir relaciones resueltas. Estos triples normalizados se agregan al estado, se registra un mensaje de confirmación y el control se pasa al agente Graph_integrator.
def graph_integrator(state: KGState) -> KGState:
print("📊 Graph Integrator: Building the knowledge graph")
G = nx.DiGraph()
for s, p, o in state["resolved_relations"]:
if not G.has_node(s):
G.add_node(s)
if not G.has_node(o):
G.add_node(o)
G.add_edge(s, o, relation=p)
state["graph"] = G
state["messages"].append(AIMessage(content=f"Built graph with {len(G.nodes)} nodes and {len(G.edges)} edges"))
state["current_agent"] = "graph_validator"
return stateLa función Graph_integrator construye el gráfico de conocimiento real que usa networkx.digraph () admite relaciones dirigidas. Se itera sobre los triples resueltos (sujeto, predicado, objeto), asegura que existan ambos nodos, y luego agrega un borde dirigido con la relación como metadatos. El gráfico resultante se guarda en el estado, se agrega un mensaje resumido y la tubería pasa al agente Graph_Validator para la validación final.
def graph_validator(state: KGState) -> KGState:
print("✅ Graph Validator: Validating knowledge graph")
G = state["graph"]
validation_report = {
"num_nodes": len(G.nodes),
"num_edges": len(G.edges),
"is_connected": nx.is_weakly_connected(G) if G.nodes else False,
"has_cycles": not nx.is_directed_acyclic_graph(G) if G.nodes else False
}
state["validation"] = validation_report
state["messages"].append(AIMessage(content=f"Validation report: {validation_report}"))
print(f" Validation report: {validation_report}")
state["current_agent"] = END
return stateLa función Graph_Validator realiza una verificación de salud básica en el gráfico de conocimiento construido. Compila un informe de validación que contiene el número de nodos y bordes, si el gráfico está débilmente conectado (es decir, cada nodo es accesible si se ignora la dirección) y si el gráfico contiene ciclos. Este informe se agrega al estado y se registra como un mensaje AI. Una vez que se completa la validación, la tubería se marca según lo finalizado al configurar el final de la finalización.
def router(state: KGState) -> str:
return state["current_agent"]
def visualize_graph(graph):
plt.figure(figsize=(10, 6))
pos = nx.spring_layout(graph)
nx.draw(graph, pos, with_labels=True, node_color="skyblue", node_size=1500, font_size=10)
edge_labels = nx.get_edge_attributes(graph, 'relation')
nx.draw_networkx_edge_labels(graph, pos, edge_labels=edge_labels)
plt.title("Knowledge Graph")
plt.tight_layout()
plt.show()
La función del enrutador dirige la tubería al siguiente agente en función del campo Current_Agent en el estado. Mientras tanto, la función Visualize_Graph utiliza matplotlib y NetworkX para mostrar el gráfico de conocimiento final, mostrando nodos, bordes y relaciones etiquetadas para una comprensión visual intuitiva.
def build_kg_graph():
workflow = StateGraph(KGState)
workflow.add_node("data_gatherer", data_gatherer)
workflow.add_node("entity_extractor", entity_extractor)
workflow.add_node("relation_extractor", relation_extractor)
workflow.add_node("entity_resolver", entity_resolver)
workflow.add_node("graph_integrator", graph_integrator)
workflow.add_node("graph_validator", graph_validator)
workflow.add_conditional_edges("data_gatherer", router,
{"entity_extractor": "entity_extractor"})
workflow.add_conditional_edges("entity_extractor", router,
{"relation_extractor": "relation_extractor"})
workflow.add_conditional_edges("relation_extractor", router,
{"entity_resolver": "entity_resolver"})
workflow.add_conditional_edges("entity_resolver", router,
{"graph_integrator": "graph_integrator"})
workflow.add_conditional_edges("graph_integrator", router,
{"graph_validator": "graph_validator"})
workflow.add_conditional_edges("graph_validator", router,
{END: END})
workflow.set_entry_point("data_gatherer")
return workflow.compile()La función build_kg_graph define el flujo de trabajo del gráfico de conocimiento completo usando Langgraph. Agrega secuencialmente a cada agente como un nodo, desde la recopilación de datos hasta la validación de gráficos, y los conecta a través de transiciones condicionales basadas en el agente actual. El punto de entrada se establece en data_gatherer, y el gráfico se compila en un flujo de trabajo ejecutable que guía la tubería automatizada de principio a fin.
def run_knowledge_graph_pipeline(topic):
print(f"🚀 Starting knowledge graph pipeline for: {topic}")
initial_state = {
"topic": topic,
"raw_text": "",
"entities": [],
"relations": [],
"resolved_relations": [],
"graph": None,
"validation": {},
"messages": [HumanMessage(content=f"Build a knowledge graph about {topic}")],
"current_agent": "data_gatherer"
}
kg_app = build_kg_graph()
final_state = kg_app.invoke(initial_state)
print(f"✨ Knowledge graph construction complete for: {topic}")
return final_stateLa función run_knowledge_graph_pipeline inicializa la tubería configurando un diccionario de estado vacío con el tema proporcionado. Construye el flujo de trabajo usando build_kg_graph (), luego lo ejecuta invocando el gráfico compilado con el estado inicial. A medida que cada agente procesa los datos, el estado evoluciona y el resultado final contiene el gráfico de conocimiento completo, validado y listo para su uso.
if __name__ == "__main__":
topic = "Artificial Intelligence"
result = run_knowledge_graph_pipeline(topic)
visualize_graph(result["graph"])Finalmente, este bloque sirve como el punto de entrada del script. Cuando se ejecuta directamente, desencadena la tubería del gráfico de conocimiento para el tema “Inteligencia artificial”, se ejecuta a través de todas las etapas de agentes y finalmente visualiza el gráfico resultante utilizando la función Visualize_Graph (). Proporciona una demostración de extremo a extremo de la generación de gráficos de conocimiento automatizado.
En conclusión, hemos aprendido cómo integrar a la perfección múltiples agentes especializados en una tubería de gráficos de conocimiento cohesivo a través de este enfoque estructurado, aprovechando a Langgraph y NetworkX. Este flujo de trabajo automatiza los procesos de extracción de entidad y relación y visualiza relaciones complejas, ofreciendo una representación clara y procesable de la información recopilada. Al ajustar y mejorar los agentes individuales, como el empleo de métodos de reconocimiento de entidades más sofisticados o integrar fuentes de datos en tiempo real, este marco fundamental se puede escalar y personalizarse para tareas de construcción de gráficos de conocimiento avanzados en varios dominios.
Mira el Cuaderno de colab. Todo el crédito por esta investigación va a los investigadores de este proyecto. Además, siéntete libre de seguirnos Gorjeo Y no olvides unirte a nuestro 90k+ ml de subreddit.
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.