En este tutorial, proporcionamos una callejero experiencia para la implementación Langgraphun situación de orquestación de IA de IA aerodinámico, basado en gráficos, integrado sin problemas con API Claude de Anthrope. A través de un código ejecutable detallado optimizado para Google Colab, los desarrolladores aprenden cómo construir y visualizar flujos de trabajo de IA como nodos interconectados que realizan tareas distintas, como originar respuestas concisas, analizar críticamente las respuestas y componer automáticamente contenido técnico en el blog. La implementación compacta destaca la inmueble intuitiva de nodo de Langgraph. Puede administrar secuencias complejas de tareas de habla natural con Claude, desde escenarios básicos de suspensión de preguntas hasta tuberías avanzadas de procreación de contenido.
from getpass import getpass
import os
anthropic_key = getpass("Enter your Anthropic API key: ")
os.environ("ANTHROPIC_API_KEY") = anthropic_key
print("Key set:", "ANTHROPIC_API_KEY" in os.environ)Involucramos de forma segura a los usuarios que ingresen su tecla API antrópica utilizando el módulo GetPass de Python, asegurando que no se muestren datos confidenciales. Luego establece esta esencia como una variable de entorno (anthrope_api_key) y confirma el almacenamiento exitoso.
import os
import json
import requests
from typing import Dict, List, Any, Callable, Optional, Union
from dataclasses import dataclass, field
import networkx as nx
import matplotlib.pyplot as plt
from IPython.display import display, HTML, clear_outputImportamos bibliotecas esenciales para construir y visualizar flujos de trabajo de IA estructurados. Incluye módulos para manejar datos (JSON, solicitudes, dataclasses), creación y visualización de gráficos (NetworkX, matplotlib), pantalla de cuaderno interactivo (iPython.display) y anotaciones de tipo (estenotipia) para veterano claridad y mantenimiento.
try:
import anthropic
except ImportError:
print("Installing anthropic package...")
!pip install -q anthropic
import anthropic
from anthropic import AnthropicNos aseguramos de que el paquete antrópico de Python esté apto para su uso. Intenta importar el módulo y, si no se encuentra, instala automáticamente usando PIP en un entorno de Google Colab. Posteriormente de la instalación, importa el cliente antrópico, esencial para interactuar con los modelos Claude a través de la API antrópica. 4o
@dataclass
class NodeConfig:
name: str
function: Callable
inputs: List(str) = field(default_factory=list)
outputs: List(str) = field(default_factory=list)
config: Dict(str, Any) = field(default_factory=dict)Esta clase de datos NodeConfig define la estructura de cada nodo en el flujo de trabajo Langgraph. Cada nodo tiene un nombre, una función ejecutable, entradas y panorama opcionales, y un diccionario de configuración opcional para juntar parámetros adicionales. Esta configuración permite definiciones de nodos modulares y reutilizables para tareas de IA basadas en gráficos.
class LangGraph:
def __init__(self, api_key: Optional(str) = None):
self.api_key = api_key or os.environ.get("ANTHROPIC_API_KEY")
if not self.api_key:
from google.colab import userdata
try:
self.api_key = userdata.get('ANTHROPIC_API_KEY')
if not self.api_key:
raise ValueError("No API key found")
except:
print("No Anthropic API key found in environment variables or Colab secrets.")
self.api_key = input("Please enter your Anthropic API key: ")
if not self.api_key:
raise ValueError("Please provide an Anthropic API key")
self.client = Anthropic(api_key=self.api_key)
self.graph = nx.DiGraph()
self.nodes = {}
self.state = {}
def add_node(self, node_config: NodeConfig):
self.nodes(node_config.name) = node_config
self.graph.add_node(node_config.name)
for input_node in node_config.inputs:
if input_node in self.nodes:
self.graph.add_edge(input_node, node_config.name)
return self
def claude_node(self, name: str, prompt_template: str, model: str = "claude-3-7-sonnet-20250219",
inputs: List(str) = None, outputs: List(str) = None, system_prompt: str = None):
"""Convenience method to create a Claude API node"""
inputs = inputs or ()
outputs = outputs or (name + "_response")
def claude_fn(state, **kwargs):
prompt = prompt_template
for k, v in state.items():
if isinstance(v, str):
prompt = prompt.replace(f"{{{k}}}", v)
message_params = {
"model": model,
"max_tokens": 1000,
"messages": ({"role": "user", "content": prompt})
}
if system_prompt:
message_params("system") = system_prompt
response = self.client.messages.create(**message_params)
return response.content(0).text
node_config = NodeConfig(
name=name,
function=claude_fn,
inputs=inputs,
outputs=outputs,
config={"model": model, "prompt_template": prompt_template}
)
return self.add_node(node_config)
def transform_node(self, name: str, transform_fn: Callable,
inputs: List(str) = None, outputs: List(str) = None):
"""Add a data transformation node"""
inputs = inputs or ()
outputs = outputs or (name + "_output")
node_config = NodeConfig(
name=name,
function=transform_fn,
inputs=inputs,
outputs=outputs
)
return self.add_node(node_config)
def visualize(self):
"""Visualize the graph"""
plt.figure(figsize=(10, 6))
pos = nx.spring_layout(self.graph)
nx.draw(self.graph, pos, with_labels=True, node_color="lightblue",
node_size=1500, arrowsize=20, font_size=10)
plt.title("LangGraph Flow")
plt.tight_layout()
plt.show()
print("nGraph Structure:")
for node in self.graph.nodes():
successors = list(self.graph.successors(node))
if successors:
print(f" {node} → {', '.join(successors)}")
else:
print(f" {node} (endpoint)")
print()
def _get_execution_order(self):
"""Determine execution order based on dependencies"""
try:
return list(nx.topological_sort(self.graph))
except nx.NetworkXUnfeasible:
raise ValueError("Graph contains a cycle")
def execute(self, initial_state: Dict(str, Any) = None):
"""Execute the graph in topological order"""
self.state = initial_state or {}
execution_order = self._get_execution_order()
print("Executing LangGraph flow:")
for node_name in execution_order:
print(f"- Running node: {node_name}")
node = self.nodes(node_name)
inputs = {k: self.state.get(k) for k in node.inputs if k in self.state}
result = node.function(self.state, **inputs)
if len(node.outputs) == 1:
self.state(node.outputs(0)) = result
elif isinstance(result, (list, tuple)) and len(result) == len(node.outputs):
for i, output_name in enumerate(node.outputs):
self.state(output_name) = result(i)
print("Execution completed!")
return self.state
def run_example(question="What are the key benefits of using a graph-based architecture for AI workflows?"):
"""Run an example LangGraph flow with a predefined question"""
print(f"Running example with question: '{question}'")
graph = LangGraph()
def question_provider(state, **kwargs):
return question
graph.transform_node(
name="question_provider",
transform_fn=question_provider,
outputs=("user_question")
)
graph.claude_node(
name="question_answerer",
prompt_template="Answer this question clearly and concisely: {user_question}",
inputs=("user_question"),
outputs=("answer"),
system_prompt="You are a helpful AI assistant."
)
graph.claude_node(
name="answer_analyzer",
prompt_template="Analyze if this answer addresses the question well: Question: {user_question}nAnswer: {answer}",
inputs=("user_question", "answer"),
outputs=("analysis"),
system_prompt="You are a critical evaluator. Be brief but thorough."
)
graph.visualize()
result = graph.execute()
print("n" + "="*50)
print("EXECUTION RESULTS:")
print("="*50)
print(f"n🔍 QUESTION:n{result.get('user_question')}n")
print(f"📝 ANSWER:n{result.get('answer')}n")
print(f"✅ ANALYSIS:n{result.get('analysis')}")
print("="*50 + "n")
return graphLa clase Langgraph implementa un situación libertino para construir y ejecutar flujos de trabajo de IA basados en gráficos utilizando Claude de Anthrope. Permite a los usuarios explicar nodos modulares, ya sea indicaciones con claude o funciones de transformación personalizadas, conectarlos a través de dependencias, visualizar toda la tubería y ejecutarlos en orden topológico. La función Run_Example demuestra esto mediante la creación de un flujo simple de preguntas y evaluación, mostrando la claridad y la modularidad de la inmueble de Langgraph.
def run_advanced_example():
"""Run a more advanced example with multiple nodes for content generation"""
graph = LangGraph()
def topic_selector(state, **kwargs):
return "Graph-based AI systems"
graph.transform_node(
name="topic_selector",
transform_fn=topic_selector,
outputs=("topic")
)
graph.claude_node(
name="outline_generator",
prompt_template="Create a brief outline for a technical blog post about {topic}. Include 3-4 main sections only.",
inputs=("topic"),
outputs=("outline"),
system_prompt="You are a technical writer specializing in AI technologies."
)
graph.claude_node(
name="intro_writer",
prompt_template="Write an engaging introduction for a blog post with this outline: {outline}nTopic: {topic}",
inputs=("topic", "outline"),
outputs=("introduction"),
system_prompt="You are a technical writer. Write in a clear, engaging style."
)
graph.claude_node(
name="conclusion_writer",
prompt_template="Write a conclusion for a blog post with this outline: {outline}nTopic: {topic}",
inputs=("topic", "outline"),
outputs=("conclusion"),
system_prompt="You are a technical writer. Summarize key points and include a forward-looking statement."
)
def assembler(state, introduction, outline, conclusion, **kwargs):
return f"# {state('topic')}nn{introduction}nn## Outlinen{outline}nn## Conclusionn{conclusion}"
graph.transform_node(
name="content_assembler",
transform_fn=assembler,
inputs=("topic", "introduction", "outline", "conclusion"),
outputs=("final_content")
)
graph.visualize()
result = graph.execute()
print("n" + "="*50)
print("BLOG POST GENERATED:")
print("="*50 + "n")
print(result.get("final_content"))
print("n" + "="*50)
return graphLa función Run_Advanced_Example muestra un uso más sofisticado de Langgraph orquestando múltiples nodos con comestibles de Claude para originar una publicación de blog completa. Comienza seleccionando un tema, luego crea un esquema, una ingreso y una conclusión, todo utilizando indicaciones estructuradas de Claude. Finalmente, un nodo de transformación ensambla el contenido en una publicación de blog formateada. Este ejemplo demuestra cómo Langgraph puede automatizar tareas complejas de procreación de contenido de varios pasos utilizando nodos modulares y conectados en un flujo transparente y ejecutable.
print("1. Running simple question-answering example")
question = "What are the three main advantages of using graph-based AI architectures?"
simple_graph = run_example(question)
print("n2. Running advanced blog post creation example")
advanced_graph = run_advanced_example()
Finalmente, activamos la ejecución de entreambos flujos de trabajo de Langgraph definidos. Primero, ejecuta el simple ejemplo de respuesta de pregunta al tener lugar una pregunta predefinida a la función run_example (). Luego, inicia el flujo de trabajo de la procreación de publicaciones de blog más renovador usando run_advanced_example (). Juntos, estas llamadas demuestran la flexibilidad experiencia de Langgraph, desde interacciones básicas basadas en aviso hasta automatización de contenido de varios pasos utilizando la API Claude de Anthrope.
En conclusión, hemos implementado Langgraph integrado con la API Claude de Anthrope, que ilustra la facilidad de diseñar flujos de trabajo de IA modulares que aprovechen modelos de habla potentes en tuberías estructuradas basadas en gráficos. A través de la visualización de los flujos de tareas y la separación de las responsabilidades entre los nodos, como el procesamiento de preguntas, la evaluación analítica, el contenido de contenido y el ensamblaje, los desarrolladores obtienen experiencia experiencia en la creación de sistemas de IA escalables y mantenibles. Las dependencias de nodo claras de Langgraph y las sofisticadas capacidades del habla de Claude proporcionan una posibilidad válido para orquestar procesos de IA complejos, especialmente para la prototipos y la ejecución rápidas en entornos como Google Colab.
Mira el Cuaderno de colab. Todo el crédito por esta investigación va a los investigadores de este esquema. Encima, siéntete soberano 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 patrón e ingeniero quimérico, ASIF se compromete a disfrutar el potencial de la inteligencia químico para el acertadamente social. Su esfuerzo más flamante es el tirada de una plataforma de medios de inteligencia químico, MarktechPost, que se destaca por su cobertura profunda de telediario de formación inevitable y de formación 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 manifiesto.
