A Step-by-Step Implementation Tutorial for Constructing Modular AI Workflows Utilizing Anthropic’s Claude Sonnet 3.7 by way of API and LangGraph

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)

import os
import json
import requests
from typing import Dict, Record, Any, Callable, Elective, Union
from dataclasses import dataclass, area
import networkx as nx
import matplotlib.pyplot as plt
from IPython.show import show, HTML, clear_output

strive:
    import anthropic
besides ImportError:
    print("Putting in anthropic package deal...")
    !pip set up -q anthropic
    import anthropic


from anthropic import Anthropic

@dataclass
class NodeConfig:
    title: str
    perform: Callable
    inputs: Record[str] = area(default_factory=listing)
    outputs: Record[str] = area(default_factory=listing)
    config: Dict[str, Any] = area(default_factory=dict)

class LangGraph:
    def __init__(self, api_key: Elective[str] = None):
        self.api_key = api_key or os.environ.get("ANTHROPIC_API_KEY")
        if not self.api_key:
            from google.colab import userdata
            strive:
                self.api_key = userdata.get('ANTHROPIC_API_KEY')
                if not self.api_key:
                    elevate ValueError("No API key discovered")
            besides:
                print("No Anthropic API key present in setting variables or Colab secrets and techniques.")
                self.api_key = enter("Please enter your Anthropic API key: ")
                if not self.api_key:
                    elevate ValueError("Please present an Anthropic API key")
       
        self.consumer = 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.title)
        for input_node in node_config.inputs:
            if input_node in self.nodes:
                self.graph.add_edge(input_node, node_config.title)
        return self
   
    def claude_node(self, title: str, prompt_template: str, mannequin: str = "claude-3-7-sonnet-20250219",
                   inputs: Record[str] = None, outputs: Record[str] = None, system_prompt: str = None):
        """Comfort methodology to create a Claude API node"""
        inputs = inputs or []
        outputs = outputs or [name + "_response"]
       
        def claude_fn(state, **kwargs):
            immediate = prompt_template
            for ok, v in state.objects():
                if isinstance(v, str):
                    immediate = immediate.exchange(f"{{{ok}}}", v)
           
            message_params = {
                "mannequin": mannequin,
                "max_tokens": 1000,
                "messages": [{"role": "user", "content": prompt}]
            }
           
            if system_prompt:
                message_params["system"] = system_prompt
               
            response = self.consumer.messages.create(**message_params)
            return response.content material[0].textual content
       
        node_config = NodeConfig(
            title=title,
            perform=claude_fn,
            inputs=inputs,
            outputs=outputs,
            config={"mannequin": mannequin, "prompt_template": prompt_template}
        )
        return self.add_node(node_config)
   
    def transform_node(self, title: str, transform_fn: Callable,
                      inputs: Record[str] = None, outputs: Record[str] = None):
        """Add an information transformation node"""
        inputs = inputs or []
        outputs = outputs or [name + "_output"]
       
        node_config = NodeConfig(
            title=title,
            perform=transform_fn,
            inputs=inputs,
            outputs=outputs
        )
        return self.add_node(node_config)
   
    def visualize(self):
        """Visualize the graph"""
        plt.determine(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 Move")
        plt.tight_layout()
        plt.present()
       
        print("nGraph Construction:")
        for node in self.graph.nodes():
            successors = listing(self.graph.successors(node))
            if successors:
                print(f"  {node} → {', '.be a part of(successors)}")
            else:
                print(f"  {node} (endpoint)")
        print()
   
    def _get_execution_order(self):
        """Decide execution order based mostly on dependencies"""
        strive:
            return listing(nx.topological_sort(self.graph))
        besides nx.NetworkXUnfeasible:
            elevate ValueError("Graph incorporates 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 stream:")
       
        for node_name in execution_order:
            print(f"- Working node: {node_name}")
            node = self.nodes[node_name]
            inputs = {ok: self.state.get(ok) for ok in node.inputs if ok in self.state}
           
            end result = node.perform(self.state, **inputs)
           
            if len(node.outputs) == 1:
                self.state[node.outputs[0]] = end result
            elif isinstance(end result, (listing, tuple)) and len(end result) == len(node.outputs):
                for i, output_name in enumerate(node.outputs):
                    self.state[output_name] = end result[i]
       
        print("Execution accomplished!")
        return self.state


def run_example(query="What are the important thing advantages of utilizing a graph-based structure for AI workflows?"):
    """Run an instance LangGraph stream with a predefined query"""
    print(f"Working instance with query: '{query}'")
   
    graph = LangGraph()
   
    def question_provider(state, **kwargs):
        return query
   
    graph.transform_node(
        title="question_provider",
        transform_fn=question_provider,
        outputs=["user_question"]
    )
   
    graph.claude_node(
        title="question_answerer",
        prompt_template="Reply this query clearly and concisely: {user_question}",
        inputs=["user_question"],
        outputs=["answer"],
        system_prompt="You're a useful AI assistant."
    )
   
    graph.claude_node(
        title="answer_analyzer",
        prompt_template="Analyze if this reply addresses the query nicely: Query: {user_question}nAnswer: {reply}",
        inputs=["user_question", "answer"],
        outputs=["analysis"],
        system_prompt="You're a essential evaluator. Be transient however thorough."
    )
   
    graph.visualize()
   
    end result = graph.execute()
   
    print("n" + "="*50)
    print("EXECUTION RESULTS:")
    print("="*50)
    print(f"n🔍 QUESTION:n{end result.get('user_question')}n")
    print(f"📝 ANSWER:n{end result.get('reply')}n")
    print(f"✅ ANALYSIS:n{end result.get('evaluation')}")
    print("="*50 + "n")
   
    return graph

def run_advanced_example():
    """Run a extra superior instance with a number of nodes for content material technology"""
    graph = LangGraph()
   
    def topic_selector(state, **kwargs):
        return "Graph-based AI methods"
   
    graph.transform_node(
        title="topic_selector",
        transform_fn=topic_selector,
        outputs=["topic"]
    )
   
    graph.claude_node(
        title="outline_generator",
        prompt_template="Create a short define for a technical weblog put up about {subject}. Embody 3-4 principal sections solely.",
        inputs=["topic"],
        outputs=["outline"],
        system_prompt="You're a technical author specializing in AI applied sciences."
    )
   
    graph.claude_node(
        title="intro_writer",
        prompt_template="Write a fascinating introduction for a weblog put up with this define: {define}nTopic: {subject}",
        inputs=["topic", "outline"],
        outputs=["introduction"],
        system_prompt="You're a technical author. Write in a transparent, participating fashion."
    )
   
    graph.claude_node(
        title="conclusion_writer",
        prompt_template="Write a conclusion for a weblog put up with this define: {define}nTopic: {subject}",
        inputs=["topic", "outline"],
        outputs=["conclusion"],
        system_prompt="You're a technical author. Summarize key factors and embody a forward-looking assertion."
    )
   
    def assembler(state, introduction, define, conclusion, **kwargs):
        return f"# {state['topic']}nn{introduction}nn## Outlinen{define}nn## Conclusionn{conclusion}"
   
    graph.transform_node(
        title="content_assembler",
        transform_fn=assembler,
        inputs=["topic", "introduction", "outline", "conclusion"],
        outputs=["final_content"]
    )
   
    graph.visualize()
    end result = graph.execute()
   
    print("n" + "="*50)
    print("BLOG POST GENERATED:")
    print("="*50 + "n")
    print(end result.get("final_content"))
    print("n" + "="*50)
   
    return graph

print("1. Working easy question-answering instance")
query = "What are the three principal benefits of utilizing graph-based AI architectures?"
simple_graph = run_example(query)


print("n2. Working superior weblog put up creation instance")
advanced_graph = run_advanced_example()