Part 03 | Business Nodes: The True Executors of the Graph

⏱ Est. reading time: 20 min Updated on 5/7/2026

🎯 Learning Objectives for This Session

Hello everyone, I'm your old friend Lao Zhang, someone who's been navigating the AI architecture circles for a decade—losing quite a bit of hair along the way, but with undiminished passion. In the last session, we got our first glimpse into the LangGraph universe, learning how it uses graph structures to orchestrate complex AI workflows. But a mere skeletal graph isn't enough; it needs flesh and blood, and workers who can actually get things done. In this session, we'll dive deep into LangGraph's core execution units: Business Nodes.

By the end of this session, you will:

  1. Thoroughly understand the essence of a Node in LangGraph: It's not just a function; it's a "department" or "expert" in our AI agency responsible for executing specific business logic.
  2. Master writing basic Worker Nodes: Learn to encapsulate LLM reasoning logic so our Planner agent can truly spring into action and start thinking.
  3. Master in-node data modification and state management: Understand how a Node interacts with the Graph's global State to ensure smooth information flow between different Agents.
  4. Build the first core node of the Agency project: Construct the Planner node for our "Omnipotent AI Content Creation Agency," taking the first step toward building a complex multi-agent workflow.

📖 Core Concepts

Imagine our "Omnipotent AI Content Creation Agency" as a high-tech company. The company has various departments: Planning, R&D, Content Creation, Editorial, etc. Each department has its unique function—receiving tasks, processing information, and passing the results to the next department.

In the LangGraph universe, these "departments" or "experts" are our Nodes.

At its core, a Node is simply a Callable (like a function). This function receives the current Graph State (think of it as the company's shared "project progress report" or "task board"), executes its own business logic (e.g., calling an LLM to reason, querying a database, invoking an external tool), and returns an update to the State.

Core Idea: A Node receives the State, processes it, and returns a "delta" (difference) of the State.

Why a "delta"? This is a brilliant aspect of LangGraph and many functional programming paradigms: Immutability of State. Nodes don't directly mutate the global State. Instead, they generate a new State fragment, and the LangGraph framework merges these fragments into the global State. This drastically reduces the complexity of concurrency and debugging—you always know "what changes a node made" rather than "what a node mutated."

Nodes in Our AI Content Agency

In our "Omnipotent AI Content Creation Agency," the first heavyweight character to take the stage is the Planner. Its responsibility is to receive the client's raw requirements and break them down into actionable, specific content creation tasks.

  • Input: Initial client requirement (topic).
  • Process: The Planner node calls a Large Language Model (LLM) to brainstorm a content outline, identify the target audience, extract core viewpoints, and plan subsequent creation steps based on the topic.
  • Output: Updates the State by adding plan_outline, target_audience, research_tasks, etc.

This process perfectly illustrates the lifecycle of a Node.

Mermaid Diagram: Introducing the Planner Node

Let's use a Mermaid diagram to visualize how the Planner node operates within our Graph. It will be our first core node, laying the foundation for the entire content creation workflow.

graph TD
    A[User Input: Initial Creation Requirement] --> B{Initialize Graph State};
    B --> C(Planner Node);
    
    subgraph Planner Node Internal Logic
        C --> C1[Receive Current State];
        C1 --> C2[Call LLM for Content Planning];
        C2 --> C3[Generate Outline, Research Tasks, etc.];
        C3 --> C4[Return State Update];
    end
    
    C --> D{Graph State Update};
    D --> E[Ready for Next Stage: e.g., Research Node];
    
    style A fill:#f9f,stroke:#333,stroke-width:2px;
    style B fill:#bbf,stroke:#333,stroke-width:2px;
    style C fill:#ccf,stroke:#333,stroke-width:2px;
    style C1 fill:#fcf,stroke:#333,stroke-width:1px;
    style C2 fill:#fcf,stroke:#333,stroke-width:1px;
    style C3 fill:#fcf,stroke:#333,stroke-width:1px;
    style C4 fill:#fcf,stroke:#333,stroke-width:1px;
    style D fill:#bbf,stroke:#333,stroke-width:2px;
    style E fill:#afa,stroke:#333,stroke-width:2px;

Diagram Explanation:

  • User Input: The starting point of everything, the raw requirement from the client.
  • Initialize Graph State: LangGraph creates an initial state based on our defined State structure.
  • Planner Node: This is the node we will focus on implementing in this session. It receives the current State.
  • Planner Node Internal Logic:
    • Receive Current State: Fetches all contextual information of the current workflow.
    • Call LLM for Content Planning: This is the core business logic. Here, we interact with the LLM, letting it play the role of a "Planner."
    • Generate Outline, Research Tasks, etc.: The LLM's output is structured and becomes the input for our subsequent Agents.
    • Return State Update: The node generates a dictionary containing the new information to be added to the State.
  • Graph State Update: The LangGraph framework merges the updates returned by the Planner Node into the global State.
  • Ready for Next Stage: The updated State is passed as input to the next node in the graph (e.g., the future Researcher Node).

Once you understand this flow, you've grasped the essence of LangGraph: Every node is a pure function that receives a state and returns a state update. The entire system achieves complex workflows through the routing and aggregation of these states.

💻 Hands-On Code Practice

Alright, enough theory—let's roll up our sleeves and get to work. Now, we will write the first core business node for our "Omnipotent AI Content Creation Agency": the Planner Node.

First, we need to define our Graph State. This State will be the shared information hub for all Agents. For convenience, we use TypedDict to define it, acting as a data structure with explicit fields and types.

# agency_state.py
from typing import TypedDict, List, Dict, Optional

class AgencyState(TypedDict):
    """
    Shared state for the AI Content Creation Agency.
    This TypedDict defines the data structure shared and manipulated by all Agents throughout the LangGraph workflow.
    """
    topic: str  # Initial content creation topic
    plan_outline: Optional[str]  # Content outline generated by the Planner
    target_audience: Optional[str] # Target audience identified by the Planner
    research_tasks: Optional[List[str]] # List of research tasks planned by the Planner
    research_results: Optional[Dict[str, str]] # Research results collected by the Researcher
    draft_content: Optional[str] # First draft written by the Writer
    editor_feedback: Optional[str] # Revision feedback provided by the Editor
    final_content: Optional[str] # Final finalized content

    # You can also add other fields, for example:
    # current_agent_name: str # Name of the currently executing Agent, useful for debugging or logging
    # error_message: Optional[str] # Error message, useful for error handling

Next, let's write the code for the Planner Node. It will be a Python function that receives the AgencyState and returns a dictionary representing the updates to the AgencyState.

We will use ChatOpenAI to simulate the LLM's capabilities. If you don't have an OpenAI API Key, you can use other LangChain-compatible LLMs, or simply return hardcoded mock data.

# nodes.py
import os
from typing import Dict, Any, List
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

# Load environment variables from the .env file, e.g., OPENAI_API_KEY
load_dotenv()

# Import our defined AgencyState
from agency_state import AgencyState

# Initialize the LLM
# We use GPT-4o here, but you can choose other models as needed
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

# --- 1. Planner Node ---
def planner_node(state: AgencyState) -> Dict[str, Any]:
    """
    Planner Node: Generates a content outline, target audience, and research tasks based on the initial topic.
    This node encapsulates the LLM reasoning logic to translate high-level requirements into a concrete execution plan.
    
    Args:
        state (AgencyState): The current global state.
                              Expected to contain the 'topic' field.

    Returns:
        Dict[str, Any]: A dictionary containing updates to the state.
                        Will add 'plan_outline', 'target_audience', 'research_tasks'.
    """
    print("\n--- Executing Planner Node ---")
    topic = state.get("topic")
    if not topic:
        raise ValueError("Missing 'topic' field in AgencyState; Planner cannot proceed.")

    # 1. Define the LLM Prompt Template
    # We want the LLM to not only provide an outline but also think about the audience and subsequent research points
    planner_prompt = ChatPromptTemplate.from_messages(
        [
            ("system", "You are an experienced content strategist and planner. Your task is to develop a detailed content creation plan based on the given topic."),
            ("human", """
            Please generate a detailed content creation plan for the following topic. The plan should include:
            1.  **Plan Outline**: Well-structured, containing at least 3-5 main sections and key points for each section.
            2.  **Target Audience**: A detailed description of the ideal reader for this content, including their interests, pain points, and knowledge level.
            3.  **Research Tasks**: A list of at least 3-5 specific research tasks to ensure the depth and accuracy of the content. These tasks should be actionable for the subsequent Researcher Agent.
            
            Topic: {topic}
            
            Please return the result in JSON format, for example:
            ```json
            {{
                "plan_outline": "...",
                "target_audience": "...",
                "research_tasks": ["...", "...", "..."]
            }}
            ```
            """)
        ]
    )

    # 2. Build the LLM Chain
    # Use .with_structured_output(dict) to ensure the LLM outputs a structured dictionary
    # Note: with_structured_output relies on the model's capabilities. If the model doesn't support it, you might need a custom parser.
    # For models like GPT-4o, it usually follows JSON formatting instructions very well.
    planner_chain = planner_prompt | llm.with_structured_output(
        schema={
            "type": "object",
            "properties": {
                "plan_outline": {"type": "string", "description": "Detailed content outline"},
                "target_audience": {"type": "string", "description": "Target audience description"},
                "research_tasks": {"type": "array", "items": {"type": "string"}, "description": "List of research tasks to be conducted"},
            },
            "required": ["plan_outline", "target_audience", "research_tasks"],
        }
    )

    # 3. Execute LLM Reasoning
    try:
        planning_output = planner_chain.invoke({"topic": topic})
        print("Planner successfully generated the plan.")
        # Directly return the LLM's output as the state update
        # with_structured_output already returns a dictionary, so we can use it directly
        return planning_output
    except Exception as e:
        print(f"Planner Node execution failed: {e}")
        # If the LLM call fails, we can choose to raise an exception or return an update containing an error message
        # Here we choose to return an error message so the Graph can handle it
        return {"error_message": f"Planner failed: {str(e)}"}

# --- Simulate Running the Planner Node ---
if __name__ == "__main__":
    print("--- Simulating Planner Node Execution ---")
    
    # 1. Simulate the initial state
    initial_state: AgencyState = {
        "topic": "Exploring the Core Advantages and Practical Use Cases of LangGraph in Building Multi-Agent Applications",
        "plan_outline": None,
        "target_audience": None,
        "research_tasks": None,
        "research_results": None,
        "draft_content": None,
        "editor_feedback": None,
        "final_content": None,
    }
    
    print(f"Initial State: {initial_state['topic']}")
    
    # 2. Call the Planner Node
    updated_fields = planner_node(initial_state)
    
    # 3. Merge updates into the state (The LangGraph framework will do this step)
    # Here we manually simulate the merge
    final_state = initial_state.copy()
    for key, value in updated_fields.items():
        # If it's a list or dict, more complex merge logic is needed; here we simplify by directly overwriting
        final_state[key] = value

    print("\n--- State Update After Planner Node Execution ---")
    print(f"Plan Outline: \n{final_state.get('plan_outline')}")
    print(f"Target Audience: {final_state.get('target_audience')}")
    print(f"Research Tasks: {final_state.get('research_tasks')}")
    print(f"Complete Final State: {final_state}")

    # Try another topic
    print("\n--- Simulating Planner Node Execution Again (New Topic) ---")
    initial_state_2: AgencyState = {
        "topic": "AI Ethics and Regulation: Challenges and Opportunities",
        "plan_outline": None,
        "target_audience": None,
        "research_tasks": None,
        "research_results": None,
        "draft_content": None,
        "editor_feedback": None,
        "final_content": None,
    }
    updated_fields_2 = planner_node(initial_state_2)
    final_state_2 = initial_state_2.copy()
    for key, value in updated_fields_2.items():
        final_state_2[key] = value

    print("\n--- State Update After Planner Node Execution (New Topic) ---")
    print(f"Plan Outline: \n{final_state_2.get('plan_outline')}")
    print(f"Target Audience: {final_state_2.get('target_audience')}")
    print(f"Research Tasks: {final_state_2.get('research_tasks')}")

Code Breakdown:

  1. Defining AgencyState: We first defined the AgencyState as a TypedDict. This is the "shared memory" of the entire agency, and all Agents read and write information through it. Clearly defining the state structure is the cornerstone of building robust multi-agent systems.
  2. The planner_node Function:
    • Input: It receives a state: AgencyState parameter. This is the current global state passed in by LangGraph.
    • Business Logic Encapsulation:
      • We extract the topic from the state.
      • We build a ChatPromptTemplate, which is the core of interacting with the LLM. We carefully designed the Prompt, asking the LLM to act as a "content strategist" and return results in a structured JSON format containing the outline, audience, and research tasks.
      • Using llm.with_structured_output is a very powerful technique. It guides the LLM to directly output data that conforms to our defined JSON Schema, greatly simplifying subsequent parsing work. It's like putting a pair of "structured output" glasses on the LLM, allowing it to see more clearly and articulate perfectly.
      • We call planner_chain.invoke() to execute the LLM reasoning.
    • Output: The function returns a dictionary. The key-value pairs in this dictionary will be used to update the AgencyState. Note that we only return the fields that need to be added or modified, not the entire AgencyState. LangGraph automatically merges these updates into the current global state.
    • Error Handling: Added a try-except block to handle potential errors during the LLM call, providing better robustness.
  3. Simulated Execution (if __name__ == "__main__":):
    • This part of the code demonstrates how to test our planner_node independently without the complete LangGraph framework.
    • We created an initial_state and then called planner_node.
    • Finally, we manually simulated the process of LangGraph merging the node's returned updates into the initial_state, printing the final state to verify the node's functionality.

Through this Planner Node, we have successfully encapsulated a complex "content planning" task into an executable, reusable, and testable LangGraph node. It receives input, leverages the power of large models for complex reasoning, and outputs results in a structured manner, updating our agency's shared project state. This perfectly embodies the Node as the true executor of the Graph!

Pitfalls & Best Practices

As a senior AI architect, I've seen far too many pitfalls. Now, I'll share some of the most common and easily triggered "traps" with you, so you can avoid taking unnecessary detours.

  1. Messy State Management: Misunderstanding Node Return Values

    • The Pitfall: A common mistake for beginners is returning a complete AgencyState object directly from a Node, or returning a dictionary that doesn't match the AgencyState structure.
    • The Consequence: This leads to incorrect state updates, ranging from data loss to the complete collapse of the Graph logic. LangGraph expects you to return a dictionary where the key-value pairs are meant to be added to or overwrite the corresponding fields in the current State.
    • How to Avoid:
      • Only return the delta: Your Node function should only return a dictionary containing the State fields you want to update or add. For example, planner_node only returns {"plan_outline": "...", "target_audience": "...", "research_tasks": [...]}.
      • Type Matching: Ensure the types of the fields in the returned dictionary match those defined in AgencyState. Strict type definitions using TypedDict or Pydantic Models can help you avoid many issues.
      • Understand Merge Logic: LangGraph's default StateGraph performs a shallow merge on dictionaries. This means if you return {"research_results": {"new_key": "value"}}, it will directly replace the entire old research_results dictionary rather than merging the internal keys. If you need deep merging, you must customize the reducer method in your StateGraph.

  2. Tight Coupling Between LLM Reasoning and Business Logic

    • The Pitfall: Stuffing all LLM calls, Prompt engineering, output parsing, and data processing logic into one massive Node function.
    • The Consequence: The code becomes hard to read, maintain, and test. If the Prompt needs tweaking or the LLM model needs switching, the entire Node might need a rewrite.
    • How to Avoid:
      • Single Responsibility Principle (SRP): A Node should only be responsible for one core business function. For example, the Planner Node is only responsible for planning, not researching or writing.
      • Modular Encapsulation: Encapsulate LLM-related logic (Prompt definitions, Chain building, parsers) into independent functions or classes. Just like our code planner_prompt | llm.with_structured_output(...), chained calls make the logic clearer.
      • Externalize Prompts: Consider storing complex Prompts in external files or configurations for easier management and iteration.

  3. Lack of Robustness: Inadequate Error Handling

    • The Pitfall: Failing to handle exceptions like LLM call failures, API timeouts, or data parsing errors.
    • The Consequence: In a production environment, if a network issue occurs or the LLM returns an unexpected format, the entire workflow will halt, resulting in a terrible user experience.
    • How to Avoid:
      • try-except is your friend: Always wrap LLM calls and other potentially failing operations in try-except blocks, just as we did in planner_node.
      • Graceful Degradation or Error States: When an error occurs, don't just let the program crash. You can return a special State update (e.g., {"error_message": "..."}) so that subsequent Nodes or the Graph's conditional_edge can evaluate and handle these errors—for instance, by routing to an error-handling flow or retrying.
      • Logging: Detailed logs help you quickly pinpoint issues. Print logs at the start and end of every Node, as well as at critical steps.

  4. Environment Configuration and Dependency Management

    • The Pitfall: Hardcoding API keys or failing to load environment variables correctly.
    • The Consequence: Poor code security, difficult deployment, and inconsistent behavior across different environments.
    • How to Avoid:
      • Use Environment Variables: All sensitive information (like API keys) should be managed via environment variables. We used python-dotenv to load the .env file, which is a great practice.
      • requirements.txt: Always maintain a clear requirements.txt file to ensure all dependencies can be installed correctly.

Remember, building a multi-agent system is like constructing a skyscraper; every Node is a brick. Only when every brick is solid and every connection is secure can the building stand tall.

📝 Summary of This Session

Congratulations, future AI architects! In this session, we not only deepened our understanding of the core concept of a LangGraph Node—how it acts as the true executor of the Graph by receiving state, executing business logic, and returning state updates—but more importantly, we built the first crucial department for our "Omnipotent AI Content Creation Agency" with our own hands: the Planner Node.

We learned how to:

  • Define a clear shared state, AgencyState, which is the foundation for collaboration among all our Agents.
  • Write a fully functional Node function that encapsulates LLM reasoning logic, using a carefully designed Prompt to make it play the role of a "content strategist."
  • Leverage llm.with_structured_output to ensure structured and reliable LLM outputs.
  • Understand the mechanism of Nodes returning state updates, rather than directly mutating the global state.

Through the hands-on practice with the Planner Node, you have taken the first step toward building complex multi-agent workflows. It translates high-level user requirements into concrete, actionable creation plans, paving the way for subsequent Agents like the Researcher and Writer.

Of course, we also explored common "pitfalls" when building Nodes and how to elegantly avoid them. These are invaluable experiences that will enable you to take charge of large-scale projects in the future.

In the next session, we will dive even deeper. We'll officially integrate our Planner Node into LangGraph's StateGraph and start defining the Edges of the graph, allowing our Planner to seamlessly pass its generated work plans to the next agent. Stay tuned!