LangGraph Affiliate Agent - Build a Stateful Earning Agent With LangChain

Consider the difference:

Stateless agent: Asked to write about electricity savings every Monday, it produces a similar article each time. Users who see repeated content disengage. The affiliate link clicks decline over time.

Stateful agent with LangGraph: It maintains a list of covered topics, tracks which topics generated affiliate conversions, prioritises under-covered areas, and builds a content library that grows in value over time without repetition.

The second model is a real content operation. The first is just a scheduled script.

pip install langgraph langchain-openai langchain requests python-dotenv

The state is a typed dictionary that flows through every node in the graph:

from typing import TypedDict, List, Optional, Annotated
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage
import operator

class AffiliateAgentState(TypedDict):
    # Core content pipeline state
    topic: str
    research_notes: str
    draft_content: str
    final_content: str

    # Affiliate tracking
    affiliate_claim_url: Optional[str]
    api_key: Optional[str]

    # Memory across runs - persisted to external store
    published_topics: Annotated[List[str], operator.add]
    successful_topics: Annotated[List[str], operator.add]

    # Conversation context
    messages: Annotated[List[BaseMessage], operator.add]
    error: Optional[str]

The Annotated fields with operator.add accumulate across runs - published topics and successful topics grow as the agent operates over time.

import os
import requests
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(model="gpt-4o", temperature=0.4)

def select_topic_node(state: AffiliateAgentState) -> AffiliateAgentState:
    """Select a topic that hasn't been covered yet."""
    published = set(state.get("published_topics", []))
    topic_candidates = [
        "reducing grocery costs for Australian families",
        "cutting your energy bill in Australia 2026",
        "Australian subscription services worth keeping",
        "fuel savings strategies for Australians",
        "insurance comparison tips for Australians",
        "loyalty programme stacking in Australia",
        "bulk buying strategies for Australian households",
        "reducing water bills in Australian states",
    ]
    available = [t for t in topic_candidates if t not in published]
    if not available:
        available = topic_candidates  # Cycle through if all covered

    selected = available[0]
    return {**state, "topic": selected}


def research_node(state: AffiliateAgentState) -> AffiliateAgentState:
    """Research the selected topic."""
    response = llm.invoke([
        SystemMessage(content=(
            "You are an Australian savings researcher. "
            "Produce 3–5 bullet points of genuinely useful, specific information about the given topic. "
            "Do not fabricate statistics. Use typical ranges and 'generally' language."
        )),
        HumanMessage(content=f"Research topic: {state['topic']}"),
    ])
    return {**state, "research_notes": response.content}


def write_node(state: AffiliateAgentState) -> AffiliateAgentState:
    """Write an article based on research notes."""
    response = llm.invoke([
        SystemMessage(content=(
            "You are an Australian personal finance writer. "
            "Write a 400–500 word article in Australian English using the provided research notes. "
            "Be direct and practical. No jargon. No emojis. "
            "End with the placeholder text: [ROOPON_LINK_HERE]"
        )),
        HumanMessage(content=(
            f"Topic: {state['topic']}\n\n"
            f"Research:\n{state['research_notes']}"
        )),
    ])
    return {**state, "draft_content": response.content}


def register_affiliate_node(state: AffiliateAgentState) -> AffiliateAgentState:
    """Call the ROOPON API to get an affiliate claim URL."""
    try:
        response = requests.post(
            "https://www.roopon.com/api/v1/agents/register",
            json={
                "agent_name": "langgraph-affiliate-agent",
                "topic": state["topic"],
                "secret": os.getenv("ROOPON_AGENT_SECRET"),
            },
            headers={"Content-Type": "application/json"},
            timeout=15,
        )
        response.raise_for_status()
        data = response.json()
        return {
            **state,
            "affiliate_claim_url": data["claim_url"],
            "api_key": data["api_key"],
            "error": None,
        }
    except requests.RequestException as e:
        return {**state, "error": str(e)}


def publish_node(state: AffiliateAgentState) -> AffiliateAgentState:
    """Embed the affiliate link and finalise the content."""
    if state.get("error"):
        return {**state, "final_content": state["draft_content"]}

    claim_url = state["affiliate_claim_url"]
    affiliate_section = (
        f"\n\nIf you want to take action on these tips, ROOPON members across Australia "
        f"get access to exclusive partner discount codes and are entered into the weekly "
        f"member prize draws (see /giveaways for current prizes and terms). Membership is $19.99 per month. "
        f"[Learn more and join ROOPON here]({claim_url})."
    )

    final = state["draft_content"].replace("[ROOPON_LINK_HERE]", affiliate_section)

    return {
        **state,
        "final_content": final,
        "published_topics": [state["topic"]],  # Accumulated via operator.add
    }


def should_retry(state: AffiliateAgentState) -> str:
    """Conditional edge: retry if affiliate registration failed."""
    if state.get("error"):
        return "write"  # Try writing again rather than failing completely
    return "publish"

def build_affiliate_graph() -> StateGraph:
    graph = StateGraph(AffiliateAgentState)

    graph.add_node("select_topic", select_topic_node)
    graph.add_node("research", research_node)
    graph.add_node("write", write_node)
    graph.add_node("register_affiliate", register_affiliate_node)
    graph.add_node("publish", publish_node)

    graph.set_entry_point("select_topic")
    graph.add_edge("select_topic", "research")
    graph.add_edge("research", "write")
    graph.add_edge("write", "register_affiliate")
    graph.add_conditional_edges(
        "register_affiliate",
        should_retry,
        {"write": "write", "publish": "publish"},
    )
    graph.add_edge("publish", END)

    return graph.compile()


# Run the agent
if __name__ == "__main__":
    app = build_affiliate_graph()
    initial_state: AffiliateAgentState = {
        "topic": "",
        "research_notes": "",
        "draft_content": "",
        "final_content": "",
        "affiliate_claim_url": None,
        "api_key": None,
        "published_topics": [],
        "successful_topics": [],
        "messages": [],
        "error": None,
    }

    result = app.invoke(initial_state)
    print("Published content:\n")
    print(result["final_content"])
    print("\nCovered topics so far:", result["published_topics"])

LangGraph supports checkpointing via SqliteSaver or PostgresSaver. Adding a checkpointer means the published_topics list survives between runs, giving you genuine long-term topic tracking:

from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("affiliate_agent.db") as checkpointer:
    app = build_affiliate_graph().compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "main-affiliate-agent"}}
    result = app.invoke(initial_state, config=config)

With persistence in place, run the agent on a weekly cron job. It will systematically build out a library of Australian savings content, each piece carrying a unique ROOPON affiliate link.

Start integrating ROOPON with your LangGraph agents at https://www.roopon.com/ai-agents. API documentation is at https://www.roopon.com/docs/agents.