Iterative Workflows in LangGraph | Agentic AI

ল্যাংগ্রাফ (LangGraph)-এ এজেন্টিক ওয়ার্কফ্লো তৈরির একটি অন্যতম শক্তিশালী প্যাটার্ন হলো ইটারেটিভ বা লুপিং ওয়ার্কফ্লো (Iterative / Looping Workflow)। এই নোটটিতে আলোচনা করা হয়েছে কীভাবে জেনারেটর, ইভালুয়েটর এবং অপ্টিমাইজার—এই তিনটি মূল উপাদানের সমন্বয়ে একটি স্বয়ংক্রিয় লুপ তৈরি করে এলএলএম (LLM)-এর আউটপুট ক্রমান্বয়ে উন্নত করা যায়।

1. Introduction & Workflow Evolution

এজেন্টিক এআই (Agentic AI) সিস্টেমে জটিল সমস্যা সমাধানের জন্য বিভিন্ন ধরনের ওয়ার্কফ্লো ডিজাইন করা হয়। ল্যাংগ্রাফে আমরা মূলত চার ধরনের ওয়ার্কফ্লো দেখতে পাই:

2. Real-World Motivation & Problem Statement

বাস্তব জীবনের সমস্যা (Real-World Use Case)

সামাজিক যোগাযোগ মাধ্যমে (যেমন: X/Twitter, LinkedIn, Instagram) নিয়মিত মানসম্মত কন্টেন্ট পোস্ট করা সময়সাপেক্ষ। একজন কনটেন্ট ক্রিয়েটর কাজের চাপে অন্যান্য প্ল্যাটফর্মে নিয়মিত সক্রিয় থাকতে পারেন না।

সাধারণ এলএলএম অটোমেশনের সীমাবদ্ধতা

একটি সাধারণ LLM-কে কোনো টপিক দিয়ে পোস্ট লিখতে বললে প্রথমবারেই সন্তোষজনক বা মানসম্মত (High Quality) কন্টেন্ট পাওয়া যায় না। আউটপুট অনেক সময় অতি-সাধারণ, পুনরাবৃত্তিমূলক বা বোরিং হতে পারে।

সমাধান

একটি ইটারেটিভ ওয়ার্কফ্লো তৈরি করা, যা একা একা কোনো পোস্ট তৈরি করবে না; বরং একটি কঠোর মূল্যায়নকারী (Evaluator) নোড পোস্টটির মান যাচাই করবে এবং গুণমান সন্তোষজনক না হওয়া পর্যন্ত অপ্টিমাইজার নোড ফিডব্যাকের ওপর ভিত্তি করে পোস্টটি উন্নত করতে থাকবে।

3. Core Mental Model: Iterative Refinement Loop

এই ইটারেটিভ ওয়ার্কফ্লোটি ৩টি প্রধান উপাদানের (Components) সমন্বয়ে গঠিত:

1. Generator LLM: ব্যবহারকারীর প্রদত্ত বিষয়বস্তু (Topic)-র ওপর প্রথম খসড়া (Initial Post/Tweet) তৈরি করে।
2. Evaluator LLM: নির্দিষ্ট নিয়মাবলি ও মূল্যায়ন ক্রাইটেরিয়া (Evaluation Criteria)-র ওপর ভিত্তি করে পোস্টটি পরীক্ষা করে। এটি দুটি আউটপুট দেয়:
   • Status: Approved অথবা Needs Improvement
   • Feedback: কন্টেন্টের সবল ও দুর্বল দিক নির্দেশক টিপ্পনি।
3. Optimizer LLM: পূর্ববর্তী পোস্ট এবং ইভালুয়েটরের দেওয়া ফিডব্যাক গ্রহণ করে পোস্টটিকে পুনরায় লিখে উন্নত করে।

4. Designing the State Schema (TweetState)

ল্যাংগ্রাফের প্রতিটি স্টেট ট্রানজিশনে ডেটা বজায় রাখতে TypedDict ব্যবহার করে স্টেট সংজ্ঞায়িত করা হয়। ক্রমান্বয়ে ইতিহাস (History) বজায় রাখার জন্য operator.add রিডিউসার ব্যবহার করা হয়েছে।

import operator
from typing import TypedDict, Literal, Annotated

class TweetState(TypedDict):
    topic: str
    tweet: str
    evaluation: Literal['Approved', 'Needs Improvement']
    feedback: str
    iteration: int
    max_iterations: int
    # History tracking using reducer functions
    tweet_history: Annotated[list[str], operator.add]
    feedback_history: Annotated[list[str], operator.add]

Note: Annotated[list[str], operator.add] নির্দেশ করে যে প্রতিবার যখন নতুন টুইট বা ফিডব্যাক স্টেটে পাঠানো হবে, ল্যাংগ্রাফ আগের স্টেট মুছে না ফেলে পুরনো তালিকার শেষে নতুন উপাদানটি Append (Merge) করবে।

5. Implementing Node 1: Generator LLM

জেনারেটর নোডের কাজ হলো স্টেটের topic গ্রহণ করা এবং সিস্টেম প্রম্পট অনুসরণ করে একটি মজার ও সংক্ষিপ্ত টুইট তৈরি করা।

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage

# Initialize LLMs
generator_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

def generate_tweet(state: TweetState) -> dict:
    messages = [
        SystemMessage(content="You are a witty and funny Twitter influencer."),
        HumanMessage(content=(
            f"Write a short, original, and funny tweet about '{state['topic']}'.\n"
            "Rules:\n"
            "- Do NOT use Q&A format.\n"
            "- Must be strictly under 280 characters.\n"
            "- Use observational humor, sarcasm, or relatable logic."
        ))
    ]
    
    response = generator_llm.invoke(messages)
    
    return {
        "tweet": response.content,
        "tweet_history": [response.content]  # Adding to history list via reducer
    }

6. Implementing Node 2: Evaluator LLM with Structured Output

ইভালুয়েটর নোড নিশ্চিত করে যে রেসপন্সটি যাতে এলোমেলো টেক্সট না হয়ে একটি নির্দিষ্ট ফর্ম্যাটে (Structured Format) আসে। এর জন্য Pydantic স্কিমা ব্যবহার করা হয়।

Pydantic Schema Definition

from pydantic import BaseModel, Field
from typing import Literal

class TweetEvaluationSchema(BaseModel):
    evaluation: Literal['Approved', 'Needs Improvement'] = Field(
        description="Select 'Approved' if the tweet is high quality, highly engaging and funny. Otherwise 'Needs Improvement'."
    )
    feedback: str = Field(
        description="Detailed constructive feedback highlighting strengths and weaknesses of the tweet."
    )

Evaluator Node Function

evaluator_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
structured_evaluator_llm = evaluator_llm.with_structured_output(TweetEvaluationSchema)

def evaluate_tweet(state: TweetState) -> dict:
    messages = [
        SystemMessage(content="You are a strict, humorless Twitter critic. You evaluate tweets strictly based on humor, originality, and virality."),
        HumanMessage(content=(
            f"Evaluate the following tweet:\n\n'{state['tweet']}'\n\n"
            "Criteria:\n"
            "- Reject if it uses cliché Q&A format.\n"
            "- Reject if it exceeds 280 characters.\n"
            "- Reject if it lacks humor or viral potential."
        ))
    ]
    
    response = structured_evaluator_llm.invoke(messages)
    
    return {
        "evaluation": response.evaluation,
        "feedback": response.feedback,
        "feedback_history": [response.feedback]
    }

7. Implementing Node 3: Optimizer LLM

অপ্টিমাইজার নোডটি ইভালুয়েটরের দেওয়া ফিডব্যাক ব্যবহার করে পোস্টটি রিরাইট করে এবং iteration কাউন্টার 1 বাড়ায়।

optimizer_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

def optimize_tweet(state: TweetState) -> dict:
    messages = [
        SystemMessage(content="You are an expert Twitter growth optimizer."),
        HumanMessage(content=(
            f"Improve the following tweet based on the critic's feedback.\n\n"
            f"Topic: {state['topic']}\n"
            f"Original Tweet: {state['tweet']}\n"
            f"Feedback: {state['feedback']}\n\n"
            "Rewrite it into a punchy, viral-worthy tweet under 280 characters. Avoid Q&A format."
        ))
    ]
    
    response = optimizer_llm.invoke(messages)
    current_iteration = state.get("iteration", 1) + 1
    
    return {
        "tweet": response.content,
        "iteration": current_iteration,
        "tweet_history": [response.content]
    }

8. Defining Graph Architecture & Looping Mechanism

গ্রাফ সম্পূর্ণ করতে এখন নোড ও এজগুলো যুক্ত করতে হবে এবং অনন্ত লুপ (Infinite Loop) বন্ধ করার জন্য সঠিক রাউটিং লজিক লিখতে হবে।

Routing Function (Conditional Logic)

from langgraph.graph import END

def route_evaluation(state: TweetState) -> str:
    # Stop loop if tweet is approved OR max iterations limit reached
    if state["evaluation"] == "Approved" or state["iteration"] >= state["max_iterations"]:
        return "Approved"
    return "Needs Improvement"

Graph Building & Linking

from langgraph.graph import StateGraph, START

# 1. Initialize Graph
builder = StateGraph(TweetState)

# 2. Add Nodes
builder.add_node("generate", generate_tweet)
builder.add_node("evaluate", evaluate_tweet)
builder.add_node("optimize", optimize_tweet)

# 3. Add Edges
builder.add_edge(START, "generate")
builder.add_edge("generate", "evaluate")

# 4. Conditional Edge (Branching Decision)
builder.add_conditional_edges(
    "evaluate",
    route_evaluation,
    {
        "Approved": END,
        "Needs Improvement": "optimize"
    }
)

# 5. Loop Back Edge
builder.add_edge("optimize", "evaluate")

# Compile Workflow
app = builder.compile()

9. Workflow Execution & Debugging Realities

গ্রাফ এক্সিকিউশন (Graph Execution)

initial_state = {
    "topic": "Indian Railways",
    "iteration": 1,
    "max_iterations": 5
}

result = app.invoke(initial_state)

print(f"Final Status: {result['evaluation']}")
print(f"Total Iterations: {result['iteration']}")
print(f"Final Tweet: {result['tweet']}")

রিয়েল-ওয়ার্ল্ড পিটফল ও ডিবাগিং মানসিকতা (Debugging & Edge Cases)

1. অসীম লুপ সমস্যা (Infinite Loop Hazard): যদি ইভালুয়েটর নোডের ক্রাইটেরিয়া অত্যন্ত কঠোর হয় এবং অপ্টিমাইজার সেই অনুযায়ী উন্নতি করতে না পারে, তবে সিস্টেমটি অসীম লুপে আটকে যেতে পারে।

   • সমাধান: স্টেটে একটি max_iterations ভেরিয়েবল রাখা এবং iteration >= max_iterations হলে বাধ্যতামূলকভাবে লুপ বন্ধ করে END এ পাঠানো।

2. মডেল নির্বাচন ও সূক্ষ্ম পার্থক্য (Model Strength Trade-offs):

   • জেনারেটর/অপ্টিমাইজারের জন্য অধিক শক্তিশালী মডেল (যেমন: gpt-4o) এবং ইভালুয়েশনের জন্য ফাস্ট/কঠোর মডেল (যেমন: gpt-4o-mini) বেছে নেওয়া উচিত।
   • যদি শক্তিশালী মডেল (GPT-4o) প্রথমবারই নিখুঁত টুইট বানিয়ে ফেলে, তবে লুপ টেস্ট করার জন্য সাময়িকভাবে অপেক্ষাকৃত দুর্বল মডেল ব্যবহার করে ইভালুয়েটর রিজেকশন নিশ্চিত করা যায়।

3. হিস্ট্রি ট্র্যাকিং (State Reducer Issues): পূর্ববর্তী সব টুইট এবং ফিডব্যাক ট্র্যাকিং না করলে অপ্টিমাইজার বারবার একই ভুল করতে পারে। operator.add ব্যবহারের মাধ্যমে সম্পূর্ণ ইতিহাস বজায় রাখা যায়।

Summary and Key Takeaways

• ইটারেটিভ ওয়ার্কফ্লোর ভূমিকা: এটি এমন একটি মেকানিজম যেখানে এলএলএম-এর আউটপুট ফিডব্যাক লুপের মাধ্যমে বারবার ফিল্টার ও উন্নত করা হয়।
• মূল উপাদান ত্রয়ী: Generator (কন্টেন্ট তৈরি করে), Evaluator (মান যাচাই ও ক্রিটিক করে), এবং Optimizer (ফিডব্যাক অনুযায়ী সংশোধন করে)।
• Pydantic Structured Output: ইভালুয়েটরের সিদ্ধান্তকে নির্দিষ্ট কী-ভ্যালু জোড়ায় (evaluation, feedback) স্ট্রাকচার্ড রাখতে with_structured_output ব্যবহার করা জরুরি।
• লুপ তৈরির উপায়: ল্যাংগ্রাফে সাধারণ add_edge("optimize", "evaluate") ব্যবহার করেই লুপ তৈরি করা সম্ভব।
• কন্ডিশনাল রাউটিং: ইভালুয়েটরের উত্তরের ওপর ভিত্তি করে গ্রাফ সিদ্ধান্ত নেয় পরবর্তী পদক্ষেপ END হবে নাকি optimize হবে।
• সেফটি ভালভ (max_iterations): অসীম লুপের খরচের ঝুঁকি ও সিস্টেম হ্যাং হওয়া এড়াতে সর্বোচ্চ ইটারেশন লিমিট রাখা বাধ্যতামূলক।
• স্টেট রিডিউসার (operator.add): লুপের মধ্য দিয়ে যাওয়ার সময় পুরনো তথ্যের ইতিহাস বজায় রাখতে ল্যাংগ্রাফ স্টেটে রিডিউসার ফাংশন ব্যবহার করা হয়।