Normalize the JSON schema and enforce deterministic serialization at the tool boundary.
Step‑by‑step mitigation
1. Canonical schema – define a JSON Schema (e.g., agent_output_schema.json) and store it in version‑controlled schemas/.
2. Deterministic serializer – wrap every LLM call:
import json, hashlib
def serialize_llm_output(obj):
# sort keys, compact separators, no whitespace
txt = json.dumps(obj, sort_keys=True, separators=(',', ':'), ensure_ascii=False)
# optional integrity tag
checksum = hashlib.sha256(txt.encode()).hexdigest()[:8]
return txt, checksum3. Validation layer – after the LLM returns a string, run jsonschema.validate against the stored schema; reject or retry on failure.
4. Framework hooks – in LangChain set output_parser=JsonOutputParser(schema_path='schemas/agent_output_schema.json'); in CrewAI use crew.output_schema; in AutoGen set agent.set_output_schema(...).
5. Post‑processing microservice – expose /normalize endpoint (FastAPI) that receives raw LLM text, applies step 2‑3, and returns the canonical JSON plus checksum.
6. Drift monitoring – compute a daily SHA‑256 of all stored outputs; alert if the distribution of checksums changes beyond a 2 % threshold.
Tool comparison (text table)
| Framework | Built‑in schema support | Deterministic flag | Retry on validation |
|-----------|------------------------|--------------------|----------------------|
| LangChain | JsonOutputParser | ensure_ascii=False | ✅ |
| CrewAI | output_schema param | none (custom wrapper) | ✅ |
| AutoGen | set_output_schema | none | ✅ |
Quick checklist
- [ ] Schema file committed?
- [ ] Serializer uses sort_keys=True and compact separators?
- [ ] Validation step present?
- [ ] Framework hook configured?
- [ ] Monitoring alert threshold set?