1zero24's picture
Upload latest hosted langfuse retest snapshot
290ff9e verified
|
Raw
History Blame Contribute Delete
15 kB

Ergo-Agentic: Architecture

Tech Stack

  • Orchestration: LangGraph (Python)
  • AI Models: Multi-model (Claude, GPT-4V, Gemini β€” configurable)
  • Rules Engine: Static JSON datasources (no database for rules)
  • Report Generation: Deterministic code (no AI)

High-Level Pipeline

Input (images + metadata)
  β”‚
  β–Ό
inspect_workspace_image ──── per image, parallel
cv_extract ──────────────── per image, parallel
  β”‚
  β–Ό
build_routing_manifest ───── deterministic join of semantic manifest + CV + metadata
  β”‚
  β–Ό
choose_applicable_parameters ── deterministic
  β”‚
  β–Ό
assess_body_and_chair / assess_desk_and_arms / assess_screens
  ──── per relevant image Γ— per model, parallel
  β”‚
  β–Ό
aggregate_findings_by_assessment ── deterministic, assessment-scoped
  β”‚
  β–Ό
confidence_gate ─────────── deterministic accept / targeted-review split
  β”‚
  β–Ό
targeted_review_evidence ── model review over weak/conflicting request batches
  β”‚
  β–Ό
global_consistency_audit ── deterministic flags only, does not change outcomes
  β”‚
  β–Ό
build_final_assessment_matrix ── deterministic

Pipeline Nodes

1. inspect_workspace_image

Type: AI (vision LLM) Runs: Once per uploaded image, in parallel Input: Single image Output: Workspace semantic manifest

WorkspaceSemanticManifest:
  image_id: str
  work_location: str | null        # homeOffice, couch, bed, diningTable, floorSitting
  person_visible: bool
  body_coverage: str               # none | partial | posture_coverage_sufficient
  visible_body_regions: list[str]  # head, torso, hips, legs, hands, feet, etc.
  posture_context_hint: str        # sitting | standing | mixed | unknown
  chair_visible: bool
  chair_type: str | null           # office chair, dining chair, stool, etc.
  desk_visible: bool
  feet_visible: bool
  screens: list                    # type, coarse_layout_position, display_visible_state
  keyboard_mouse_visible: bool
  detected_accessories: list       # ergonomic objects: footrest, laptop stand,
                                   # sit-stand converter, Wacom tablet,
                                   # document reader, wrist rest, etc.
  notes: str                       # anything unusual

Purpose: Understand what each image contains so downstream routing can avoid images that cannot answer certain questions. This node does not judge posture quality, screen arrangement, primary screen intent, or standing desk presence from a standing person. It only reports visible scene contents and coverage.

2. build_routing_manifest

Type: Deterministic code (no AI) Input: WorkspaceSemanticManifest[] + cv_results[] + user metadata Output: RoutingManifest + scene_config

RoutingManifest is the planner's truth source. It merges semantic coverage, CV availability, pose/detection hints, and metadata into stable routing facts:

RoutingManifest:
  body_and_chair_coverage: FocusGroupCoverage
  desk_and_arms_coverage: FocusGroupCoverage
  screens_coverage: FocusGroupCoverage
  standing_work_possible: bool
  standing_desk_equipment_visible: bool
  posture_context: sitting | standing | mixed
  image_focus_coverage: dict[focus_group, list[image_id]]
  warnings: list[RoutingWarning]
  evidence_sources: list[EvidenceSourceRef]

Standing desk equipment is true only when sit/stand equipment is visible or metadata says it exists. A standing person can set standing_work_possible, but does not enable standing desk equipment parameters by itself.

3. choose_applicable_parameters

Type: Deterministic code (no AI) Input: RoutingManifest + common-assessment-parameters.json Output: Execution plan

ExecutionPlan:
  assessable_parameters: list[str]          # human-readable labels with image coverage
  assessable_parameter_ids: list[str]       # parameter IDs used as stable join keys
  assessable_parameter_details: list[dict]  # id, label, group, question for traces
  skipped_parameters: list[str]             # no image can answer these
  skipped_parameter_details: list[dict]     # id, label, group, question for traces
  focus_group_assignments:
    "body_and_chair": list[image_id]        # which images to send
    "desk_and_arms": list[image_id]
    "screens": list[image_id]
  scene_config:
    screen_count: int
    has_standing_desk: bool
    work_location: str

Logic:

  • If routing has no posture_coverage_sufficient, skip body/chair parameters
  • If no image has feet_visible, skip feet grounding
  • If screen_count == 1, skip screen arrangement parameters
  • If standing_desk_equipment_visible == false, skip standing desk parameters
  • Screen type logic: single monitor β†’ monitor params; single laptop β†’ laptop params; mixed/multiple β†’ include arrangement rubrics, but the screen pass still judges the outcome
  • Filters common-assessment-parameters.json to only relevant parameters

4. assess_body_and_chair / assess_desk_and_arms / assess_screens

Type: AI (vision LLM) Runs: Per image Γ— per model Γ— per focus group (all parallel) Input: Image(s) + parameter rubric for the focus group

Three focus groups, each assessing 3-4 parameters:

Pass 1: Body & Chair

Visual focus: The person's body and chair Parameters:

  • Back posture (slouching / sitting back / perched / leaning forward)
  • Sitting height (hips vs knees)
  • Seat pan depth (gap behind knee)
  • Feet grounding (feet flat, dangling, etc.)

Pass 2: Desk & Arms

Visual focus: Desk surface and arm positioning Parameters:

  • Desk height relative to elbow (sitting)
  • Standing desk height relative to elbow (if applicable)
  • Armrest height relative to elbow
  • Keyboard & mouse placement (distance from desk edge)

Pass 3: Screens

Visual focus: Monitor/laptop area Parameters:

  • Screen/laptop height relative to eye level
  • Screen/laptop distance from user
  • Screen arrangement β€” only if multiple screens (is person centered on primary?)

Screen logic: Single monitor β†’ assess monitor only. Single laptop β†’ assess laptop only. Mixed setup β†’ monitor is primary. Agents simply assess what they see; mapping to specific outcome key variants (twoMulti-, twoScreens-, etc.) happens in the aggregator.

Output per parameter:

Observation:
  parameter_id: str
  outcome: str | null              # the option key, or null if not determinable
  visibility: clear | partial | not_visible
  confidence: "high" | "medium" | "low"
  evidence_note: str               # what the model actually observed (specific visual cues)

Prompt design:

  • Each pass receives ONLY its 3-4 parameters plus a visual cue guide describing what to look at
  • The parameter rubric includes the reference images (optionImage) as visual anchors
  • The model is asked: "Which of these options best matches what you see?"
  • Evidence notes must reference specific observable features, not subjective impressions
  • Explicitly told: "If you cannot clearly determine this parameter from the image, return null"

See agent-orchestration.md for the full visual cue guide per focus group.

5. Assessment-Scoped Worst-Case Aggregator

Type: Deterministic code (no AI) Input: All observations from all vision passes Output: Aggregated results + conflicts

This is still one deterministic reducer node in the graph, but it does not blend unrelated tasks together. It first buckets observations by assessment_id and then applies worst-case aggregation inside each bucket.

Logic per assessment:

  1. Across images: If image 1 shows chair-posture-sittingBack but image 2 shows chair-posture-slouching, take the worst outcome (highest risk). This follows the "worst observed" rule.

  2. Across models: Collect votes. If all 3 models agree, high confidence in the result. If they disagree, flag as a conflict.

  3. Risk ranking: Uses issue-based-outcomes.json to rank outcomes by risk level (high > medium > low > null/good habit) and postureScore (lower score = worse).

AggregatedResult:
  parameter_id: str
  worst_outcome: str
  model_votes: dict[str, str]      # model_name -> outcome
  agreement: bool                  # all models agree?
  confidence: str                  # derived from vote agreement
  
Conflict:
  parameter_id: str
  disagreements: dict[str, str]    # model_name -> outcome
  images_analyzed: list[str]

6. Layered Review

Type: deterministic gate + targeted AI review + deterministic audit Runs: Gate once, targeted review only for weak/conflicting findings, audit once Input: Candidate findings, conflicts, relevant images, routing summary Output: Final confirmed outcome list plus audit flags

Purpose:

  • Accept strong, uncontested findings without another model call
  • Resolve conflicts or weak evidence using only relevant images and rubrics
  • Batch related weak findings into one review request when they share focus group, posture context, and image subset
  • Catch any obvious errors (e.g., model said "slouching" but person is clearly standing)
  • Check matrix consistency without directly overriding outcomes
  • Produce the definitive list of outcome keys

Output:

ReviewResult:
  final_outcomes: list[str]        # confirmed outcome keys
  overrides: list[Override]        # where review agent changed the aggregated result
  skipped_parameters: list[str]    # insufficient evidence
  global_audit_flags: list[GlobalAuditFlag]

7. Outcome Matrix Builder

Type: Deterministic code (no AI) Input: Final outcomes + evidence trail + scene config Output: Outcome matrix (parameter x outcome x evidence)

The primary output of the agentic pipeline. A structured matrix of which parameters were assessed, which outcome(s) were selected, evidence summaries, and review decisions. This is the core deliverable β€” everything downstream is deterministic transformation.

8. Report Builder (optional, separate concern)

Type: Deterministic code (no AI) Input: Outcome matrix + all datasource JSONs + metadata Output: Complete report JSON (matching report-sample.json structure)

Can run inside or outside the LangGraph pipeline. The agentic work is done once the outcome matrix is built.

Logic:

  1. For each outcome in the matrix:
    • Look up in issue-based-outcomes.json β†’ get isGoodHabit, riskLevel, currentHabit, recommendation, postureScore
    • Look up in body-based-outcomes.json β†’ get affected body parts, conditions, body-part-specific recommendations
  2. Build goodHabits list from outcomes where isGoodHabit == true
  3. Build issueBasedReport from outcomes where isGoodHabit == false
  4. Build potentialRiskPart by grouping issues per body part, taking worst condition per body part
  5. Build actionPlans from recommendations, ordered by priority (points)
  6. Match mergedProducts from product datasource based on outcomes
  7. Calculate ergoPostureScore from posture scores

Anti-Hallucination Strategy

Layer Mechanism
Image Analyzer Produces a visibility/scene manifest only; no ergonomic judgments
Routing Manifest Deterministically maps semantic/CV/metadata facts to focus-group coverage
Focused passes 3-4 params per call, not 12+; model stays grounded on one visual region
Visual cue guides Each pass includes specific visual cues (what to look at, what angles to check)
Reference images Option images from rubric serve as visual anchors ("does it look like A or B?")
Observable evidence Models must cite specific visual features, not subjective impressions
Null allowed Models can say "not determinable" β€” no pressure to fill every field
Multi-model voting Single-model hallucination caught by disagreement with other models
Layered review Strong findings are accepted deterministically; only weak/conflicting findings get targeted image review

Model Routing

Models are configurable per role:

model_config:
  image_analyzer: "claude-sonnet"
  vision_passes:
    - "claude-sonnet"
    - "gpt-4o"
    - "gemini-pro-vision"
  review_agent: "claude-opus"

This is designed for experimentation β€” swap models, compare accuracy, find the best combination.

LangGraph State Schema

class ErgoState(TypedDict):
    # Input
    images: list[ImageInput]
    metadata: dict
    
    # Image Analyzer output
    image_manifests: list[WorkspaceSemanticManifest]
    cv_results: list[ImageCVResult]
    routing_manifest: RoutingManifest
    work_location: str
    scene_config: SceneConfig
    
    # Planner output
    assessable_parameters: list[str]          # human-readable labels
    assessable_parameter_ids: list[str]       # stable IDs for internal joins
    assessable_parameter_details: list[dict]
    skipped_parameters: list[str]             # stable IDs
    skipped_parameter_details: list[dict]
    execution_plan: ExecutionPlan
    
    # Vision pass outputs
    observations: list[Observation]
    
    # Aggregator output
    aggregated_findings: dict[str, AggregatedFinding]
    candidate_findings: dict[str, CandidateFinding]
    conflicts: list[Conflict]
    
    # Review output
    review_requests: list[ReviewRequest]
    review_decisions: list[ReviewDecision]
    global_audit_flags: list[GlobalAuditFlag]
    final_outcomes: list[str]
    evidence_trail: dict[str, Evidence]
    
    # Primary output
    outcome_matrix: OutcomeMatrix    # parameter Γ— outcome Γ— evidence
    
    # Optional report
    report: dict | None

Scalability Considerations

  • Vision passes are embarrassingly parallel: images Γ— models Γ— focus_groups can all run concurrently
  • For 3 images, 3 models, 3 focus groups = 27 parallel calls + 3 image analyzer calls + 1 review call = 31 total LLM calls
  • Report building is sub-second (JSON lookups)
  • New parameters/groups only require datasource updates; pipeline auto-adapts via Parameter Planner

See Also

For detailed implementation specs, see docs/architecture/: