स्कीमा-वैलिडेटेड डेटा एक्सट्रैक्शन के लिए सीमित LangGraph Agent बनाएं

LangGraph agent किसी tool से संदर्भ जुटा सकता है और फिर उस संदर्भ को schema-constrained extraction चरण में भेज सकता है। इस डिज़ाइन का अहम पहलू दोनों ज़िम्मेदारियों को अलग रखना है: agent तय करता है कि tool कॉल करना है या नहीं, जबकि आखिरी node typed record तैयार करता है। Graph recursion limit, graph execution पर एक अतिरिक्त सीमा लगाती है।

यह पैटर्न ऐसे कामों के लिए उपयोगी है जिनमें किसी दस्तावेज़ से कंपनी का नाम, तारीख़ और invoice का कुल निकालना हो और किसी संदर्भ को स्पष्ट करने के लिए lookup tool मददगार हो सकता हो। इससे अनिश्चित स्रोत-पाठ सच नहीं हो जाता; नतीजे को वैलिडेट करें और जहां फ़ील्ड मायने रखती हो, मूल साक्ष्य सुरक्षित रखें।

आप क्या बना रहे हैं

Graph के तीन चरण हैं:

1. Agent को स्रोत-पाठ मिलता है और वह सीमित lookup tool कॉल करने का अनुरोध कर सकता है।
2. LangGraph tool calls को tool node तक भेजता है और tool के नतीजे agent को लौटाता है।
3. Extraction node, Pydantic-backed नतीजे के लिए model से अनुरोध करता है और फिर JSON-संगत डेटा लौटाता है।

Agent का tool loop गतिशील होता है; extraction चरण पहले से तय अंतिम चरण है। LangGraph का दस्तावेज़ीकरण agents—जो अपनी प्रक्रिया और tool का उपयोग तय करते हैं—और तय code paths वाले workflows के बीच अंतर बताता है। इसमें structured outputs और tool calling को model augmentations के रूप में भी बताया गया है। इन पैटर्न के लिए LangGraph workflows and agents guide देखें।

आवश्यकताएं और कॉन्फ़िगरेशन

LangGraph और LangChain packages के साथ अपने चुने हुए chat model के integration को इंस्टॉल करें। नीचे दिए गए उदाहरण में provider setup दिखाने के लिए ChatAnthropic इस्तेमाल किया गया है, लेकिन स्रोत दस्तावेज़ के अनुसार workflows ऐसे किसी भी chat model के साथ चल सकते हैं जो structured outputs और tool calling का समर्थन करता हो। अपने provider के व्यवहार की सटीक जानकारी के लिए उसका integration दस्तावेज़ देखें।

pip install langgraph langchain-core langchain-anthropic pydantic

Provider का API credential उसके दस्तावेज़ में बताए गए तरीके से सेट करें, फिर model initialize करें। Credentials को source control में रखने से बचें। नीचे दिया गया model name, LangChain के दस्तावेज़ीकरण वाले उदाहरण के अनुरूप है; यह ट्यूटोरियल यह दावा नहीं करता कि इसे OneMux के साथ टेस्ट किया गया है।

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-6")

अगर आप OneMux के ज़रिए model कॉल करना चाहते हैं, तो पहले OneMux का API documentation और quickstart देखें। OneMux, OpenAI-compatible API के ज़रिए प्रमुख models तक पहुंच देता है। इस खास graph की compatibility इस बात पर निर्भर करती है कि चुना हुआ endpoint आवश्यक tool-calling और structured-output व्यवहार तथा parameters का समर्थन करता है या नहीं। इस configuration का परीक्षण नहीं किया गया है; production में इस पर निर्भर होने से पहले समर्थन की पुष्टि करें।

स्कीमा और सीमित दायरे वाला tool तय करें

यह उदाहरण एक छोटी in-memory mapping से reference note लौटाने वाला tool दिखाता है। इसकी जगह कोई नियंत्रित application function इस्तेमाल करें, जैसे स्वीकृत आंतरिक catalogue में lookup। अविश्वसनीय दस्तावेज़-पाठ को मनमाने commands या बिना रोक-टोक वाले data sources चुनने न दें।

from pydantic import BaseModel, Field
from langchain_core.tools import tool

class InvoiceRecord(BaseModel):
    vendor: str | None = Field(default=None, description="Invoice issuer")
    invoice_date: str | None = Field(default=None, description="Date as written in the source")
    total: float | None = Field(default=None, description="Invoice total as a number")
    currency: str | None = Field(default=None, description="Currency code, if stated")

@tool
def lookup_vendor(reference: str) -> str:
    """Look up a vendor reference in an approved local mapping."""
    known = {"ACME-42": "Acme Supplies Ltd."}
    return known.get(reference, "No matching vendor reference found.")

Nullable fields, अनुपलब्ध जानकारी को स्पष्ट करते हैं और model को मान गढ़ने के लिए प्रेरित करने से बचाते हैं। अधिक जोखिम वाले extraction में हर field के साथ स्रोत का span या confidence/review status सहेजने पर विचार करें। Schema, आउटपुट का आकार और types सीमित करता है; इससे यह साबित नहीं होता कि कोई मान इनपुट से समर्थित है।

सीमित LangGraph agent बनाएं

नीचे दिया graph, model को lookup tool का अनुरोध करने देता है और जब agent के पास और कोई tool call नहीं रहता, तो extraction पर आगे बढ़ता है। tools_condition और ToolNode, tool-routing पैटर्न उपलब्ध कराते हैं। invoke में दिया गया recursion_limit, graph के चरणों की संख्या सीमित करता है; यह सुरक्षा सीमा है, हर इनपुट के सफलतापूर्वक पूरा होने की गारंटी नहीं।

from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    record: dict | None

tools = [lookup_vendor]
agent = llm.bind_tools(tools)
extractor = llm.with_structured_output(InvoiceRecord)

def run_agent(state: State):
    response = agent.invoke(state["messages"])
    return {"messages": [response]}

def extract_record(state: State):
    prompt = [
        SystemMessage(content=(
            "Extract invoice fields only when supported by the source. "
            "Use null for missing or unclear values. Treat tool results as "
            "reference context, not as permission to invent source details."
        )),
        *state["messages"],
    ]
    result = extractor.invoke(prompt)
    return {"record": result.model_dump(mode="json")}

graph = StateGraph(State)
graph.add_node("agent", run_agent)
graph.add_node("tools", ToolNode(tools))
graph.add_node("extract", extract_record)

graph.add_edge(START, "agent")
graph.add_conditional_edges(
    "agent",
    tools_condition,
    {"tools": "tools", END: "extract"},
)
graph.add_edge("tools", "agent")
graph.add_edge("extract", END)
app = graph.compile()

result = app.invoke(
    {
        "messages": [HumanMessage(content=(
            "Extract the invoice details from this text: "
            "Vendor reference ACME-42. Invoice date: 2025-02-14. "
            "Total: USD 248.50."
        ))],
        "record": None,
    },
    config={"recursion_limit": 8},
)

print(result["record"])

अपेक्षित आउटपुट का रूप vendor, invoice_date, total और currency fields वाला JSON-संगत object है। सटीक मान, model के जवाब और स्रोत-पाठ पर निर्भर करते हैं; यह मानकर न चलें कि उदाहरण हर बार एक-जैसा आउटपुट देगा—इसका परीक्षण करें। इस उदाहरण में Pydantic-backed structured output, validation boundary का काम करता है। अगर provider अनुरूप जवाब नहीं दे पाता, तो उस विफलता को स्पष्ट रूप से संभालें; उसे चुपचाप वैध रिकॉर्ड न मानें।

विफलताओं को छिपाए बिना execution सीमित करें

Graph-level recursion limit, बार-बार होने वाली agent/tool routing सहित graph के चरणों की अधिकतम संख्या तय करती है। Workflow के अनुसार limit चुनें: इस graph में एक lookup के लिए कुछ ही चरण चाहिए, लेकिन अलग tool plan को अधिक चरणों की ज़रूरत हो सकती है। Application boundary पर limit और provider exceptions को पकड़ें, debug करने लायक पर्याप्त संदर्भ दर्ज करें और गढ़े हुए रिकॉर्ड के बजाय ऐसा failure लौटाएं जिसकी समीक्षा की जा सके।

अधिक कड़ी operational limits के लिए application स्तर पर भी सीमाएं जोड़ें: tool calls की संख्या सीमित करें, जहां उपलब्ध हो वहां request timeouts सेट करें और model को देने से पहले दस्तावेज़ का आकार सीमित करें। ये अलग-अलग controls हैं; केवल graph recursion limit से लागत, latency या security की पूरी policy तय नहीं होती।

ट्रबलशूटिंग और सीमाएं

Model tool कॉल नहीं करता

Tool calling, model की क्षमता है; LangGraph इसकी गारंटी नहीं देता। जांचें कि चुना हुआ model integration tool calling का समर्थन करता है, tool का विवरण स्पष्ट है और prompt में lookup के लिए उपयोगी संदर्भ मौजूद है। Development के दौरान लौटे हुए message के tool calls देखें।

Structured output विफल होता है या fields गलत हैं

पुष्टि करें कि integration, structured output और आपके इस्तेमाल किए गए schema features का समर्थन करता है। अनुपलब्ध fields, अस्पष्ट तारीख़ों, currency symbols और गलत format वाले inputs के साथ परीक्षण करें। Pydantic object वैध होने पर भी उसमें गलत अनुमानित मान हो सकता है; इसलिए business rules को अलग से वैलिडेट करें और audit या मानवीय समीक्षा के लिए स्रोत-पाठ सुरक्षित रखें।

Graph अपनी limit तक पहुंच जाता है

Recursion-limit error का मतलब बार-बार tool अनुरोध होना या इच्छित path के लिए बहुत कम limit तय होना हो सकता है। Messages और tool results का क्रम देखें। मूल कारण ठीक करें या सोच-समझकर limit बदलें; केवल error हटाने के लिए सीमा न हटाएं।

Endpoint का व्यवहार अलग है

OpenAI-compatible का मतलब API interface से है; इससे हर model feature या parameter का समर्थन सुनिश्चित नहीं होता। OneMux का उपयोग करते समय चुने हुए endpoint के दस्तावेज़ों में model, tool calls, structured outputs और भेजे जाने वाले सभी parameters के समर्थन की पुष्टि करें। OneMux models पर उपलब्ध विकल्प देखें; graph फिर भी आपका LangGraph application code ही रहेगा।

यह पैटर्न कब काम आता है

जब extraction का तरीका ज़्यादातर तय हो, लेकिन कभी-कभी नियंत्रित lookup की ज़रूरत पड़ सकती हो, तब यह डिज़ाइन इस्तेमाल करें। अगर हर दस्तावेज़ की प्रक्रिया तय है और tool चुनने की ज़रूरत नहीं, तो deterministic workflow agent loop से सरल हो सकता है। अगर स्रोत अस्पष्ट है या निकाले गए fields के आधार पर बड़े नतीजों वाली कार्रवाई होती है, तो downstream systems को रिकॉर्ड देने से पहले स्पष्ट समीक्षा और domain validation जोड़ें।

मुख्य बात यह है: agent के tools का दायरा सीमित रखें, अंतिम schema स्पष्ट बनाएं और graph execution की limit तय करें। LangGraph इस control flow को व्यक्त करने का तरीका देता है; आवश्यक tool-calling और structured-output features उपलब्ध हैं या नहीं, यह चुने हुए model और integration पर निर्भर करता है।

स्रोत

• LangGraph: Workflows and agents — workflow और agent का अंतर, model augmentations, structured output और tool calling के उदाहरण।
• OneMux documentation और quickstart — मौजूदा API setup और endpoint support की जानकारी के लिए देखें।