Designing Streaming Decision Agents for Dynamic Environments

How to Design a Streaming Decision Agent with Partial Reasoning, Online Replanning, and Reactive Mid-Execution Adaptation in Dynamic Environments

The Streaming Decision Agent implements a dynamic grid world with moving obstacles and a shifting goal to test real-time adaptability. It utilizes an online A* planner in a receding-horizon loop to commit to near-term moves while maintaining a 5000-node expansion budget.

Why This Matters

Traditional static planners fail in non-stationary environments where obstacles move or goals shift mid-execution. By integrating continuous monitoring and partial reasoning updates, developers can transition from rigid trajectory following to responsive systems that override stale plans when local risks, such as proximity to dynamic obstacles, exceed safety thresholds.

Key Insights

• Online A* Planning: The agent utilizes a computational budget of 5000 expansions to find paths incrementally in shifting environments (Tutorial context).
• Receding-Horizon Control: By committing to a short horizon of 6 steps, the system prevents execution of stale data while maintaining progress toward a moving target.
• Reactive Risk Modeling: A lightweight risk gate set at 0.85 allows the agent to override planned moves with safer alternatives if the local environment becomes hazardous.
• Structured Streaming: Using Pydantic’s BaseModel, the agent emits real-time ‘plan’, ‘decide’, and ‘observe’ events to provide transparent reasoning throughout its lifecycle.
• Adaptive Overrides: The system implements an ‘action_risk’ function that evaluates neighbors to ensure agent safety regardless of the global path’s instructions.

Working Examples

Definition of the streaming event schema using Pydantic for structured reasoning updates.

import random, math, time
from dataclasses import dataclass, field
from typing import List, Tuple, Dict, Optional, Generator, Any
from collections import deque, defaultdict
try:
    from pydantic import BaseModel, Field
except Exception:
    raise RuntimeError("Please install pydantic: `!pip -q install pydantic` (then rerun).")

class StreamEvent(BaseModel):
    t: float = Field(..., description="Wall-clock time (seconds since start)")
    kind: str = Field(..., description="event type, e.g., plan/update/act/observe/done")
    step: int = Field(..., description="agent step counter")
    msg: str = Field(..., description="human-readable partial reasoning summary")
    data: Dict[str, Any] = Field(default_factory=dict, description="structured payload")

A risk model that evaluates the local safety of a planned move based on obstacle proximity and boundaries.

def action_risk(world: DynamicGridWorld, next_pos: Coord) -> float:
    x, y = next_pos
    near = 0
    for dx, dy in [(1,0),(-1,0),(0,1),(0,-1)]:
        c = (x+dx, y+dy)
        if world.in_bounds(c) and c in world.obstacles:
            near += 1
    edge = 1 if (x in [0, world.w-1] or y in [0, world.h-1]) else 0
    return 0.25 * near + 0.15 * edge

Practical Applications

• Dynamic Navigation Systems: Agents use receding-horizon control to navigate around unexpected obstacles; failing to replan on target movement leads to high goal-miss rates.
• Industrial Robotics: Reactive overrides allow robots to adjust paths when human workers enter a workspace; relying on a static path results in safety violations or emergency shutdowns.
• Autonomous Logistics: Systems implementing partial reasoning updates can stream status to human supervisors; ignoring surprise environmental shifts leads to inefficient pathing and stalled throughput.

References:

• https://www.marktechpost.com/2026/03/11/how-to-design-a-streaming-decision-agent-with-partial-reasoning-online-replanning-and-reactive-mid-execution-adaptation-in-dynamic-environments/