2.2 - End-to-End Architecture: Input, Processing, Output and Feedback Loops

Build exercise: Build a validated pipeline with a closed feedback loop

Intermediate · 90 minutes

You'll practice:

• Map a use case onto the seven stages and name the failure each stage catches
• Implement layered output validation: stop reason, schema, business rules
• Design a bounded repair-then-escalate failure policy
• Capture per-request telemetry that supports diagnosis
• Turn logged failures into a regression gate

1. Pick a use case (invoice extraction, ticket triage, contract clause tagging) and write a table of the seven stages: what your system does at each one and the specific failure it is there to catch.

   The exam skill is locating the right stage for a control. Writing the table forces you to assign every requirement (schema, permissions, idempotency, audit) to a deterministic stage or to an explicit model behaviour.

   You should see: A seven-row table where no row says only 'call Claude', and where at least three controls are deterministic code rather than prompt text.

   Hints

   1. For each requirement you must always meet, could a sentence in a prompt actually guarantee it?

   2. List the hard requirements first (never act twice, never exceed the limit, always valid shape) and assign each to a code stage; then list soft qualities (tone, completeness) and assign them to evals or review.

   3. Invoice example: intake rejects files over a size limit and duplicate invoice IDs; context puts the extraction instructions first and the document last; call uses structured outputs; validation checks total equals sum of lines and vendor exists; delivery writes idempotently by invoice ID; feedback logs failures to failures.jsonl.

2. Implement the model call with a JSON schema through output_config.format, and reject any response whose stop_reason is not end_turn before parsing.

   Stop reason is the cheapest validation layer. A truncated or refused response must never be parsed as complete.

   You should see: A function that returns parsed data for a normal call and raises a clear error for max_tokens or refusal responses.

   Hints

   1. Which stop reasons mean the text you are about to parse might be incomplete or absent?

   2. Check response.stop_reason before json.loads, and keep the schema strict with additionalProperties false.

   3. import anthropic, json
      client = anthropic.Anthropic()
      MODEL = 'claude-sonnet-5'  # check the models overview page for current IDs
      SCHEMA = {'type': 'object', 'properties': {'vendor': {'type': 'string'}, 'lines': {'type': 'array', 'items': {'type': 'object', 'properties': {'desc': {'type': 'string'}, 'amount': {'type': 'number'}}, 'required': ['desc', 'amount'], 'additionalProperties': False}}, 'total': {'type': 'number'}}, 'required': ['vendor', 'lines', 'total'], 'additionalProperties': False}
      def extract(text):
          r = client.messages.create(model=MODEL, max_tokens=1024, messages=[{'role': 'user', 'content': text}], output_config={'format': {'type': 'json_schema', 'schema': SCHEMA}})
          if r.stop_reason != 'end_turn':
              raise RuntimeError('unusable response: ' + str(r.stop_reason))
          text = next(b.text for b in r.content if b.type == 'text')
          return json.loads(text), r

3. Write a deterministic business-rule validator and a bounded repair loop: one repair attempt that feeds the specific validation error back, then return a needs_review result instead of bad data.

   Cross-field rules cannot live in the schema. A bounded repair with the concrete error is a cheap first recovery, and the escalation path prevents both silent bad output and infinite retry.

   You should see: Clean invoices return status ok; an invoice you deliberately corrupt returns needs_review with the error list after at most two model calls.

   Hints

   1. What is the rule the schema cannot express, and what should happen when the repair also fails?

   2. Implement check(d) returning a list of error strings, loop for max_repairs + 1 attempts appending the errors to the prompt, and return needs_review at the end.

   3. def check(d):
          errs = []
          if abs(sum(l['amount'] for l in d['lines']) - d['total']) > 0.01:
              errs.append('total does not equal the sum of line amounts')
          return errs
      def process(text, max_repairs=1):
          prompt = text
          for _ in range(max_repairs + 1):
              d, r = extract(prompt)
              errs = check(d)
              if not errs:
                  return {'status': 'ok', 'data': d}
              prompt = text + '\n\nYour previous answer failed validation: ' + '; '.join(errs) + '. Re-extract carefully.'
          return {'status': 'needs_review', 'data': d, 'errors': errs}

4. Add a per-request log record with request ID, prompt version, model, input and output token counts, latency, stop reason, validation outcome and final status, appended as one JSON line.

   Without this record you cannot diagnose a failure, attribute cost, or build eval cases from production. It is the raw material of the feedback loop.

   You should see: A requests.jsonl file with one line per call, including the request ID you could quote to support.

   Hints

   1. If someone reports a bad output tomorrow, what would you need to reconstruct what happened?

   2. The Python SDK exposes response._request_id and response.usage; add timing around the call and write a dict per attempt.

   3. import time
      def logged_extract(text, prompt_version='v1'):
          t0 = time.time()
          d, r = extract(text)
          rec = {'request_id': r._request_id, 'prompt_version': prompt_version, 'model': MODEL, 'input_tokens': r.usage.input_tokens, 'output_tokens': r.usage.output_tokens, 'latency_s': round(time.time() - t0, 2), 'stop_reason': r.stop_reason}
          with open('requests.jsonl', 'a') as f:
              f.write(json.dumps(rec) + '\n')
          return d, r

5. Close the loop: write each needs_review case to failures.jsonl with its input and the reason, then write a regression script that replays failures.jsonl and fails the build if more than an agreed number still end in needs_review.

   A feedback loop is closed when failures become tests that gate the next change. This is the difference between monitoring and improvement.

   You should see: After you change the prompt, running the regression script reports how many previously failing cases now pass, and exits non-zero when the gate is not met.

   Hints

   1. What happens to a failure today after the reviewer fixes it by hand?

   2. Store the input with a 'reason' field, then replay all stored inputs through process() and count non-ok results against a threshold you choose in advance.

   3. import sys
      ALLOWED = 0  # agreed in advance
      cases = [json.loads(l) for l in open('failures.jsonl')]
      bad = [c for c in cases if process(c['input'])['status'] != 'ok']
      print(f'{len(cases) - len(bad)}/{len(cases)} previously failing cases now pass')
      sys.exit(1 if len(bad) > ALLOWED else 0)