চেইনের নির্ভরযোগ্যতা: এরর রেট, শুরুর ভুল, সংশোধন লুপ ও ফলব্যাক

import json, re
from collections import Counter
from promptlab import Client, estimate_tokens, render

# Re-defined compactly from the previous page.
ask = lambda client, text: client.complete([{"role": "user", "content": text}]).text
EXTRACT = """Read the support ticket. Reply with JSON only, with the keys order_id, issue, action, email.
issue is one of: damaged, late, wrong_item, other. action is one of: refund, replace, none.
Use null for anything the ticket does not state. Do not guess.
{examples}
### TICKET
{ticket}"""
DRAFT = """Write a polite reply of at most 60 words to a customer.
You may use only these facts:
order: {order_id}
problem: {issue}
our next step: {step}
Do not promise anything that is not listed."""
STEP = {"refund": "we will start your refund", "replace": "we will send a replacement",
        "none": "a colleague will review your case"}
ENUMS = {"issue": {"damaged", "late", "wrong_item", "other"}, "action": {"refund", "replace", "none"}}

def toy_extract(messages, **settings):
    """Stand-in. Faults: bad enum if order % 7 == 0, swapped digits if order % 6 == 0.
    Copies hints like "phrase" -> label from an examples block (a crude few-shot stand-in)."""
    head, ticket = messages[-1]["content"].split("### TICKET", 1)
    text = ticket.lower()
    order = re.search(r"order\s*#?\s*(\d{4,})", text)
    mail = re.search(r"[\w.]+@[\w.]+\w", text)
    issue = next((k for w, k in [("cracked", "damaged"), ("late", "late"), ("wrong", "wrong_item")] if w in text), "other")
    if issue == "other":
        issue = next((lab for phrase, lab in re.findall(r'"(.+?)" -> (\w+)', head) if phrase in text), "other")
    action = next((k for w, k in [("refund", "refund"), ("replace", "replace")] if w in text), "none")
    oid = order and order.group(1)
    if oid and int(oid) % 6 == 0:
        oid = oid[:-2] + oid[-1] + oid[-2]           # fault: still looks like a valid order number
    elif oid and int(oid) % 7 == 0:
        issue = "broken"                              # fault: value outside the allowed list
    return json.dumps({"order_id": oid, "issue": issue, "action": action, "email": mail and mail.group(0)})

def toy_draft(messages, **settings):
    """Stand-in. Faults: filler if order % 5 == 0, a promise if order % 4 == 3."""
    f = dict(l.split(": ", 1) for l in messages[-1]["content"].splitlines() if ": " in l)
    n = int(f["order"])
    text = f"Hello, thank you for writing about order {f['order']}. We are sorry about the {f['problem']} problem. Our next step: {f['our next step']}."
    if n % 5 == 0: text += " We value every customer and appreciate your patience while we look into this." * 3
    if n % 4 == 3: text += " We guarantee a result within 24 hours."
    return text

TRICKY = ["Order 41113 arrived smashed. Please refund. Aiko, aiko@example.com",
          "Order 41222 came in pieces. Please replace it. Omar, omar@example.com"]
EXAMPLES = '''Examples of unusual wording:
"smashed" -> damaged
"in pieces" -> damaged'''
for label, ex in [("no examples", ""), ("examples in the extract stage", EXAMPLES)]:
    ext = Client(toy_extract)
    got = [json.loads(ask(ext, render(EXTRACT, examples=ex, ticket=t)))["issue"]
           for t in TRICKY]
    print(f"{label:30} -> {got}")
print("extra size of the extract prompt:", estimate_tokens(EXAMPLES), "tokens (rough)")

no examples                    -> ['other', 'other']
examples in the extract stage  -> ['damaged', 'damaged']
extra size of the extract prompt: 18 tokens (rough)

def make_cases(n=24):
    """24 invented tickets; every 4th has a long thread, 2 have no order number."""
    words, cases = {"cracked": "damaged", "late": "late", "wrong": "wrong_item"}, []
    for i in range(n):
        word, order, action = list(words)[i % 3], (None if i % 12 == 11 else str(40000 + 137 * i)), ["refund", "replace"][i % 2]
        ticket = (f"Order {order} is {word}." if order else f"My parcel is {word}.") + f" Please {action}. Ana, ana{i}@example.com"
        if i % 4 == 0: ticket += "\n" + "\n".join(f"> message {k}: still waiting" for k in range(40))
        cases.append({"ticket": ticket, "gold": {"order_id": order, "issue": words[word], "action": action}})
    return cases
CASES = make_cases()

def rubric(reply, order_id, step):
    """Fixed rubric in plain Python: names of failed criteria ([] = pass)."""
    failed = []
    if len(reply.split()) > 60: failed.append("too_long")
    if order_id not in reply: failed.append("missing_order")
    if step not in reply: failed.append("missing_step")
    if re.search(r"guarantee|promise", reply.lower()): failed.append("promise")
    return failed

def check_record(text, ticket, grounded=False):
    try: rec = json.loads(text)
    except json.JSONDecodeError: return None, ["not JSON"]
    problems = [f"{k}: {rec.get(k)!r} not allowed" for k, ok in ENUMS.items() if rec.get(k) not in ok]
    if grounded:   # a copied value must really appear in the source text
        problems += [f"{k}: {rec[k]!r} not in ticket" for k in ("order_id", "email") if rec.get(k) and rec[k] not in ticket]
    return rec, problems

def run(ticket, ext, wr, grounded=False):
    raw = ask(ext, render(EXTRACT, examples="", ticket=ticket))
    rec, problems = check_record(raw, ticket, grounded)
    if problems: return {"status": "stopped", "why": "; ".join(problems)}
    if not rec["order_id"]: return {"status": "needs_info"}
    step = STEP[rec["action"]]
    prompt = render(DRAFT, order_id=rec["order_id"], issue=rec["issue"].replace("_", " "), step=step)
    return {"status": "drafted", "order": rec["order_id"], "step": step, "reply": ask(wr, prompt)}

def judge(case, res):
    """ok = right outcome; stopped = chain refused to continue; silent_wrong = wrong and nobody warned."""
    g = case["gold"]
    if g["order_id"] is None: return "ok" if res["status"] == "needs_info" else "silent_wrong"
    if res["status"] != "drafted": return "stopped"
    good = res["order"] == g["order_id"] and not rubric(res["reply"], g["order_id"], STEP[g["action"]])
    return "ok" if good else "silent_wrong"

def evaluate(fn):
    c = Counter(judge(case, fn(case["ticket"])) for case in CASES)
    return {k: c[k] for k in ("ok", "stopped", "silent_wrong")}

BIG = """Read the support ticket and write a polite reply of at most 60 words.
State the order number, the problem and our next step (refund, replacement or review by a colleague).
### TICKET
{ticket}"""

def toy_big(messages, **settings):
    """Stand-in for ONE large prompt: invents an order number if none is given, forgets it on long input."""
    text = messages[-1]["content"].split("### TICKET", 1)[1].lower()
    m = re.search(r"order\s*(\d{4,})", text)
    order = m.group(1) if m else "10000"
    issue = next((k for w, k in [("cracked", "damaged"), ("late", "late"), ("wrong", "wrong item")] if w in text), "other")
    action = next((k for w, k in [("refund", "refund"), ("replace", "replace")] if w in text), "none")
    ref = "your order" if len(text) > 600 else f"order {order}"
    reply = f"Hello, thank you for writing about {ref}. We are sorry about the {issue} problem. Our next step: {STEP[action]}."
    return reply

def run_big(ticket, client):
    reply = ask(client, render(BIG, ticket=ticket))
    m = re.search(r"order (\d+)", reply)
    return {"status": "drafted", "order": m and m.group(1), "reply": reply}

big = Client(toy_big)
ext, wr = Client(toy_extract), Client(toy_draft)
print("one large prompt :", evaluate(lambda t: run_big(t, big)), "| calls:", big.calls)
print("chain, 2 stages :", evaluate(lambda t: run(t, ext, wr)), "| calls:", ext.calls + wr.calls)

one large prompt : {'ok': 16, 'stopped': 0, 'silent_wrong': 8} | calls: 24
chain, 2 stages : {'ok': 12, 'stopped': 2, 'silent_wrong': 10} | calls: 44

ext, s1, s3 = Client(toy_extract), 0, 0
for case in CASES:                           # stage 1 alone: compare the record with the known answer
    g, text = case["gold"], ask(ext, render(EXTRACT, examples="", ticket=case["ticket"]))
    rec, problems = check_record(text, case["ticket"])
    s1 += not problems and rec["order_id"] == g["order_id"] and rec["issue"] == g["issue"]
    if g["order_id"]:                        # stage 3 alone: always feed it the CORRECT facts
        step = STEP[g["action"]]
        reply = toy_draft([{"role": "user", "content": render(DRAFT, order_id=g["order_id"], issue=g["issue"].replace("_", " "), step=step)}])
        s3 += not rubric(reply, g["order_id"], step)
with_order = sum(1 for c in CASES if c["gold"]["order_id"])   # stage 3 only runs when an order exists
p1, p3 = s1 / len(CASES), s3 / with_order
whole = evaluate(lambda t: run(t, Client(toy_extract), Client(toy_draft)))["ok"] / len(CASES)
print(f"extract stage alone: {p1:.2f}   draft stage alone: {p3:.2f}")
print(f"product of the stage rates: {p1 * p3:.2f}   measured on the whole chain: {whole:.2f}")

extract stage alone: 0.75   draft stage alone: 0.64
product of the stage rates: 0.48   measured on the whole chain: 0.50

for grounded in (False, True):
    ext, wr = Client(toy_extract), Client(toy_draft)
    print("grounding check:", grounded, evaluate(lambda t: run(t, ext, wr, grounded)))
case = next(c for c in CASES if c["gold"]["order_id"] == "40548")
res = run(case["ticket"], Client(toy_extract), Client(toy_draft), grounded=True)
print(case["gold"]["order_id"], "->", res["status"], "|", res["why"])

grounding check: False {'ok': 12, 'stopped': 2, 'silent_wrong': 10}
grounding check: True {'ok': 12, 'stopped': 6, 'silent_wrong': 6}
40548 -> stopped | order_id: '40584' not in ticket

Improve this reply.

Fix ONLY the defects listed below. Change nothing else; keep every other word.
DEFECTS: too_long, promise
REPLY: <the draft reply, pasted here>

REVISE = """Fix ONLY the defects listed below. Change nothing else; keep every other word.
DEFECTS: {defects}
REPLY: {reply}"""
GENERIC = "Improve this reply.\nREPLY: {reply}"

def toy_reviser(messages, **settings):
    """Stand-in. Given a defect list it removes the offending sentences; told only to 'improve' it rewrites everything."""
    text = messages[-1]["content"]
    reply = text.split("REPLY: ", 1)[1]
    if "DEFECTS:" not in text:
        return "Thank you for your message. We are sorry for the trouble and will reply soon."
    defects = text.split("DEFECTS: ", 1)[1].split("\n")[0].split(", ")
    drop = [w for d, w in [("too_long", "We value"), ("promise", "guarantee")] if d in defects]
    return " ".join(s for s in re.split(r"(?<=\.)\s+", reply) if not any(w in s for w in drop))

def improve(res, reviser, targeted=True, gate=True, attempts=1):
    reply, used = res["reply"], 0
    while used < attempts and (failed := rubric(reply, res["order"], res["step"])):
        prompt = render(REVISE, defects=", ".join(failed), reply=reply) if targeted else render(GENERIC, reply=reply)
        candidate, used = ask(reviser, prompt), used + 1
        new = rubric(candidate, res["order"], res["step"])
        if not gate or (len(new) < len(failed) and set(new) <= set(failed)):
            reply = candidate
    return {**res, "reply": reply}

def with_revision(left, **opts):
    ext, wr, rev = Client(toy_extract), Client(toy_draft), Client(toy_reviser)
    def fn(ticket):
        res = run(ticket, ext, wr, grounded=True)
        if res["status"] != "drafted": return res
        res = improve(res, rev, **opts)
        left.append(len(rubric(res["reply"], res["order"], res["step"])))
        return res
    return fn

for name, opts in [("no revision", dict(attempts=0)), ("generic, always keep", dict(targeted=False, gate=False)),
                   ("generic, rubric gate", dict(targeted=False)), ("defect list + gate", {})]:
    left = []
    print(f"{name:21}", evaluate(with_revision(left, **opts)), "failed criteria left:", sum(left))

no revision           {'ok': 12, 'stopped': 6, 'silent_wrong': 6} failed criteria left: 7
generic, always keep  {'ok': 12, 'stopped': 6, 'silent_wrong': 6} failed criteria left: 12
generic, rubric gate  {'ok': 12, 'stopped': 6, 'silent_wrong': 6} failed criteria left: 7
defect list + gate    {'ok': 18, 'stopped': 6, 'silent_wrong': 0} failed criteria left: 0

def toy_critic(messages, **settings):
    # stand-in critic, swayed by a polite tone
    return "OK, polite and clear." if "sorry" in messages[-1]["content"].lower() else "NEEDS WORK, no apology."

ext, wr, critic = Client(toy_extract), Client(toy_draft), Client(toy_critic)
table = Counter()
for case in CASES:
    res = run(case["ticket"], ext, wr, grounded=True)
    if res["status"] != "drafted": continue
    says = ask(critic, "Is this reply good? Answer OK or NEEDS WORK.\nREPLY: " + res["reply"]).split(",")[0]
    table[(says, "rubric pass" if not rubric(res["reply"], res["order"], res["step"]) else "rubric fail")] += 1
for key in sorted(table): print(key, table[key])

('OK', 'rubric fail') 6
('OK', 'rubric pass') 10

def toy_stuck(messages, **settings):
    return messages[-1]["content"].split("REPLY: ", 1)[1]        # stand-in: "revises" but changes nothing

def finalize(res, reviser, max_attempts=2):
    out = improve(res, reviser, attempts=max_attempts)
    if not rubric(out["reply"], res["order"], res["step"]):
        return {**out, "status": "drafted"}
    safe = f"Hello, thank you for writing about order {res['order']}. Our next step: {STEP['none']}."
    assert not rubric(safe, res["order"], STEP["none"])           # the fallback itself must pass the rubric
    return {**out, "reply": safe, "step": STEP["none"], "status": "fallback", "needs_person": True}

bad = {"order": "42055", "step": STEP["refund"], "reply": "Order 42055. Our next step: we will start your refund. We guarantee it."}
stuck = Client(toy_stuck)
out = finalize(bad, stuck)
print(out["status"], "after", stuck.calls, "revision calls:", out["reply"])

fallback after 2 revision calls: Hello, thank you for writing about order 42055. Our next step: a colleague will review your case.

big, e2, w2 = Client(toy_big), Client(toy_extract), Client(toy_draft)
e3, w3, rev = Client(toy_extract), Client(toy_draft), Client(toy_reviser)
def full(ticket):
    res = run(ticket, e3, w3, grounded=True)
    return finalize(res, rev) if res["status"] == "drafted" else res

print("one large prompt   ", evaluate(lambda t: run_big(t, big)), "calls:", big.calls)
print("chain, no revision ", evaluate(lambda t: run(t, e2, w2, grounded=True)), "calls:", e2.calls + w2.calls)
print("full chain         ", evaluate(full), "calls:", e3.calls + w3.calls + rev.calls)

one large prompt    {'ok': 16, 'stopped': 0, 'silent_wrong': 8} calls: 24
chain, no revision  {'ok': 12, 'stopped': 6, 'silent_wrong': 6} calls: 40
full chain          {'ok': 18, 'stopped': 6, 'silent_wrong': 0} calls: 46

North: 120 units at 15
South: 80 units at 20
East: 45 units at 30

NOTES = "North: 120 units at 15\nSouth: 80 units at 20\nEast: 45 units at 30"
parse = lambda notes: [{"region": a, "units": int(b), "price": int(c)} for a, b, c in re.findall(r"(\w+): (\d+) units at (\d+)", notes)]

def compute(rows, notes):
    """Stage 2 in plain Python: every row must match the notes exactly; then add up the revenue."""
    for r in rows:
        if f"{r['region']}: {r['units']} units at {r['price']}" not in notes:
            raise ValueError(f"{r['region']}: row does not match the notes")
    revenue = {r["region"]: r["units"] * r["price"] for r in rows}
    return {**revenue, "Total": sum(revenue.values())}

rows = parse(NOTES)
print(compute(rows, NOTES))
rows[2]["units"] = 54                      # the extractor swapped two digits
try: compute(rows, NOTES)
except ValueError as exc: print("stopped:", exc)

{'North': 1800, 'South': 1600, 'East': 1350, 'Total': 4750}
stopped: East: row does not match the notes

toy_rows = lambda m, **s: json.dumps(parse(m[-1]["content"]))                       # stand-in extractor
toy_swap = lambda m, **s: toy_rows(m).replace('"units": 45', '"units": 54')           # stand-in: swapped digits

def toy_writer(messages, **settings):        # stand-in: adds an unsupported claim
    facts = re.findall(r"(\w+): (\d+)", messages[-1]["content"])
    return "Revenue: " + ", ".join(f"{k} {v}" for k, v in facts) + ". Sales show strong growth."

def report_rubric(text, figures):
    failed = ["missing_" + k for k, v in figures.items() if str(v) not in text]
    return failed + (["unsupported_claim"] if "growth" in text.lower() else [])

def report(notes, extractor, writer, reviser):
    rows = json.loads(ask(extractor, "Extract rows.\n" + notes))  # 1: model
    try: figures = compute(rows, notes)                                                                    # 2: code
    except ValueError as exc: return "stopped", str(exc)
    facts = "\n".join(f"{k}: {v}" for k, v in figures.items())
    text = ask(writer, "Write a report from:\n" + facts)         # 3: model
    failed = report_rubric(text, figures)
    if failed:                                                   # one gated revision, then the fallback
        cand = ask(reviser, f"DEFECTS: {', '.join(failed)}\nREPLY: {text}")
        if not report_rubric(cand, figures): return "revised", cand
        return "fallback", "Revenue by region (table only): " + facts.replace("\n", "; ")
    return "ok", text

def toy_cut(messages, **settings):           # stand-in reviser
    return messages[-1]["content"].split("REPLY: ", 1)[1].replace(" Sales show strong growth.", "")

for name, ex, rev in [("normal", toy_rows, toy_cut), ("digit swap", toy_swap, toy_cut), ("stuck reviser", toy_rows, toy_stuck)]:
    print(name, "->", *report(NOTES, Client(ex), Client(toy_writer), Client(rev)))

normal -> revised Revenue: North 1800, South 1600, East 1350, Total 4750.
digit swap -> stopped East: row does not match the notes
stuck reviser -> fallback Revenue by region (table only): North: 1800; South: 1600; East: 1350; Total: 4750