बिना सेंसर LLM API पर एजेंट: टूल कॉलिंग, फ्रेमवर्क और गार्डरेल्स

टूल कॉलिंग लूप की संरचना

फ्रेमवर्क को हटा दें और हर एजेंट केवल चार ही कार्य करता है। आप संवाद और tools सूची भेजते हैं। मॉडल या तो टेक्स्ट के साथ जवाब देता है, जिसका अर्थ है कि कार्य समाप्त हो गया है, या tool_calls के साथ, जिसका अर्थ है कि उसे कुछ निष्पादित करने की आवश्यकता है। आप प्रत्येक कॉल चलाते हैं और प्रत्येक कॉल के लिए एक tool संदेश जोड़ते हैं, कॉल के id का उल्लेख करते हुए। फिर आप लंबे संवाद के साथ मॉडल से फिर से पूछते हैं।

शाखा का संकेत finish_reason है। जब यह tool_calls के बराबर होता है, तो सहायक संदेश में content शून्य पर सेट होता है और एक tool_calls सरणी होती है, जिसमें प्रत्येक प्रविधि में एक id, एक फ़ंक्शन नाम और JSON का एक तर्क स्ट्रिंग होता है। संदर्भ पृष्ठ सटीक आकार दिखाता है। दो विवरण लोगों को फँसाते हैं: तर्क एक स्ट्रिंग है जिसे पार्स करना आवश्यक है, और कॉल्स वाला सहायक संदेश अपने टूल परिणामों से पहले, क्रम में इतिहास में बना रहना चाहिए।

शून्य से एक पूर्ण लूप

नीचे दिया गया स्क्रिप्ट एक फ़ंक्शन परिभाषित करता है, इसे JSON स्कीमा में वर्णित करता है, और जब तक मॉडल सादा पाठ उत्पन्न न करे, लूप चलाता है। यह API की ओर इशारा करने वाले आधिकारिक Python SDK का उपयोग करता है। हर पास में असिस्टेंट संदेश, फिर प्रत्येक कॉल के लिए एक टूल संदेश, और एक स्टेप काउंटर जोड़ा जाता है जो रनअवे व्यवहार को रोकता है।

Copyimport json, os
from openai import OpenAI

client = OpenAI(base_url="https://api.uncensoredapis.com/v1", api_key=os.environ["API_KEY"])

TICKETS = {"T-101": "open", "T-102": "closed"}

def ticket_status(ticket_id: str) -> str:
    return TICKETS.get(ticket_id, "unknown ticket")

TOOLS = [{
    "type": "function",
    "function": {
        "name": "ticket_status",
        "description": "Look up the status of a support ticket by id.",
        "parameters": {
            "type": "object",
            "properties": {"ticket_id": {"type": "string"}},
            "required": ["ticket_id"],
        },
    },
}]
REGISTRY = {"ticket_status": ticket_status}

messages = [{"role": "user", "content": "Are tickets T-101 and T-102 still open?"}]

for step in range(6):                      # hard cap on loop turns
    r = client.chat.completions.create(
        model="uncensored", messages=messages, tools=TOOLS, max_tokens=400)
    msg = r.choices[0].message
    messages.append(msg.model_dump(exclude_none=True))
    if r.choices[0].finish_reason != "tool_calls":
        print(msg.content)
        break
    for call in msg.tool_calls:
        fn = REGISTRY.get(call.function.name)
        try:
            args = json.loads(call.function.arguments)
            result = fn(**args) if fn else f"error: unknown tool {call.function.name}"
        except Exception as exc:
            result = f"error: {exc}"
        messages.append({"role": "tool", "tool_call_id": call.id, "content": str(result)})
else:
    print("stopped: step limit reached")

ध्यान दें कि लूप विफलता पर क्या करता है: त्रुटियाँ स्ट्रिंग्स के रूप में टूल परिणाम बन जाती हैं। यह मॉडल को समस्या पढ़ने और फिर से प्रयास करने की अनुमति देता है, उदाहरण के लिए एक खराब टिकट id को ठीक करना, आपके प्रोग्राम के क्रैश होने के बजाय। for ... else क्लॉज़ केवल तब फायर होता है जब अंतिम उत्तर के बिना कैप हिट हो जाता है, जो यह पता लगाने का तरीका है कि मॉडल चक्कर में फँस गया है।

प्रॉम्प्ट को केंद्रित रखें। एक छोटा सिस्टम संदेश जो बताता है कि कब टूल कॉल करना है और कब सीधे जवाब देना है, निरर्थक कॉल कम करता है, और प्रत्येक स्किप्ड कॉल एक राउंड ट्रिप, टोकन और आपके रेट लिमिट का एक स्लॉट बचाता है।

टूल निष्पादन की रक्षा

मॉडल आउटपुट अविश्वसनीय इनपुट है। मॉडल द्वारा चुना गया टूल नाम या तर्क गलत, अजीब हो सकता है, या मॉडल द्वारा एक वेब पेज या उपयोगकर्ता से पढ़े गए टेक्स्ट से प्रभावित हो सकता है। हर कॉल को सार्वजनिक इंटरनेट से आने वाले अनुरोध की तरह मानें।

• नाम द्वारा अनुमति सूची। एक रजिस्ट्री में फ़ंक्शन को खोजें जो आपने बनाया है। कोड को इम्पोर्ट या एवल्यूएट करने के लिए स्ट्रिंग का उपयोग कभी न करें।
• पढ़ने और लिखने को अलग करें। केवल-पढ़ने वाले टूल स्वतंत्र रूप से चलते हैं। हर वह चीज जो ईमेल भेजती है, पैसा खर्च करती है, डेटा हटाती है या फ़ाइलों को छूती है, उसे स्पष्ट मानवीय अनुमोदन की आवश्यकता होती है।
• तर्कों की सत्यापन उस स्कीमा के खिलाफ जो आपने घोषित किया था: आवश्यक कुंजियाँ, कोई अतिरिक्त कुंजियाँ नहीं, प्रकार और लंबता सीमाएँ।
• समय और आउटपुट को बाँधें। टाइमआउट लागू करें, और उन्हें वापस फ़ीड करने से पहले परिणामों को ट्रंक करें ताकि एक विशाल टूल आउटपुट आपके 100,000-टोकन कॉन्टेक्स्ट विंडो को न खाए।
• कार्य और लागत पर सीमा लगाएं। प्रत्येक कार्य के लिए चरणों की संख्या और प्रत्येक सत्र के लिए टोकन की मात्रा को सीमित करें, usage ब्लॉक का उपयोग करें।
• नाम, तर्क और परिणाम के साथ प्रत्येक कॉल लॉग करें ताकि आप व्यवहार की जाँच कर सकें।

Copyimport json
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutTimeout

ALLOWED = {"ticket_status", "lookup_invoice"}      # read-only tools only
NEEDS_APPROVAL = {"refund_order"}                  # side effects wait for a human
POOL = ThreadPoolExecutor(max_workers=4)

def validate(args: dict, schema: dict) -> None:
    props = schema["properties"]
    for key in schema.get("required", []):
        if key not in args:
            raise ValueError(f"missing argument: {key}")
    for key, val in args.items():
        if key not in props:
            raise ValueError(f"unexpected argument: {key}")
        if props[key]["type"] == "string" and not isinstance(val, str):
            raise ValueError(f"{key} must be a string")
        if isinstance(val, str) and len(val) > 200:
            raise ValueError(f"{key} too long")

def run_tool(name, raw_args, registry, schemas, approve=lambda n, a: False):
    if name in NEEDS_APPROVAL:
        args = json.loads(raw_args)
        if not approve(name, args):
            return "denied: this action needs human approval"
    elif name not in ALLOWED:
        return f"error: tool {name} is not permitted"
    args = json.loads(raw_args)
    validate(args, schemas[name]["parameters"])
    future = POOL.submit(registry[name], **args)
    try:
        return str(future.result(timeout=5))[:2000]    # time and size limits
    except FutTimeout:
        return "error: tool timed out"

टूल परिणामों को भी अविश्वसनीय मानें। यदि एक टूल स्क्रेप्ड टेक्स्ट लौटाता है जिसमें निर्देश शामिल हैं, तो मॉडल उन्हें फॉलो कर सकता है। ऐसे सामग्री को टूल संदेश में स्पष्ट डिलिमीटर में लपेटें और अपने सिस्टम प्रॉम्प्ट में बताएं कि टूल आउटपुट डेटा है, कमांड नहीं।

मॉडल द्वारा उपयोग योग्य टूल स्कीमा लिखना

स्कीमा वह एकमात्र दस्तावेज़ है जिसे मॉडल देखता है, इसलिए इसे एक सार्वजनिक API की तरह ही सावधानी से तैयार करें। कार्यों के नाम ऐसे क्रियापद और संज्ञाओं से रखें जो कार्य से मेल खाते हों, जैसे ticket_status या lookup_invoice, न कि सामान्य लेबल जैसे run या helper। description को एक निर्देश के रूप में लिखें: फ़ंक्शन क्या करता है, यह क्या लौटाता है, और कब इसे उपयोग न करें।

• पैरामीटर कम और सरल रखें। दो या तीन सरल फ़ील्ड्स दस फ़ील्ड्स वाले नेस्टेड ऑब्जेक्ट की तुलना में कहीं अधिक बार सही ढंग से कॉल किए जाते हैं।
• बंद सेट के लिए एन्यूमेरेटर्स (enums) का उपयोग करें, उदाहरण के लिए "enum": ["open", "closed"], ताकि मॉडल कोई मान गढ़ न सके।
• प्रत्येक संपत्ति का वर्णन इकाई या उदाहरण के साथ करें: "टिकट आईडी जैसे T-101" एक सादे स्ट्रिंग से बेहतर है।
• संक्षिप्त परिणाम लौटाएं। स्ट्रिंग या छोटे JSON ऑब्जेक्ट सबसे अच्छे काम करते हैं; लंबे डंप कॉन्टेक्स्ट को कम लाभ के लिए जलाते हैं।
• एक ओम्निबस टूल जिसमें मोड फ्लैग हो, की तुलना में कई संकीर्ण टूल्स को प्राथमिकता दें।

जब कॉल वैलिडेशन में विफल होती है, तो उठाए जाने के बजाय वैलिडेशन संदेश को टूल परिणाम के रूप में लौटाएं। मॉडल आमतौर पर अगली बारी में कॉल को ठीक कर देता है, क्योंकि यह यह पढ़ सकता है कि क्या गलत था। यदि यह उसी तरह दो बार विफल होता है, तो लूप को न बदलकर विवरण को फिर से लिखें।

LangChain कॉन्फ़िगरेशन

LangChain की ChatOpenAI क्लास एक base_url स्वीकार करती है, इसलिए किसी कस्टम प्रदाता पैकेज की आवश्यकता नहीं है। model को uncensored पर सेट करें, अपने वातावरण से कुंजी पास करें, और bind_tools के साथ टूल्स बांड करें। langchain-openai और langchain-core इंस्टॉल करें।

Copyimport os
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def ticket_status(ticket_id: str) -> str:
    """Return the status of a support ticket."""
    return {"T-101": "open", "T-102": "closed"}.get(ticket_id, "unknown ticket")

llm = ChatOpenAI(
    model="uncensored",
    base_url="https://api.uncensoredapis.com/v1",
    api_key=os.environ["API_KEY"],
    temperature=0.3,
    max_tokens=400,
)
llm_with_tools = llm.bind_tools([ticket_status])

ai = llm_with_tools.invoke("What is the status of ticket T-101?")
for call in ai.tool_calls:
    print(call["name"], call["args"], "->", ticket_status.invoke(call["args"]))

invoke का परिणाम एक AIMessage है जिसकी tool_calls सूची में पार्स किए गए नाम और तर्क होते हैं, इसलिए आपको मैन्युअल JSON चरण छोड़ना पड़ता है। एक पूर्ण स्वतंत्र लूप प्राप्त करने के लिए, LangChain के एजेंट कंस्ट्रक्टरों को समान llm ऑब्जेक्ट दें; एंडपॉइंट को केवल टूल कॉलिंग को सम्मानित करना होता है, जो वह करता है। यदि कोई एजेंट अपने टूलों को अनदेखा करता प्रतीत होता है, तो जाँच करें कि डॉकस्ट्रिंग और टाइप इंट्स मौजूद हैं, क्योंकि वे वह स्कीमा बन जाते हैं जिसे मॉडल देखता है।

LlamaIndex कॉन्फ़िगरेशन

LlamaIndex में OpenAILike सर्वर के लिए शामिल है जो चैट पूर्णता फॉर्मेट का पालन करते हैं। तीन सेटिंग्स महत्वपूर्ण हैं। api_base API की ओर इशारा करता है, is_chat_model=True चैट रूट को चुनता है, और is_function_calling_model=True फ्रेमवर्क को बताता है कि यह टूल्स भेज सकता है। context_window को 100000 पर सेट करें ताकि इसके टेक्स्ट स्प्लिटर और मेमोरी बफर्स सही ढंग से आकार ले सकें। llama-index-llms-openai-like इंस्टॉल करें।

Copyimport os
from llama_index.llms.openai_like import OpenAILike
from llama_index.core.tools import FunctionTool

def ticket_status(ticket_id: str) -> str:
    """Return the status of a support ticket."""
    return {"T-101": "open", "T-102": "closed"}.get(ticket_id, "unknown ticket")

llm = OpenAILike(
    model="uncensored",
    api_base="https://api.uncensoredapis.com/v1",
    api_key=os.environ["API_KEY"],
    is_chat_model=True,
    is_function_calling_model=True,
    context_window=100000,
    max_tokens=400,
)

tool = FunctionTool.from_defaults(fn=ticket_status)
response = llm.predict_and_call([tool], "Check ticket T-102 for me.")
print(response)

यहाँ से आप llm को एजेंट वर्कर या क्वेरी इंजन को दे सकते हैं। याद रखें कि रेट्रीवल पाइपलाइनें एक प्रॉम्प्ट में कई चंक भर सकती हैं; प्रॉम्प्ट और पूर्णता के बीच साझा 100,000-टोकन विंडो के साथ, प्राप्त किए गए चंकों की संख्या को सीमित करें और max_tokens उचित रखें।

बजट पर एजेंट संचालित करना

एजेंट एक ही उपयोगकर्ता संदेश को कई अनुरोधों में बदल देता है, इसलिए क्षमता की योजना बनाएँ। पाँच टूल राउंड्स वाले कार्य में छह मॉडल कॉल्स लगते हैं। दस समानांतर कार्य अल्पकाल में छत्तीस अनुरोध उत्पन्न करते हैं, और प्रति मिनट 300 अनुरोध की सीमा एक चैट ऐप जितनी जल्दी पहुँचती है, उससे भी जल्दी पहुँच जाती है। त्रुटियों और रेट लिमिट्स के प्लेबुक से टोकन-बुकेथ्रॉटल को अपने क्लाइंट के सामने रखें।

प्रत्येक टूल परिणाम जोड़े जाने के कारण संदर्भ बढ़ता जाता है। एक बार उपयोग हो जाने पर पुराने टूल आउटपुट को सारांशित करें या हटा दें, और usage.prompt_tokens पर नज़र रखें; जब यह 50,000 के करीब पहुँच जाए, तो इतिहास को संपीड़ित करें। लागत के लिए, 6 कॉल्स वाले एक उदाहरण कार्य की कल्पना करें जिसमें औसतन 3,000 प्रॉम्प्ट टोकन और 200 पूर्णता टोकन हैं: यह 18,000 इनपुट टोकन (प्रति मिलियन टोकन $0.25 पर लगभग $0.0045) और 1,200 आउटपुट टोकन (प्रति मिलियन टोकन $1.00 पर लगभग $0.0012) है। ये टोकन गणनाएँ योगों के लिए कल्पनाएँ हैं, माप नहीं। यदि आप एंडपॉइंट से नए हैं, तो docs और मुफ़्त ट्रायल क्रेडिट से शुरू करें।

एजेंट का परीक्षण और डिबगिंग

एजेंट ऐसे तरीकों से विफल होते हैं जिनकी एकल फ़ंक्शन पर यूनिट टेस्ट नहीं पकड़ पाते, इसलिए बातचील का परीक्षण करें। हर रन के लिए पूर्ण संदेश सूची रिकॉर्ड करें, जिसमें टूल तर्क और परिणाम शामिल हों, और लॉग से विफलताओं को रीप्ले करें। तीन आदतें जल्दी फायदा पहुंचाती हैं।

1. मॉडल को झूठा बनाएं। क्लाइंट को स्टब करें ताकि यह स्क्रिप्टेड tool_calls संदेश लौटाए। यह आपकी रजिस्ट्री, सत्यापन और लूप टर्मिनेशन को नेटवर्क और खर्च के बिना परीक्षण करता है।
2. पिन सैंपलिंग। टूल-संचालित कार्यों के लिए कम temperature, जैसे 0.2, का उपयोग करें। कम यादृच्छिकता अधिक दोहराए जाने योग्य तर्क JSON देती है, और आप लेखन-केंद्रित चरणों के लिए इसे बाद में बढ़ा सकते हैं।
3. ट्राजेक्टरी पर दावा करें। कुछ निश्चित प्रॉम्प्ट्स के लिए, केवल अंतिम पाठ की जाँच न करें, बल्कि यह जाँचें कि कौन से टूल कॉल किए गए और किस क्रम में।

finish_reason को भी देखें। टूल कॉल के बीच में length का मान इसका संकेत है कि तर्क कट गए हैं; max_tokens (32,000 तक) बढ़ाएं और JSON पूर्ण हो जाएगा। टूल के साथ स्ट्रीमिंग भी काम करती है, लेकिन प्रत्येक कॉल इंडेक्स के लिए तर्क के टुकड़ों को जमा करें और स्ट्रीम बंद होने के बाद ही पार्स करें। उन चंक्स के बारे में पुनः याद दिलाने के लिए, संदर्भ के स्ट्रीमिंग सेक्शन को देखें।