جولائی 2025 میں، Claude Code recursive loop 5 گھنٹے میں 16,000 USD اور 50,000 USD کے درمیان جل گیا۔ کوئی کریش یا غلطیاں نہیں تھیں، اور چونکہ کسی نے ہمیں یہ نہیں بتایا کہ کب رکنا ہے، ایجنٹوں نے بالکل ویسا ہی کیا جیسا کہ انہیں بتایا گیا تھا۔
4 مہینوں کے بعد، 4 ایجنٹوں کے ساتھ ایک LangChain لوپ 11 دن تک چلا اور اس کی قیمت 47,000 USD تھی۔ بل آنے تک کسی نے نوٹس نہیں لیا۔ پائپ لائن نے ہمارے ٹیسٹوں میں صحیح کارکردگی کا مظاہرہ کیا اور ایجنٹوں نے بالکل وہی کیا جیسا کہ ہدایت کی گئی تھی۔ ایک ہی پیٹرن.
یہ ٹیوٹوریل غائب کمانڈز کے بارے میں ہے۔
ہم پانچ چھوٹے Python پرائمیٹوز بناتے ہیں جو زیادہ تر ایجنٹ لوپ کی غلطیوں کو جاری ہونے سے پہلے پکڑ لیتے ہیں۔
-
کوئی راستہ نہیں مخصوص مصنف لوپ شروع ہونے سے پہلے انجام دیا جانا ضروری ہے۔
-
کوئی راستہ نہیں سرکٹ بریکر سخت حد سے تجاوز کرنے پر لوپ سے باہر نکلیں۔
-
کوئی راستہ نہیں لیجر صرف ضمیمہ SQLite آڈٹ ٹریل میں ہر چیز کو لاگ کرتا ہے۔
-
نہیں ایجنٹ لوپ جو تینوں کو آپس میں جوڑتا ہے۔
-
کوئی راستہ نہیں سطح کا جائزہ خیال یہ ہے کہ نیچے دھارے کے نظام کو کچھ بھی موصول ہونے سے پہلے انسانی تصدیق پر مجبور کیا جائے۔
آخر کار آپ کے پاس ایک کام کرنے والا ذخیرہ ہوگا جس میں آپ کے تمام ایجنٹ پروجیکٹ جاسکتے ہیں۔ مکمل کوڈ github.com/dannwaneri/production-safe-agent-loop پر ہے۔
انڈیکس
-
ایسا کیوں ہوتا رہتا ہے؟
-
شرطیں
-
مرحلہ 1: تعمیر سے پہلے مکمل کی وضاحت کریں۔
-
مرحلہ 2: رن ٹائم پر تکمیل کو انجام دیں۔
-
مرحلہ 3: سب کچھ ریکارڈ کریں۔
-
مرحلہ 4: حدود کا احترام کرتے ہوئے لوپ
-
مرحلہ 5: سطح کا جائزہ لیں۔
-
مرحلہ 6: حقیقی زندگی کی مثال، SEO آڈیٹر
-
پلگ ایبل ایل ایل ایم کلائنٹ
-
ٹیسٹ چلائیں
-
آپ نے کیا بنایا
-
اگلے اقدامات
ایسا کیوں ہوتا رہتا ہے؟
کمپنی کو مشکل میں ڈالنے والا حساب کتاب آسان تھا۔ ایک چیٹ بوٹ کی قیمت تقریباً $0.04 فی تعامل ہے۔ سکیلڈ ملٹی ایجنٹ ورک فلو لاگت $1.20 ہے۔ یہ ایک 30x ضرب ہے، اور پیداوار کے معیارات سے پتہ چلتا ہے کہ پیچیدہ کاموں کے لیے 70x تک پہنچا جا سکتا ہے۔
مسئلہ یہ نہیں ہے کہ ایجنٹ مہنگے ہیں۔ مسئلہ یہ ہے کہ زیادہ تر ٹیموں نے چیٹ بوٹ کے اخراجات کے لیے بجٹ بنایا ہے اور ایجنٹ کے فن تعمیر کو تعینات کیا ہے۔ گارٹنر نے پایا کہ پائلٹ چیٹ بوٹس اور پروڈکشن ایجنٹ ورک فلو کے درمیان ٹوکن کی کھپت کا فرق 5 سے 30 گنا تک ہے۔ FinOps فاؤنڈیشن کی 2026 اسٹیٹ آف FinOps رپورٹ کے مطابق، 73% کمپنیوں نے کہا کہ ان کی AI کی قیمتیں اصل توقع سے زیادہ تھیں۔
ایک بار جب آپ اسے دیکھیں گے تو طریقہ کار آسان ہے۔ اگر ایجنٹ کسی آپریشن میں ناکام ہو جاتا ہے اور دوبارہ کوشش کرتا ہے، تو یہ نئے سرے سے شروع نہیں ہوتا ہے۔ دوبارہ کوشش کرنے سے پہلے ہر ناکام کوشش کے لیے پوری سیاق و سباق کی ونڈو کو دوبارہ پڑھا جاتا ہے۔ پہلی تکرار کے لیے 100 ٹوکن درکار ہیں۔ 2 تکرار کی لاگت 200 ہے۔ دس دہرانے پر ہزاروں ڈالر خرچ ہو سکتے ہیں۔ آپ ہر ناکامی کی ادائیگی جاری رکھیں گے، ملی سیکنڈ بہ ملی سیکنڈ۔
# This is the entire problem in three lines
while True:
result = agent.run(task)
# done when...?
سوالیہ نشان یہ ہے کہ پیسہ کہاں جاتا ہے۔
ایک اور چیز جو چیزوں کو بدتر بناتی ہے وہ یہ ہے کہ ایجنٹ نمایاں طور پر ناکام نہیں ہوتے ہیں۔ موجودہ کوڈ ایک غیر متعینہ حالت تک پہنچ جاتا ہے اور کریش ہو جاتا ہے۔ LLM ابہام کو حل کرنے اور مددگار بننے کی کوشش کرتا ہے۔ براہ کرم دوبارہ کوشش کریں۔ ٹول کال کو دوبارہ فارمیٹ کریں۔ تصدیقی ایجنٹ چلائیں۔ تصدیق کرنے والے ایجنٹ نے کچھ دریافت کیا۔ اصلاحی افسران کو برطرف۔ کسی نے بھی اس بات کی وضاحت نہیں کی کہ "صحیح” کا کیا مطلب ہے۔ آپ کے تمام ڈیش بورڈز پر لوپس خوبصورت نظر آتے ہیں – سرگرمیاں، ٹول کالز، فیصد مکمل، وغیرہ – خاموشی سے اپنا بجٹ کھاتے ہوئے
گارٹنر نے پیش گوئی کی ہے کہ 40% ایجنٹ پروجیکٹس 2027 تک معاشی ناکامی کی وجہ سے ترک کر دیے جائیں گے۔ زیادہ تر ناکامیاں روکی جا سکتی ہیں۔ برطرفی کی شرائط ہیں، بہتر ماڈل نہیں۔
شرطیں
git clone https://github.com/dannwaneri/production-safe-agent-loop
cd production-safe-agent-loop
pip install -r requirements.txt
export ANTHROPIC_API_KEY=sk-...
مرحلہ 1: تعمیر سے پہلے مکمل کی وضاحت کریں۔
ایجنٹ کی ترقی میں سب سے مہنگی غلطی غلط ماڈل کا انتخاب نہ کرنا یا دوبارہ کوشش کی حد کو کھونا ہے۔ تعمیر شروع ہوتی ہے اس سے پہلے کہ آپ ایک جملے میں ایک سوال کا جواب دے سکیں۔
تکمیل کیسی نظر آتی ہے؟
زیادہ تر ٹیمیں اس کا جواب نہیں دے سکتیں۔ یہ اس لیے نہیں کہ میں لاپرواہ ہوں، اس لیے کہ میں ٹرمینل کھولنے سے پہلے کسی چیز پر زبردستی نہیں کرتا۔ مخصوص مصنف فنکشن کو نافذ کرتا ہے۔
# spec_writer.py
from spec_writer import SpecWriter
spec = SpecWriter(db_path="spec.db").run()
جب آپ کال کرتے ہیں۔ .run()اسے واپس نہیں کیا جائے گا جب تک کہ تین سوالوں کے جواب نہ مل جائیں۔
-
یہ کیا کرتا ہے؟
-
یہ کیا نہیں کرتا؟
-
ایک جملے میں کیا نظر آتا ہے؟
تیسرا سوال اہم ہے۔ یہ سب سے مشکل کام بھی ہے۔ "کسی ایجنٹ سے سائٹ کا آڈٹ کروائیں” جواب نہیں ہے۔ "ایجنٹ ٹارگٹ یو آر ایل کو کرال کرتا ہے اور سب کچھ نکالتا ہے۔ ٹیگز، جھنڈا غائب یا زیادہ طوالت، اور سٹاپ” جوابات ہیں۔ ان میں سے ایک سرکٹ بریکر کو نافذ کرنے کے لیے کچھ دیتا ہے۔
وضاحتیں SQLite میں محفوظ ہیں۔ SpecResult ڈیٹا کلاس session_id. وہ ID وہ دھاگہ بن جاتا ہے جو وضاحتیں، لیجر کی قطاریں، اور لوپ کے نتائج کو جوڑتا ہے۔ ایک سیشن کو سرے سے آخر تک ٹریک کیا جا سکتا ہے۔
@dataclass(frozen=True)
class SpecResult:
what_it_does: str
what_it_does_not: str
done_looks_like: str
session_id: str
frozen=True یہ ضروری ہے۔ تصریح ایک وعدہ ہے، مسودہ نہیں۔ ایک بار لکھنے کے بعد، ایک لوپ اس پر چلتا ہے. کوئی درمیانی اصلاحات نہیں ہیں۔
جانچ کے مقاصد کے لیے، SpecWriter انجیکشن قبول کریں۔ input_fn اور output_fn کال کے قابل۔ کوئی معیاری ان پٹ بندر پیچنگ کی ضرورت نہیں ہے۔ دیکھیں tests/test_spec_writer.py کام کرنے والی مثال کے طور پر – خاندان چھوٹا ہے۔ scripted_input ایک مددگار جو جنریٹر سے جوابات واپس کرتا ہے اور انہیں pytest کے ذریعے ٹیسٹ کے لیے مخصوص SQLite فائل میں لکھتا ہے۔ tmp_path فکسچر SQLite :memory: کیونکہ یہ یہاں محفوظ نہیں ہے۔ SpecWriter فی طریقہ ایک نیا کنکشن کھولتا ہے۔ :memory: کنکشن اس کا اپنا الگ تھلگ ڈیٹا بیس ہوتا ہے۔
مرحلہ 2: رن ٹائم پر تکمیل کو انجام دیں۔
اپ اسٹریم ختم کرنے کی شرائط کی وضاحت کرنا ایک نظم و ضبط ہے۔ سرکٹ بریکرز اثر میں ہیں۔
# circuit_breaker.py
from circuit_breaker import CircuitBreaker, CircuitBreakerError
breaker = CircuitBreaker(turn_limit=5, token_limit=15000)
breaker.check(turn_count, accumulated_tokens) # raises on breach
دو چھتیں ہیں۔ دونوں مشکل ہیں۔
turn_limit ایک لوپ LLM کو کال کرنے کی تعداد کو محدود کرتا ہے۔ token_limit تمام موڑ پر کل ٹوکن کی کھپت کو محدود کرتا ہے۔ ٹرپنگ میں سے ایک اٹھاتا ہے CircuitBreakerError فوری طور پر
حدود سخت ہیں۔ turn_count == turn_limit اس کی اجازت ہے۔ turn_count == turn_limit + 1 سفر کوئی رعایتی مدت یا انتباہ نہیں ہے۔ زبردستی روکنا انسان کو چوکی چلانے پر مجبور کرتا ہے۔
from dataclasses import dataclass
@dataclass
class CircuitBreakerError(Exception):
reason: str # "turn_ceiling" or "token_ceiling"
turn_count: int
accumulated_tokens: int
def __post_init__(self) -> None:
super().__init__(
f"circuit breaker tripped: {self.reason} "
f"(turn={self.turn_count}, tokens={self.accumulated_tokens})"
)
class CircuitBreaker:
def __init__(self, turn_limit: int = 5, token_limit: int = 15000) -> None:
self.turn_limit = turn_limit
self.token_limit = token_limit
def check(self, turn_count: int, accumulated_tokens: int) -> None:
if turn_count > self.turn_limit:
self._trip("turn_ceiling", turn_count, accumulated_tokens)
if accumulated_tokens > self.token_limit:
self._trip("token_ceiling", turn_count, accumulated_tokens)
def _trip(self, reason: str, turn_count: int, accumulated_tokens: int) -> None:
print(
"\n=== CIRCUIT BREAKER CHECKPOINT ===\n"
f"reason : {reason}\n"
f"turn_count : {turn_count} / limit {self.turn_limit}\n"
f"tokens_used : {accumulated_tokens} / limit {self.token_limit}\n"
"action : halt loop, surface to human reviewer\n"
"=================================="
)
raise CircuitBreakerError(
reason=reason,
turn_count=turn_count,
accumulated_tokens=accumulated_tokens,
)
CircuitBreakerError یہ ایک استثناء ہے، واپسی کوڈ نہیں۔ یہ جان بوجھ کر ہے۔ واپسی کوڈ کو نظر انداز کیا جا سکتا ہے۔ غیر پکڑے گئے مستثنیات ایسا نہیں کر سکتے۔ ایک خاموش خلاف ورزی ناممکن ہے. ایک انسانی پڑھنے کے قابل چیک پوائنٹ بینر کو معیاری آؤٹ پٹ پر پرنٹ کیا جاتا ہے: _trip() پہلے ایک استثنیٰ ڈالا جاتا ہے، اس لیے اگر کال کرنے والا استثناء کو نگل بھی لے، تب بھی آپریٹر ریاست کو دیکھ سکتا ہے۔
اہم اصول: فون .check() پہلے کوئی LLM کال، LLM کال کے بعد نہیں۔ پرواز کے بعد کی تصدیق کا مطلب ہے کہ آپ نے اپنے ٹوکنز کو پہلے ہی جلا دیا ہے اس سے پہلے کہ آپ کو معلوم ہو کہ حد سے تجاوز کر گیا ہے۔
# Wrong — post-flight
result = client.messages.create(...)
breaker.check(turn_count, accumulated_tokens) # too late
# Right — pre-flight
breaker.check(turn_count, accumulated_tokens) # raises before any spend
result = client.messages.create(...)
ڈیفالٹس (5 موڑ، 15,000 ٹوکن) سخت ٹیوٹوریل ڈیمو کے ساتھ مطابقت رکھتے ہیں۔ پیداواری بجٹ مختلف ہیں۔ انسٹی ٹیوٹ پر ایڈجسٹمنٹ:
# Production example — tighter token budget, more turns
breaker = CircuitBreaker(turn_limit=10, token_limit=50000)
مرحلہ 3: سب کچھ ریکارڈ کریں۔
سرکٹ بریکر آپ کے بینک کھاتوں کی حفاظت کرتے ہیں۔ لیجر آپ کی سمجھ کی حفاظت کرتا ہے کہ کیا ہوا ہے۔
زیادہ تر ٹیمیں ڈیبگنگ کے لیے لاگ ان کرتی ہیں۔ ایک مسئلہ ہونے کے بعد، آپ جاننا چاہتے ہیں کہ کیا غلط ہوا ہے۔ لیجرز کے مختلف مقاصد ہوتے ہیں۔ یہ گورننس ہے۔ ہر قطار ثابت کرتی ہے کہ آیا لوپ حدود کے اندر رہا، آیا نہیں، اور بالکل کب۔
# ledger.py
from ledger import Ledger
ledger = Ledger(db_path="ledger.db")
ledger.write(
session_id=spec.session_id,
turn_count=1,
state_origin="llm",
input_str=task,
token_delta=523,
execution_time_ms=1240,
pass_fail=True,
)
ایک لائن فی موڑ۔ صرف شامل کریں، کوئی اپ ڈیٹ نہیں، کوئی حذف نہیں۔ عدم تغیر کلید ہے۔ جس لیجر میں ترمیم کی جا سکتی ہے وہ نوٹ بک ہے، لیجر نہیں۔
سکیما:
CREATE TABLE IF NOT EXISTS ledger (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
turn_count INTEGER NOT NULL,
state_origin TEXT NOT NULL,
input_hash TEXT NOT NULL,
token_delta INTEGER NOT NULL,
execution_time_ms INTEGER NOT NULL,
pass_fail INTEGER NOT NULL, -- 1=pass, 0=fail
breach_reason TEXT, -- NULL unless circuit breaker fired
created_at TEXT NOT NULL -- ISO 8601, UTC
);
CREATE INDEX IF NOT EXISTS idx_ledger_session ON ledger(session_id);
جسو بناتا ہے۔ get_session(session_id) – بنیادی پڑھنے کا راستہ – جیسا کہ لیجر بڑھتا ہے مسلسل تلاش۔
وضاحت کے قابل تین فیصلے:
-
input_hash~ نہیںinput_text. خام ان پٹ سٹرنگز برقرار نہیں ہیں۔ صرف SHA-256 ہیشز قبول کی جاتی ہیں۔ اس کے دو فائدے ہیں: پورے عمل میں ایک ہی ان پٹ کا پتہ لگایا جا سکتا ہے اور آڈٹ ٹریل میں کوئی PII داخل نہیں کیا جاتا ہے۔ -
pass_failپسندINTEGER~ نہیںBOOLEAN. SQLite میں کوئی بولین قسم نہیں ہے۔1اور0یہ سرکاری ہے۔ API کے کنارے پر Python ergonomics کو صاف کریں اور ڈسک پر SQL کی قسمیں ٹھیک کریں۔ -
created_atپسندdatetime.now(timezone.utc).isoformat().datetime.utcnow()Python 3.12 میں فرسودہ۔ ٹائم زون سے آگاہ ٹائم اسٹیمپ ٹائم زون کو عبور کرنے والے کسی بھی سسٹم پر فٹگن کو روکتے ہیں۔
سیشن کے لحاظ سے تلاش کریں:
rows = ledger.get_session(spec.session_id)
for row in rows:
print(f"Turn {row.turn_count}: {'PASS' if row.pass_fail else 'FAIL'} "
f"| {row.token_delta} tokens | {row.execution_time_ms}ms")
مرحلہ 4: حدود کا احترام کرتے ہوئے لوپ
ایجنٹ لوپ تین پرائمیٹوز کو ایک ساتھ جوڑتا ہے۔ یہ واحد جزو ہے جو ایل ایل ایم کہتا ہے۔ باقی سب کچھ مقامی ہے۔
# agent_loop.py
from agent_loop import AgentLoop
loop = AgentLoop(spec, breaker, ledger, client)
result = loop.run(task)
# LoopResult(success, turns, total_tokens, session_id, breach_reason)
موڑ کی ساخت کو مندرجہ ذیل ترتیب میں بیان کیا گیا ہے۔
-
circuit_breaker.check(turn_count, accumulated_tokens)– اگر آپ حد سے تجاوز کرتے ہیں تو یہ بڑھ جائے گا۔ -
client.messages.create(...)– ایک حقیقی LLM کال -
ledger.write(...)– 1 قطار، صرف شامل کریں۔ -
اگر
stop_reason == "end_turn"واپسی ورنہ لوپ.
ہم بغیر کسی استثناء کے ہر LLM کال سے پہلے فلائٹ چیک کرتے ہیں۔
def run(self, task: str) -> LoopResult:
session_id = self.spec.session_id
messages: list[dict] = [{"role": "user", "content": task}]
turn = 0
total_tokens = 0
try:
while True:
turn += 1
self.circuit_breaker.check(turn, total_tokens)
started = time.perf_counter()
response = self.client.messages.create(
model=self.model,
max_tokens=self.max_tokens,
system=self._system_prompt(),
messages=messages,
)
elapsed_ms = int((time.perf_counter() - started) * 1000)
turn_tokens = (
getattr(response.usage, "input_tokens", 0)
+ getattr(response.usage, "output_tokens", 0)
)
total_tokens += turn_tokens
text = self._text_from(response)
messages.append({"role": "assistant", "content": text})
self.ledger.write(
session_id=session_id,
turn_count=turn,
state_origin="llm",
input_str=task,
token_delta=turn_tokens,
execution_time_ms=elapsed_ms,
pass_fail=True,
)
if getattr(response, "stop_reason", "end_turn") == "end_turn":
return LoopResult(
success=True,
turns=turn,
total_tokens=total_tokens,
session_id=session_id,
)
messages.append({"role": "user", "content": "continue"})
except CircuitBreakerError as err:
self.ledger.write(
session_id=session_id,
turn_count=turn,
state_origin="circuit_breaker",
input_str=task,
token_delta=0,
execution_time_ms=0,
pass_fail=False,
breach_reason=err.reason,
)
return LoopResult(
success=False,
turns=turn,
total_tokens=total_tokens,
session_id=session_id,
breach_reason=err.reason,
)
def _system_prompt(self) -> str:
return (
"You are an agent working on a tightly-scoped task.\n\n"
f"What this does: {self.spec.what_it_does}\n"
f"What this does NOT do: {self.spec.what_it_does_not}\n"
f"Done looks like: {self.spec.done_looks_like}\n"
)
@staticmethod
def _text_from(response) -> str:
content = getattr(response, "content", None)
if not content:
return ""
block = content[0]
return getattr(block, "text", "") or ""
اس متن میں کچھ اختیارات قابل ذکر ہیں:
-
پوری
while True:ایک کے طور پر پیک کیاtry/except CircuitBreakerError. چیک ہر موڑ کے اوپر کیا جاتا ہے، لہذا خلاف ورزی اسی طرح پکڑی جاتی ہے چاہے یہ موڑ 1 پر ہو یا 6 موڑ پر۔ -
input_str=taskہر لیجر قطار پر – اصل آپریشن، آخری ثانوی پیغام نہیں۔ کہinput_hashپھر کالم قطاروں کو گروپ کرتا ہے جو پھانسیوں میں ایک ہی ابتدائی ان پٹ کا اشتراک کرتی ہے۔ -
pass_fail=Trueہر ایل ایل ایم کی باری واپسی کے لیے,Falseصرف خلاف ورزی کی صورت میں۔ پاس/فیل فلیگ اس بات پر نظر رکھتا ہے کہ لوپ چل رہا ہے یا نہیں۔ پہنچ گئے یہ اس بارے میں نہیں ہے کہ آیا ماڈل کا آؤٹ پٹ اچھا ہے یا نہیں، لیکن آیا یہ ایک جائز قطار ہے۔ معیار کی تشخیص ایک الگ تشویش ہے۔ -
_system_prompt()تینوں وضاحتی فیلڈز استعمال کریں۔ہاںdone_looks_like. ماڈل میں منفی رینج ہے (what_it_does_not) کم از کم جہاں تک مثبت حد تک۔ -
time.perf_counter()~ نہیںtime.time()– نیرس اور انٹرمیڈیٹ وال کلاک ایڈجسٹمنٹ سے غیر متاثر۔
LoopResult.session_id سے وراثت میں ملا ہے spec.session_id. لیجر کی قطاریں بغیر کسی جوڑ کے تفصیلات سے منسلک ہیں۔ ایک سیشن ID، ایک ٹریس ایبل ایگزیکیوشن، ختم ہونا شروع کریں۔
مرحلہ 5: سطح کا جائزہ لیں۔
سرکٹ بریکر آپ کے بینک کھاتوں کی حفاظت کرتے ہیں۔ لیجر ریکارڈ کرتا ہے کہ کیا ہوا۔ لیکن نہ ہی آپ کو یہ بتاتا ہے کہ کیا وہی ہوا جس کا وعدہ کیا گیا تھا۔
وہ خلا ہے جہاں غلط لوپ کو قبول کیا جاتا ہے۔ پالش آؤٹ پٹ، گرین ڈیش بورڈ، اپنے وعدوں کو پورا کرنے میں ناکام رہا۔ جائزہ لینے والا نمونے کو دیکھتا ہے، اس کے مناسب ہونے کا تعین کرتا ہے، اور اسے منظور کرتا ہے۔ کسی نے یہ نہیں پوچھا کہ کیا اصل وعدہ پورا ہوا؟
اس خلا کو بند کرنے والی سطحوں کا جائزہ لیں۔ SQLite سیشن کو پڑھتا ہے، پانچ عناصر کا ایک فریم جمع کرتا ہے، اور ڈاون اسٹریم کو آؤٹ پٹ حاصل کرنے سے پہلے موازنہ پر مجبور کرتا ہے۔
from review_surface import ReviewSurface
rs = ReviewSurface(spec_db_path="spec.db", ledger_db_path="ledger.db")
print(rs.render(session_id))
فریم 5 عناصر پر مشتمل ہے:
-
اصل وعدہ — مخصوص جدول سے لیا گیا: یہ کیا کرتا ہے، کیا نہیں کرتا، اور جب یہ ہو جاتا ہے تو کیسا لگتا ہے۔
-
پاس کرنے کا معیار –
done_looks_likeواضح بینچ مارکس کے ساتھ پیش کردہ فیلڈز -
فرق – پہلا موڑ ان پٹ بمقابلہ آخری باری آؤٹ پٹ، موڑ مکمل، کل ٹوکن، لوپ کی خلاف ورزی یا نہیں
-
ثبوت — سیشن میں لیجر کی تمام قطاریں: مرحلہ پاس/فیل، ٹوکن ڈیلٹا، عمل درآمد کا وقت
-
غیر حل شدہ مفروضے۔ — خلاف ورزی کرنے والی قطاروں اور ناکام گردشوں سے ماخوذ۔ جب یہ صاف ہو جائے تو اسے خالی چھوڑ دیں۔
جائزہ لینے والے کے اطمینان پر، ہم تصدیق کرتے ہیں کہ:
attestation = rs.attest(
session_id=result.session_id,
reviewer="daniel",
notes="Output matches spec. Approved."
)
print(attestation.frame_hash)
.attest() کو لکھیں۔ attestations میں میز ledger.db. کہ frame_hash ایک ہی سیشن کی توثیق کرنے والے جائزہ لینے والوں کے درمیان حتمی فریم شدہ ڈیٹا کا SHA-256۔ یہ شکریہ کی رسید ہے۔ اس سے ثابت ہوتا ہے کہ جائزہ لینے والے نے بالکل ٹھیک فریم کو پیش کردہ کے طور پر دیکھا، خلاصہ یا پیرا فریس نہیں۔
منظوری اس بات کی تصدیق کرتی ہے کہ عمل مکمل ہو گیا ہے۔ تصدیق اس بات کی تصدیق کرتی ہے کہ جائزہ لینے والے نے وعدے کے ساتھ آؤٹ پٹ کا موازنہ کیا ہے۔ ایک بار جب لوپ کسی ریگولیٹڈ ہستی کو چھوتا ہے، تو یہ ایک مختلف قانونی دستاویز ہوتی ہے۔
@dataclass(frozen=True)
class ReviewFrame:
session_id: str
original_promise: SpecResult
acceptance_criteria: str
diff: DiffResult
evidence: tuple # tuple[LedgerRow, ...]
unresolved_assumptions: tuple # tuple[str, ...]
created_at: str
ReviewFrame اسی وجہ سے اسے منجمد کیا گیا تھا۔ SpecResult – فریم ایک ثبوت ہے، مسودہ نہیں۔ evidence اور unresolved_assumptions فہرستیں ٹیپلز ہیں کیونکہ وہ ہیش ایبل نہیں ہیں اور فکسڈ ڈیٹا کلاسز کو ہیش ایبل فیلڈز کی ضرورت ہوتی ہے۔
جائزہ اسکرین کے ذریعے مکمل اختتام سے آخر تک بہاؤ مندرجہ ذیل ہے: examples/review_example.py ریپو سے۔ سیشن مکمل ہونے کے بعد چلتا ہے۔ یہ پانچ عناصر کا فریم پیش کرتا ہے، ثبوت کی درخواست کرتا ہے، اور اگر منظور ہو جائے تو ایک رسید بناتا ہے۔
لوپ آپ کی طرف دوڑتا ہے۔ ڈاؤن اسٹریم سسٹم کو اس وقت تک کچھ نہیں ملتا جب تک کوئی دستخط نہ کرے۔
مرحلہ 6: حقیقی دنیا کی مثالیں — SEO آڈیٹرز
پیٹرن صرف حقیقی مسائل کے لیے معنی رکھتے ہیں۔ یہ میرے SEO-ایجنٹ پروجیکٹ کے پیچھے وہی ایجنٹ فن تعمیر ہے۔
SEO آڈٹ کا ایک فطری بہاؤ ہوتا ہے: رینگنا، ٹوٹے ہوئے حصوں کو سرفیس کرنا، انہیں ٹھیک کرنا، اور ان کے دوبارہ انڈیکس ہونے کا انتظار کرنا۔ ایجنٹ کو مسلسل چلانے سے اس کا چکر تبدیل نہیں ہوتا ہے۔ آپ صرف اہم لمحات کے درمیان خالی جگہ پر ٹوکن جلاتے ہیں۔ لوپ میں جکڑے ہوئے کرون جاب ایک ایماندار فن تعمیر ہے۔
# examples/seo_audit_example.py
import requests
from bs4 import BeautifulSoup
import anthropic
from spec_writer import SpecWriter
from circuit_breaker import CircuitBreaker
from ledger import Ledger
from agent_loop import AgentLoop
def crawl_url(url: str) -> str:
response = requests.get(url, timeout=10)
soup = BeautifulSoup(response.text, "html.parser")
title = soup.find("title")
meta_desc = soup.find("meta", attrs={"name": "description"})
h1_tags = soup.find_all("h1")
return (
f"URL: {url}\n"
f"Title: {title.text if title else 'MISSING'}\n"
f"Meta description: "
f"{meta_desc['content'] if meta_desc else 'MISSING'}\n"
f"H1 count: {len(h1_tags)}\n"
f"H1 tags: {[h.text[:50] for h in h1_tags]}"
)
def run_seo_audit(url: str) -> None:
# Step 1: Define done before the loop starts
spec = SpecWriter(db_path="spec.db").run()
# Step 2: Initialise circuit breaker and ledger
breaker = CircuitBreaker(turn_limit=5, token_limit=15000)
ledger = Ledger(db_path="ledger.db")
client = anthropic.Anthropic()
# Step 3: Crawl the URL
site_data = crawl_url(url)
# Step 4: Run the loop
# AgentLoop catches CircuitBreakerError internally and returns
# LoopResult(success=False, breach_reason=...). Branch on the
# result — do NOT wrap loop.run() in try/except CircuitBreakerError.
loop = AgentLoop(spec, breaker, ledger, client)
result = loop.run(
f"Audit this page for SEO issues:\n\n{site_data}"
)
# Step 5: Print the ledger
print(f"\nResult: {'SUCCESS' if result.success else 'BREACH'}")
if not result.success:
print(f"Breach reason: {result.breach_reason}")
print(f"Turns: {result.turns} | Tokens: {result.total_tokens}")
print("\nAudit trail:")
for row in ledger.get_session(result.session_id):
status = "PASS" if row.pass_fail else "FAIL"
print(f" Turn {row.turn_count}: {status} | "
f"{row.token_delta} tokens | {row.execution_time_ms}ms")
if __name__ == "__main__":
import sys
run_seo_audit(sys.argv[1] if len(sys.argv) > 1 else "https://example.com")
چلائیں:
python examples/seo_audit_example.py https://yourdomain.com
مخصوص مصنف ایک پیغام دکھاتا ہے۔ لوپ چلتا ہے، حد سے تجاوز کرنے پر سرکٹ بریکر ٹرپ کرتا ہے، اور لیجر ہر گردش کو ریکارڈ کرتا ہے۔ آؤٹ پٹ آپ کے سامنے آتا ہے اور آپ فیصلہ کرتے ہیں کہ کیا ٹھیک کرنا ہے۔
لوپ آپ کی طرف دوڑتا ہے، جگہ نہیں۔
پلگ ایبل ایل ایل ایم کلائنٹ
لوپ ان تمام کلائنٹس پر کام کرتا ہے جو مطمئن ہیں: LLMClient پروٹوکول (بذریعہ ڈیفالٹ اینتھروپک)۔ ~20 لائن اڈاپٹر کے ساتھ اپنا اپنا لائیں۔
# agent_loop.py
from typing import Protocol, runtime_checkable
@runtime_checkable
class MessagesEndpoint(Protocol):
def create(self, *, model: str, max_tokens: int,
system: str, messages: list) -> object: ...
@runtime_checkable
class LLMClient(Protocol):
messages: MessagesEndpoint
messages یہ ایک مثالی خاصیت ہے (گھروں والی کلاس نہیں) کیونکہ اس طرح اصل اینتھروپک SDK اسے بے نقاب کرتا ہے۔ anthropic.Anthropic().messages.create(...). اسے نیسٹڈ کلاس کے طور پر ماڈلنگ کرنے کا مطلب ہے کہ اصل کلائنٹ پروٹوکول کو مطمئن نہیں کرتا ہے۔ کہ @runtime_checkable ڈیکوریٹر آپ کو مطابقت برقرار رکھنے کو یقینی بنانے کی اجازت دیتے ہیں۔ isinstance(client, LLMClient)ریپوزٹری کا ٹیسٹ سویٹ اس دعوے کو بالکل ٹھیک استعمال کرتا ہے۔ FakeClient ڈبل ٹیسٹ۔
ذیل میں اوپن اے آئی اڈاپٹر کی ایک مثال دی گئی ہے (صرف مثالی مقاصد کے لیے؛ پروڈکشن اڈاپٹر اسٹریمنگ، ٹولنگ، اور غلطی کی اقسام کا نقشہ بھی بناتا ہے):
# openai_adapter.py — illustrative pseudocode, not production-ready.
from openai import OpenAI as _OpenAI
class _MessagesAdapter:
def __init__(self, client):
self._client = client
def create(self, *, model, max_tokens, system, messages):
completion = self._client.chat.completions.create(
model=model,
max_tokens=max_tokens,
messages=[{"role": "system", "content": system}] + messages,
)
# Reshape OpenAI's response into the Anthropic-shaped surface
# AgentLoop reads: response.usage.{input,output}_tokens,
# response.content[0].text, response.stop_reason.
return _adapt_response(completion)
class OpenAIAdapter:
def __init__(self, api_key: str):
self._client = _OpenAI(api_key=api_key)
self.messages = _MessagesAdapter(self._client) # instance attr, not a nested class
اڈاپٹر پیٹرن واضح طور پر سکھانے کے قابل ہے. فراہم کنندہ APIs ظاہری شکل کا اشتراک نہیں کرتے ہیں۔ انسانیت کے پاؤں system اعلیٰ ترین سطح پر۔ OpenAI اسے ایک پیغام کی صف میں رکھتا ہے۔ اڈاپٹر شیم ~ 20 لائنیں لمبا ہے اور اسے کسی بھی چیز کو دوبارہ لکھے بغیر لوپ فراہم کنندہ کو اجناسٹک بنا دیتا ہے۔ براہ کرم اس کا حوالہ دیں۔ self.messages کو تفویض کیا جاتا ہے۔ __init__ لہذا یہ ہر اڈاپٹر مثال کی اصل خصوصیات ہیں اور اصل SDK جیسی ظاہری شکل ہے۔
ٹیسٹ چلائیں
python -m pytest tests/
لاگو علاقہ:
python -m coverage run --source=circuit_breaker,ledger,spec_writer,agent_loop,review_surface -m pytest tests/
python -m coverage report -m
80 ٹیسٹ، تمام 5 بنیادی ماڈیولز کی 100% کوریج۔ لوپ اس کے لیے چلتا ہے: FakeClient ڈبل ٹیسٹ کی وضاحت ان لائن tests/test_agent_loop.py. یہ مندرجہ ذیل کو مطمئن کرتا ہے: LLMClient بتھ ٹائپنگ کے ذریعے پروٹوکول: messages پر مقرر ہے selfتو client.messages.create(...) یہ ایک ہی شے کی طرف واپس جاتا ہے اور ہر ٹیسٹ کے منظر نامے کے لیے اسکرپٹ شدہ جوابات فراہم کرتا ہے۔ ذخیرہ کلون کریں اور اسے چلائیں۔ pytest دیکھیں کہ کیا تمام 80 ٹیسٹ نیٹ ورک کو چھوئے بغیر یا API کلید کی ضرورت کے بغیر پاس ہو جاتے ہیں۔
circuit_breaker.py 100% کوریج ہے اور کوئی بھی راستہ بغیر ٹیسٹ شدہ نہیں ہے۔ یہ مالیاتی حفاظت کا حصہ ہے۔ جو بھی راستہ اس سے گزرتا ہے اسے پھانسی دی جاتی ہے۔
آپ نے کیا بنایا
اس ٹیوٹوریل میں، ہم نے پانچ چھوٹے پرائمیٹوز بنائے ہیں، جن میں سے ہر ایک کو آزادانہ طور پر استعمال کیا جا سکتا ہے۔
| ماڈیول | کردار | سموچ |
|---|---|---|
spec_writer.py |
لوپ چلنے سے پہلے تین جوابات پر مجبور کرتا ہے۔ | 104 |
circuit_breaker.py |
موڑ اور ٹوکن پر سخت چھت | 41 |
ledger.py |
صرف SQLite آڈٹ ٹریل شامل کریں۔ | 113 |
agent_loop.py |
ایک لوپ جو دونوں کا احترام کرتا ہے۔ | 128 |
review_surface.py |
پانچ عناصر کے فریم کو جمع کریں اور انسانی گواہی کو ریکارڈ کریں۔ | 114 |
پیٹرن: اپ اسٹریم ڈسپلن حدود کی وضاحت کرتے ہیں۔ ڈاؤن اسٹریم نفاذ سرکٹ کو توڑ دیتا ہے۔ نہ ہی خود پولیس کو ماڈل پر بھروسہ کرتا ہے۔
ایک لوپ جو باہر نکلنے کی شرط کے بغیر چلتا ہے وہ خود مختار نہیں ہے۔ ادائیگی کی تقریب کا انتظار ہے۔
شروع کرنے سے پہلے، وضاحت کریں کہ تیار شدہ شکل کیسی ہوگی۔ یہ وہی ہے، اور یہ ہمیشہ ایسا ہی رہا ہے۔
اگلے اقدامات
ذخیرہ github.com/dannwaneri/production-safe-agent-loop پر ہے۔
اگر آپ مزید جانا چاہتے ہیں تو، تین قدرتی توسیعات ہیں:
1. تقسیم شدہ نظاموں میں گریجویشن
SQLite لیجرز الگ تھلگ ترتیب وار لوپس پر کام کرتے ہیں۔ جس لمحے آپ مشترکہ ریاست پر ایک سے زیادہ ایجنٹ چلاتے ہیں، آپ کو سیریلائزبل آئسولیشن کی ضرورت ہوتی ہے۔ اس کا مطلب یہ ہے کہ فلیٹ JSON کو کنکرنٹ رائٹ خود بخود ٹوٹ جاتی ہیں۔ README تین اہم موڑ کو دستاویز کرتا ہے جہاں سے ایک فلیٹ لیجر کو گریجویٹ ہونا ضروری ہے:
2. کرپٹوگرافک دستخط
تعمیل پیمانے کے نظام کے لیے جہاں لوپ کے چلنے پر کوئی آڈیٹر موجود نہیں تھا، اکیلے SQLite قطاریں کافی نہیں ہوں گی۔ ڈیٹا بیس ایڈمنسٹریٹر چلا سکتے ہیں: UPDATE سوال Ed25519 دستخط ہر لیجر کی قطار کو ایک رسید میں لپیٹتا ہے جو ثابت کرتا ہے کہ لاگ ان کو پھانسی کے بعد سے تبدیل نہیں کیا گیا ہے۔ لیکن یہ ایک مختلف ٹیوٹوریل ہے۔
کرون جاب کنکشن
SEO آڈٹ ایجنٹ کا دیانت دار فن تعمیر خود مختار 24/7 آپریشن نہیں ہے۔ ایک کرون جاب جو ایک شیڈول پر چلتی ہے، کسی بھی بدعنوانی کو ڈھونڈتی ہے، اور پھر اسے روک دیتی ہے۔ 0 3 * * 2 python examples/seo_audit_example.py https://yourdomain.com بس۔ لوپ آپ کی طرف دوڑتا ہے، جگہ نہیں۔
اگر آپ کو اپنے اسٹیک (سرکٹ بریکرز، آڈٹ ٹریلز، پروڈکشن سیفٹی ایجنٹ لوپس) کے لیے بنائے گئے اس فن تعمیر کی ضرورت ہے تو کچھ فری لانس کام کریں۔ dannwaneri.com/ai-agents/