LangGraph: कोड वेरिफिकेशन के लिए टूल कॉलिंग और बाउंडेड एक्ज़ीक्यूशन वाला एजेंट कैसे बनाएं

ऑटोनॉमस कोड मॉडिफिकेशन के लिए एक सामान्य लीनियर प्रॉम्प्ट चेन से कहीं अधिक की आवश्यकता होती है। जब कोई एजेंट किसी रिपॉजिटरी में बदलाव करता है, तो उसे फाइलों को पढ़ना, कोड एडिट्स लागू करना, टेस्ट सुइट्स चलाना, आउटपुट की जांच करना और रिग्रेशन दिखने पर खुद को सुधारना होता है। हालांकि, साइक्लिकल एजेंट पैटर्न्स—जो अक्सर ReAct (Reasoning and Acting) प्रतिमान पर आधारित होते हैं—अगर बिना किसी सीमा के छोड़ दिए जाएं तो परिचालन संबंधी जोखिम पैदा करते हैं। टूटी हुई डिपेंडेंसीज़ या लगातार फेल हो रहे टेस्ट्स का सामना करने वाला एजेंट महंगे और अनबाउंडेड एक्ज़ीक्यूशन लूप्स में फंस सकता है।

Google AI के LangGraph के लिए प्रलेखित ReAct आर्किटेक्चर का उपयोग करते हुए, यह गाइड कोर स्टेट-ग्राफ मैकेनिक्स को एक विशेष, बाउंडेड कोडिंग एजेंट के रूप में अनुकूलित करती है। यह कार्यान्वयन कोड एडिट्स को निष्पादित करता है, ऑटोमेटेड टेस्ट सुइट्स के माध्यम से बदलावों को सत्यापित करता है, और सख्त इटरेशन बजट लागू करता है।

---

कोर आर्किटेक्चर: स्टेट, नोड्स और कंडीशनल एजेस

LangGraph एजेंट वर्कफ़्लो को डायरेक्टेड ग्राफ्स के रूप में मॉडल करता है। स्टैटिक डायरेक्टेड एसाइक्लिक ग्राफ्स (DAGs) के विपरीत, LangGraph स्वाभाविक रूप से साइकिल्स को सपोर्ट करता है, जो पुनरावृत्तीय (iterative) एजेंट रीज़निंग के लिए आवश्यक हैं।

          [Entrypoint]
               │
               ▼
        ┌─────────────┐
        │  call_model │◄────────────┐
        └──────┬──────┘             │
               │                    │
     [should_continue]              │
        /             \             │
   (has tools)     (done / max)     │
      /                 \           │
     ▼                   ▼          │
┌───────────┐         ┌─────┐       │
│ call_tool │         │ END │       │
└─────┬─────┘         └─────┘       │
      │                             │
      └─────────────────────────────┘

यह ग्राफ तीन आर्किटेक्चरल तत्वों पर निर्भर करता है:

1. स्टेट: एक सेंट्रलाइज्ड टाइप्ड डेटा स्ट्रक्चर जो मैसेज हिस्ट्री, रन मेटाडेटा और स्पष्ट स्टेप काउंटर्स को स्टोर करता है।
2. नोड्स: सामान्य Python फंक्शन्स जो वर्तमान स्टेट प्राप्त करते हैं, आइसोलेटेड कंप्यूटेशन चलाते हैं (मॉडल को कॉल करना या टूल निष्पादित करना), और अपडेटेड स्टेट एट्रिब्यूट्स लौटाते हैं।
3. कंडीशनल एजेस: निर्णय लेने वाले फंक्शन्स जो स्टेट का निरीक्षण करते हैं (जैसे टूल कॉल्स की उपस्थिति या इटरेशन लिमिट का पार होना) ताकि यह तय किया जा सके कि टूल रनर की ओर रूट करना है या END पर समाप्त करना है।

---

पूर्वापेक्षाएँ और डिपेंडेंसीज़

सुनिश्चित करें कि आपके डेवलपमेंट एनवायरनमेंट में आवश्यक ग्राफ ऑर्केस्ट्रेशन और मॉडल ड्राइवर पैकेज मौजूद हैं:

pip install langgraph langchain-core langchain-openai

OpenAI-कंपैटिबल गेटवे के माध्यम से एजेंट को इन्फरेंस प्रोवाइडर्स से कनेक्ट करने के लिए, अपने एनवायरनमेंट वैरिएबल्स कॉन्फ़िगर करें। OneMux का उपयोग कई मॉडल्स में एकीकृत की-मैनेजमेंट और खर्च नियंत्रण को आसान बनाता है:

export ONEMUX_API_KEY="your_onemux_api_key"
export ONEMUX_BASE_URL="https://api.onemux.net/v1"

(नोट: हमेशा यह सत्यापित करें कि आपका चुना गया टारगेट मॉडल OpenAI-कंपैटिबल एंडपॉइंट्स के माध्यम से स्ट्रक्चर्ड टूल-कॉलिंग पैरामीटर्स का समर्थन करता है)।

---

स्टेप 1: बाउंडेड एजेंट स्टेट को परिभाषित करना

मानक संवादात्मक वर्कफ़्लो में, MessagesState चैट हिस्ट्री को जोड़ने का काम संभालता है। कोडिंग कार्यों के लिए, हम Python के TypedDict का उपयोग करके AgentState को स्पष्ट रूप से परिभाषित करते हैं ताकि मैसेज सीक्वेंस के साथ-साथ एक स्पष्ट काउंटर (step_count) और अधिकतम सीमा (max_steps) को ट्रैक किया जा सके।

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    step_count: int
    max_steps: int

add_messages रिड्यूसर यह सुनिश्चित करता है कि आने वाली मैसेज सूचियाँ पिछले ऑब्जर्वेशन्स को ओवरराइट करने के बजाय बातचीत के इतिहास में ठीक से जुड़ती जाएं।

---

स्टेप 2: फाइल सिस्टम और वेरिफिकेशन टूल्स को लागू करना

एजेंट को सोर्स फाइलों का निरीक्षण करने, अपडेट लागू करने और रिपॉजिटरी टेस्ट चलाने के लिए टूल्स की आवश्यकता होती है। हम इन ऑपरेशन्स को LangChain के @tool डेकोरेटर का उपयोग करके समाहित करते हैं।

import subprocess
from langchain_core.tools import tool

@tool
def read_file(filepath: str) -> str:
    """Reads contents of a file from the repository."""
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            return f.read()
    except Exception as e:
        return f"Error reading {filepath}: {str(e)}"

@tool
def write_file(filepath: str, content: str) -> str:
    """Overwrites a file with modified source code."""
    try:
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(content)
        return f"Successfully updated {filepath}"
    except Exception as e:
        return f"Error writing {filepath}: {str(e)}"

@tool
def run_test_suite(test_target: str) -> str:
    """Runs pytest on a specific file or test target."""
    try:
        result = subprocess.run(
            ["pytest", test_target],
            capture_output=True,
            text=True,
            timeout=30
        )
        return f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}\nExit Code: {result.returncode}"
    except subprocess.TimeoutExpired:
        return "Error: Test execution timed out after 30 seconds."
    except Exception as e:
        return f"Error executing tests: {str(e)}"

tools = [read_file, write_file, run_test_suite]
tools_by_name = {t.name: t for t in tools}

---

स्टेप 3: यूनिफाइड API के माध्यम से मॉडल सेटअप

हम यूनिफाइड प्रोवाइडर प्रॉक्सी एंडपॉइंट की ओर इशारा करते हुए ChatOpenAI का उपयोग करके मॉडल को इनिशियलाइज़ करते हैं। फंक्शन कॉलिंग के लिए पैरामीटर कम्पैटिबिलिटी सुनिश्चित करने के लिए OneMux Docs और Quickstart Guide देखें।

import os
from langchain_openai import ChatOpenAI

api_key = os.getenv("ONEMUX_API_KEY")
base_url = os.getenv("ONEMUX_BASE_URL", "https://api.onemux.net/v1")

llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.0,
    api_key=api_key,
    base_url=base_url
)

# Bind tools to the model instance
model = llm.bind_tools(tools)

---

स्टेप 4: ग्राफ नोड्स और लूप गार्ड्स का निर्माण

नोड्स लॉजिक निष्पादित करते हैं और बाउंड्स को बढ़ाते हैं। यदि कोई एजेंट टूटे हुए टेस्ट को ठीक किए बिना बार-बार लूप करता है, तो एज लॉजिक यह पता लगाता है कि step_count >= max_steps हो गया है और सीधे ग्राफ कंप्लीशन की ओर रूट कर देता है।

from langchain_core.messages import ToolMessage, AIMessage
from langchain_core.runnables import RunnableConfig

def call_model(state: AgentState, config: RunnableConfig):
    """Calls the model and increments the iteration step count."""
    response = model.invoke(state["messages"], config)
    current_steps = state.get("step_count", 0) + 1
    return {
        "messages": [response],
        "step_count": current_steps
    }

def call_tool(state: AgentState):
    """Executes tool calls generated by the most recent model response."""
    last_message = state["messages"][-1]
    outputs = []
    
    for tool_call in getattr(last_message, "tool_calls", []):
        tool_inst = tools_by_name.get(tool_call["name"])
        if tool_inst:
            result = tool_inst.invoke(tool_call["args"])
        else:
            result = f"Error: Tool {tool_call['name']} not found."
            
        outputs.append(
            ToolMessage(
                content=str(result),
                name=tool_call["name"],
                tool_call_id=tool_call["id"]
            )
        )
    return {"messages": outputs}

def should_continue(state: AgentState):
    """Evaluates tool requests and halts execution if boundaries are crossed."""
    messages = state["messages"]
    last_message = messages[-1]
    step_count = state.get("step_count", 0)
    max_steps = state.get("max_steps", 5)

    # Boundary check: Halt if loop budget exceeded
    if step_count >= max_steps:
        return "end"

    # Continue if the model requested another tool call
    if getattr(last_message, "tool_calls", None):
        return "continue"
        
    return "end"

---

स्टेप 5: ग्राफ संकलन (Compilation) और असेंबली

StateGraph का उपयोग करके घटकों को इकट्ठा करें और वर्कफ़्लो को संकलित (compile) करें:

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)

# Register Nodes
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tool)

# Set Entrypoint
workflow.set_entry_point("agent")

# Define Conditional and Static Edges
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "tools",
        "end": END
    }
)
workflow.add_edge("tools", "agent")

# Compile Graph
app = workflow.compile()

---

स्टेप 6: टेस्ट वेरिफिकेशन के साथ कोड बदलावों को निष्पादित करना

वर्कफ़्लो चलाने के लिए, सख्त निष्पादन सीमाओं के साथ एक कोडिंग कार्य परिभाषित करें:

initial_input = {
    "messages": [(
        "user",
        "Fix the bug in utils/math_lib.py where divide_numbers fails on zero values. "
        "Verify your fix by running the test suite on tests/test_math_lib.py. "
        "Only stop once all tests pass."
    )],
    "step_count": 0,
    "max_steps": 6
}

for event in app.stream(initial_input, stream_mode="values"):
    latest = event["messages"][-1]
    if isinstance(latest, AIMessage) and latest.tool_calls:
        print(f"[Action] Invoking tools: {[t['name'] for t in latest.tool_calls]}")
    elif isinstance(latest, ToolMessage):
        print(f"[Observation] {latest.name} -> {latest.content[:100]}...")
    else:
        print(f"[Agent] {latest.content}")

[Action] Invoking tools: ['read_file']
[Observation] read_file -> def divide_numbers(a, b):
    return a / b...
[Action] Invoking tools: ['write_file']
[Observation] write_file -> Successfully updated utils/math_lib.py...
[Action] Invoking tools: ['run_test_suite']
[Observation] run_test_suite -> STDOUT:
============================= test session starts ==============================
... 1 passed in 0.02s...
[Agent] The divide_numbers function has been updated to handle zero division safely, and all tests in tests/test_math_lib.py pass.

---

सीमाएँ और प्रोडक्शन हार्डनिंग (Production Hardening)

प्रोडक्शन एनवायरनमेंट्स में कोड-मॉडिफाइंग एजेंट्स को डिप्लॉय करते समय, निम्नलिखित सुरक्षा उपायों पर विचार करें:

• सैंडबॉक्सिंग एक्ज़ीक्यूशन: स्थानीय रूप से pytest जैसे मनमाने शेल कमांड चलाने से एनवायरनमेंट करप्ट होने या असुरक्षित स्क्रिप्ट निष्पादन का जोखिम रहता है। टेस्ट एक्ज़ीक्यूशन को अल्पकालिक (ephemeral) Docker कंटेनरों या सुरक्षित सैंडबॉक्स में अलग रखें।
• स्टेट टोकन सीमाएं: बड़े टूल आउटपुट्स—जैसे कई पेजों के स्टैक ट्रेस—कॉन्टेक्स्ट विंडो को तेज़ी से समाप्त कर सकते हैं। टेस्ट आउटपुट मैसेजेस को स्टेट हिस्ट्री में लिखने से पहले छोटा (truncate) करें।
• मॉडल लागत प्रबंधन: पुनरावृत्तीय ReAct लूप्स टोकन खपत को कई गुना बढ़ा देते हैं। OneMux Pricing डैशबोर्ड के माध्यम से उपयोग को ट्रैक करें, और प्रति-कार्य लागत को अनुकूलित करने के लिए OneMux Models पर कम लागत वाले रीज़निंग मॉडल्स का मूल्यांकन करें।

---

स्रोत

• Google AI for Developers: ReAct agent from scratch with Gemini and LangGraph (अपडेटेड 2 सितंबर, 2026)।