API incidents की जाँच के लिए सीमित LangGraph Agent बनाएँ

निर्णय: diagnosis अपने-आप करें, remediation सोच-समझकर

API incident agent logs, health checks और हाल के deploy metadata से evidence जुटाकर समय बचा सकता है। उसे चुपचाप services restart नहीं करनी चाहिए, releases rollback नहीं करने चाहिए और configuration नहीं बदलनी चाहिए। अधिक सुरक्षित डिज़ाइन में केवल-पठन वाले diagnosis को side-effect करने वाली remediation से अलग रखा जाता है, agent के काम की सीमा तय होती है और विफल या अनुपलब्ध evidence operator को साफ़ दिखाई देता है।

यह एक व्यावहारिक डिज़ाइन है, परीक्षित integration या production के लिए तैयार service नहीं। नीचे दिए गए tool bodies उदाहरण के लिए हैं: इन्हें अपने authenticated और सीमित permissions वाले observability clients से बदलें।

आवश्यकताएँ और workflow

आपको Python, अपने application में configure किया गया LangGraph-संगत chat model और उन diagnostic systems के लिए client functions चाहिए जिनका आप पहले से इस्तेमाल करते हैं। इस उदाहरण में माना गया है कि ये client functions data लौटाते हैं या exceptions उठाते हैं। इसमें किसी खास model provider, endpoint या tool integration की धारणा नहीं है।

एक उपयोगी incident run इस सीमित चक्र का पालन करता है:

1. प्रभावित service और time window के लिए पहले से तय किए गए कुछ signals इकट्ठा करें।
2. हर परिणाम को successful, failed या unavailable के रूप में चिह्नित करें।
3. Model से evidence का सारांश देने और यह बताने को कहें कि क्या अब भी अज्ञात है।
4. Graph steps की तय सीमा पूरी होने पर रुकें; state बदलने वाली किसी भी कार्रवाई से पहले human review माँगें।

Status field का उद्देश्य operational है: log query fail होने पर उसे ऐसी सफल log search न दिखाएँ जिसमें कोई matching event नहीं मिला।

स्पष्ट failure signals वाले diagnostic tools तय करें

Tool contract को सीमित रखें। Service का नाम और सीमित time window अनिवार्य करें; जहाँ fixed query पर्याप्त हो, वहाँ model को मनमानी query करने की अनुमति न दें। Structured results लौटाएँ, ताकि बाद के nodes और incident report खाली evidence और tool failure में अंतर कर सकें।

from typing import Literal, TypedDict
from langchain.tools import tool

class Signal(TypedDict):
    status: Literal["ok", "error", "unavailable"]
    source: str
    data: str
    error: str | None


def query_logs(service: str, minutes: int) -> Signal:
    if not 1 <= minutes <= 60:
        return {
            "status": "error", "source": "logs", "data": "",
            "error": "minutes must be between 1 and 60",
        }
    try:
        # Replace with a read-only, scoped observability client call.
        rows = logs_client.search(service=service, minutes=minutes)
        return {
            "status": "ok", "source": "logs",
            "data": str(rows), "error": None,
        }
    except TimeoutError as exc:
        return {
            "status": "unavailable", "source": "logs", "data": "",
            "error": f"timeout: {exc}",
        }

@tool
def inspect_api(service: str, minutes: int = 15) -> dict:
    """Collect scoped, read-only API incident signals."""
    return query_logs(service, minutes)

इसी result shape का इस्तेमाल करके health checks या deployment metadata के लिए अलग-अलग tools जोड़ें। हर exception को पकड़कर खाली string न लौटाएँ: broken integration की जाँच के लिए पर्याप्त error context रखें, लेकिन logs और model context में secrets या संवेदनशील payloads न आने दें।

Execution सीमित करें और अधूरे runs स्पष्ट करें

create_agent का इस्तेमाल करके बनाए गए agent के लिए invocation के समय graph recursion limit तय करें। यह limit graph steps को सीमित करती है; यह उपयोगी उत्तर की गारंटी नहीं देती और न ही tool-level timeouts की जगह लेती है। आसपास के application में limit पूरी होने की स्थिति संभालें और run को पूर्ण diagnosis मानने के बजाय स्पष्ट “incomplete” status लौटाएँ।

from langchain.agents import create_agent

agent = create_agent(
    model=your_configured_chat_model,
    tools=[inspect_api],
    system_prompt=(
        "Diagnose API incidents using available evidence. "
        "Report each source as ok, error, or unavailable. "
        "Do not claim a cause without supporting evidence. "
        "Do not perform remediation."
    ),
)

try:
    result = agent.invoke(
        {"messages": [{"role": "user", "content":
            "Investigate elevated errors for payments-api over the last 15 minutes."}]},
        config={"recursion_limit": 12},
    )
except Exception as exc:
    # Map known graph-limit and provider/tool failures to your app's
    # incident-run status; do not expose credentials or raw secrets.
    result = {"status": "incomplete", "reason": type(exc).__name__}

यह limit value केवल उदाहरण है, हर स्थिति के लिए सुझाई गई एक ही value नहीं। इसकी value उन tools और model turns की संख्या के आधार पर चुनें जिन्हें आप अनुमति देते हैं; साथ ही tool और application layers पर timeouts और request-size limits जोड़ें। Run ID, चुनी गई service और time window, आज़माए गए tools, tool status और termination reason दर्ज करें। इन signals से operator समझ सकता है कि “कोई error नहीं मिला” या “log backend का समय समाप्त हो गया”।

Remediation से पहले human review जोड़ें

अगर बाद में ऐसे tools जोड़ें जो production state बदलते हैं, तो उन tools को LangChain के human-in-the-loop middleware और checkpointer से नियंत्रित करें। Documented middleware approval, editing, rejection या direct response के लिए चुनी गई tool calls को रोक सकता है। इसका state सहेजा जाता है, ताकि निर्णय के बाद execution फिर शुरू हो सके; production में इस्तेमाल के लिए in-memory विकल्प के बजाय persistent checkpointer आवश्यक है, जिसे यहाँ prototyping के लिए दिखाया गया है। LangChain human-in-the-loop documentation देखें।

एक व्यावहारिक policy यह है कि read-only diagnostics बिना interruption चलें और restart, rollback या configuration बदलने वाले tools के लिए approval अनिवार्य हो। प्रस्तावित action को अस्वीकार करते समय उसे reject करें; documentation चेतावनी देती है कि “respond” का उपयोग tool की तरह कार्रवाई करने के लिए है, side-effect करने वाली call को अस्वीकार करने के लिए नहीं। बदले गए arguments को अपने tool implementation में validate करें: documented middleware edits को प्रदर्शित schema के अनुसार validate नहीं करता।

जिन failure cases के लिए तैयारी करें

• खाली परिणाम: “कोई matching event नहीं मिला” बताएँ, “service स्वस्थ है” नहीं।
• Tool timeout या authorization failure: विफल source और error category बनाए रखें; बिना सीमा के बार-बार retry न करें।
• बार-बार tool calls: Graph execution सीमित करें और tool inputs पर रोक लगाएँ। Retries जोड़ें तो स्पष्ट और कम retry policy रखें तथा attempts की संख्या दिखाएँ।
• Interrupt फिर शुरू न हो पाए: जाँचें कि checkpointer configure है और resume करते समय वही thread identifier इस्तेमाल हो रहा है। Source documentation के अनुसार human review के लिए checkpointing आवश्यक है।
• भ्रामक अंतिम सारांश: Report में दावों के समर्थन में जुटाए गए signals का उल्लेख अनिवार्य करें और unavailable sources तथा अनसुलझे सवालों की सूची दें।

Incident-run checklist

Deploy करने से पहले पुष्टि करें कि diagnostic tools read-only और सीमित दायरे में हैं; time windows और query volume सीमित हैं; tool errors दिखाई देते रहते हैं; graph execution पर limit है; और हर side-effect करने वाली कार्रवाई human decision के लिए रुकती है। Timeout, खाली परिणाम, अस्वीकार की गई कार्रवाई और interrupted run को resume करने वाले paths की जाँच करें। अच्छा agent केवल विश्वसनीय लगने वाला root cause नहीं बताता—वह यह भी दिखाता है कि उसने वास्तव में कौन-सा evidence हासिल किया और उसका diagnosis कहाँ तक सीमित है।

स्रोत

• LangChain: Human-in-the-loop middleware — interrupt decisions, conditional tool review, checkpointing और रुके हुए execution को फिर शुरू करना।