Date: 2023-12-20 Topic: System architecture for automated ad compliance checking
Today I designed the architecture for an intelligent advertisement review system that combines LLMs with specialized tools for compliance checking.
The system uses a layered design:
Access Layer → File upload, validation, response formatting
Business Layer → Image analysis, text recognition, compliance check
Model Layer → VL model, LLM, image processing, caching
Infrastructure → Logging, monitoring, configuration, resources
async def identify_scene(self, text_data: str, image_info: Dict) -> Dict:
# Use VL model for image analysis
image_result = await self._execute_vl_chat({
"file": image_info.get("file", ""),
"prompt": "Analyze product type and main content"
})
# Check for health food indicators
health_food_keywords = ["health food", "health function", "nutritional"]
if any(kw in text_data for kw in health_food_keywords):
return {
"scene_type": "advertisement",
"product_type": "health food",
"specific_scene": "health food ad review",
"confidence": 0.95
}
return {"scene_type": "unknown", "confidence": 0}async def _execute_react_loop(self, prompt, max_steps=5, retry_count=3):
current_prompt = prompt
history = []
for step in range(max_steps):
for attempt in range(retry_count):
try:
response = await llm_chat(current_prompt)
# Check for final answer
final = self._extract_final_answer(response)
if final:
final["reasoning_process"] = history
return final
# Parse and execute tool call
tool_name, args = await self._parse_latest_plugin_call(response)
if tool_name:
result = await self._execute_tool(tool_name, args)
history.append({
"step": step + 1,
"thought": self._extract_thought(response),
"action": f"{tool_name}: {json.dumps(args)}",
"result": result
})
current_prompt += f"\nObservation: {json.dumps(result)}"
break
except Exception as e:
if attempt < retry_count - 1:
await asyncio.sleep(1)
continue
return {"error": "No valid result", "reasoning_process": history}async def parse_raw_rule(self, raw_text: str, max_retries=3):
for attempt in range(max_retries):
try:
prompt = self._prepare_rule_prompt(raw_text)
response = await llm_chat(prompt, temperature=0.7)
json_text = self._clean_json_string(response)
parsed_rules = json.loads(json_text)
if self._validate_rules(parsed_rules):
return parsed_rules
except Exception as e:
logger.error(f"Attempt {attempt + 1} failed: {e}")
return {"error": "Rule parsing failed", "rules": []}For health food ad review:
- Scenario Recognition:
{"product_type": "health food", "confidence": 0.95} - Rule Matching: Verify approval numbers, required disclaimers
- Review Results:
{"is_compliant": true, "suggestions": [...]}
The system architecture balances flexibility with structure. The ReAct controller allows dynamic reasoning, while the rule engine provides consistency.
Key design decisions:
- Retry mechanisms at every level - LLM calls are unreliable
- History tracking for auditability - every decision is logged
- Caching for performance - same analysis shouldn't repeat
The challenge was balancing LLM "creativity" with regulatory precision. Temperature tuning and validation layers help, but there's inherent uncertainty.
- Prompt chaining patterns
- LLM output validation strategies
- Regulatory compliance APIs
- Multi-model orchestration