In this tutorial, we build a Streaming Decision Agent that thinks and acts in an online, changing environment while continuously streaming safe, partial reasoning updates. We implement a dynamic grid world with moving obstacles and a shifting goal, then use an online A* planner in a receding-horizon loop to commit to only a few near-term moves and re-evaluate frequently. As we execute, we make intermediate decisions that can override the plan when a step becomes invalid or locally risky, allowing us to adapt mid-run rather than unthinkingly following a stale trajectory.<\/p>\n
import random, math, time\nfrom dataclasses import dataclass, field\nfrom typing import List, Tuple, Dict, Optional, Generator, Any\nfrom collections import deque, defaultdict\n\n\ntry:\n from pydantic import BaseModel, Field\nexcept Exception:\n raise RuntimeError(\"Please install pydantic: `!pip -q install pydantic` (then rerun).\")\n\n\nclass StreamEvent(BaseModel):\n t: float = Field(..., description=\"Wall-clock time (seconds since start)\")\n kind: str = Field(..., description=\"event type, e.g., plan\/update\/act\/observe\/done\")\n step: int = Field(..., description=\"agent step counter\")\n msg: str = Field(..., description=\"human-readable partial reasoning summary\")\n data: Dict[str, Any] = Field(default_factory=dict, description=\"structured payload\")\n\n\nCoord = Tuple[int, int]<\/code><\/pre>\n<\/div>\n<\/div>\nWe define the streaming event schema and core type structures that allow us to emit structured reasoning updates. We use Pydantic to formalize the structure of each streamed decision or observation safely and consistently. We establish the foundational interface that powers incremental reasoning throughout the agent lifecycle.<\/p>\n
\n\n\n\n<\/div>\n<\/div>\n<\/div>\n<\/div>\nCopy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n@dataclass\nclass DynamicGridWorld:\n w: int = 18\n h: int = 10\n obstacle_ratio: float = 0.18\n seed: int = 7\n move_obstacles_every: int = 6\n spawn_obstacle_prob: float = 0.25\n clear_obstacle_prob: float = 0.15\n target_jitter_prob: float = 0.35\n rng: random.Random = field(init=False)\n obstacles: set = field(init=False, default_factory=set)\n agent: Coord = field(init=False, default=(1, 1))\n target: Coord = field(init=False, default=(15, 7))\n step_count: int = field(init=False, default=0)\n\n\n def __post_init__(self):\n self.rng = random.Random(self.seed)\n self.reset()\n\n\n def reset(self):\n self.step_count = 0\n self.obstacles = set()\n for y in range(self.h):\n for x in range(self.w):\n if (x, y) in [(1, 1), (self.w - 2, self.h - 2)]:\n continue\n if self.rng.random() < self.obstacle_ratio:\n self.obstacles.add((x, y))\n self.agent = (1, 1)\n self.target = (self.w - 2, self.h - 2)\n self._ensure_free(self.agent)\n self._ensure_free(self.target)\n\n\n def _ensure_free(self, c: Coord):\n if c in self.obstacles:\n self.obstacles.remove(c)\n\n\n def in_bounds(self, c: Coord) -> bool:\n x, y = c\n return 0 <= x < self.w and 0 <= y < self.h\n\n\n def passable(self, c: Coord) -> bool:\n return c not in self.obstacles\n\n\n def neighbors4(self, c: Coord) -> List[Coord]:\n x, y = c\n cand = [(x+1,y), (x-1,y), (x,y+1), (x,y-1)]\n return [p for p in cand if self.in_bounds(p) and self.passable(p)]\n\n\n def manhattan(self, a: Coord, b: Coord) -> int:\n return abs(a[0]-b[0]) + abs(a[1]-b[1])\n\n\n def maybe_world_changes(self) -> Dict[str, Any]:\n changes = {\"obstacles_added\": [], \"obstacles_cleared\": [], \"target_moved\": False}\n self.step_count += 1\n if self.rng.random() < self.target_jitter_prob:\n tx, ty = self.target\n options = [(tx+1,ty),(tx-1,ty),(tx,ty+1),(tx,ty-1)]\n options = [c for c in options if self.in_bounds(c) and c != self.agent]\n self.rng.shuffle(options)\n for c in options[:3]:\n if c not in self.obstacles:\n self.target = c\n changes[\"target_moved\"] = True\n break\n if self.step_count % self.move_obstacles_every == 0:\n for _ in range(4):\n if self.rng.random() < self.clear_obstacle_prob and self.obstacles:\n c = self.rng.choice(tuple(self.obstacles))\n self.obstacles.remove(c)\n changes[\"obstacles_cleared\"].append(c)\n for _ in range(5):\n if self.rng.random() < self.spawn_obstacle_prob:\n c = (self.rng.randrange(self.w), self.rng.randrange(self.h))\n if c != self.agent and c != self.target:\n self.obstacles.add(c)\n changes[\"obstacles_added\"].append(c)\n self._ensure_free(self.agent)\n self._ensure_free(self.target)\n return changes\n\n\n def step(self, action: str) -> Dict[str, Any]:\n ax, ay = self.agent\n move = {\"R\": (ax+1, ay), \"L\": (ax-1, ay), \"D\": (ax, ay+1), \"U\": (ax, ay-1), \"S\": (ax, ay)}[action]\n moved = False\n if self.in_bounds(move) and self.passable(move):\n self.agent = move\n moved = True\n changes = self.maybe_world_changes()\n done = (self.agent == self.target)\n return {\"moved\": moved, \"agent\": self.agent, \"target\": self.target, \"done\": done, \"changes\": changes}\n\n\n def render(self, path: Optional[List[Coord]] = None) -> str:\n path_set = set(path or [])\n lines = []\n for y in range(self.h):\n row = []\n for x in range(self.w):\n c = (x, y)\n if c == self.agent:\n row.append(\"A\")\n elif c == self.target:\n row.append(\"T\")\n elif c in path_set:\n row.append(\"\u00b7\")\n elif c in self.obstacles:\n row.append(\"\u2588\")\n else:\n row.append(\" \")\n lines.append(\"\".join(row))\n border = \"+\" + \"-\" * self.w + \"+\"\n body = \"n\".join([\"|\" + ln + \"|\" for ln in lines])\n return f\"{border}n{body}n{border}\"<\/code><\/pre>\n<\/div>\n<\/div>\nWe construct a dynamic grid world that evolves with shifting obstacles and a moving target. We simulate environmental non-stationarity to test online planning under uncertainty. We implement rendering and world-transition logic to observe how the agent reacts to real-time changes.<\/p>\n
\n\n\n\n<\/div>\n<\/div>\n<\/div>\n<\/div>\nCopy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n@dataclass\nclass PlanResult:\n path: List[Coord]\n cost: float\n expanded: int\n reason: str\n\n\ndef astar(world: DynamicGridWorld, start: Coord, goal: Coord, max_expand: int = 5000) -> PlanResult:\n frontier = []\n import heapq\n heapq.heappush(frontier, (world.manhattan(start, goal), 0, start))\n came_from: Dict[Coord, Optional[Coord]] = {start: None}\n gscore: Dict[Coord, float] = {start: 0}\n expanded = 0\n while frontier and expanded < max_expand:\n f, g, cur = heapq.heappop(frontier)\n expanded += 1\n if cur == goal:\n path = []\n c = cur\n while c is not None:\n path.append(c)\n c = came_from[c]\n path.reverse()\n return PlanResult(path=path, cost=gscore[cur], expanded=expanded, reason=\"found_path\")\n for nb in world.neighbors4(cur):\n ng = gscore[cur] + 1\n if nb not in gscore or ng < gscore[nb]:\n gscore[nb] = ng\n came_from[nb] = cur\n h = world.manhattan(nb, goal)\n heapq.heappush(frontier, (ng + h, ng, nb))\n return PlanResult(path=[start], cost=float(\"inf\"), expanded=expanded, reason=\"no_path_or_budget\")\n\n\ndef path_to_actions(path: List[Coord]) -> List[str]:\n actions = []\n for (x1,y1),(x2,y2) in zip(path, path[1:]):\n if x2 == x1+1 and y2 == y1: actions.append(\"R\")\n elif x2 == x1-1 and y2 == y1: actions.append(\"L\")\n elif x2 == x1 and y2 == y1+1: actions.append(\"D\")\n elif x2 == x1 and y2 == y1-1: actions.append(\"U\")\n else: actions.append(\"S\")\n return actions\n\n\ndef action_risk(world: DynamicGridWorld, next_pos: Coord) -> float:\n x, y = next_pos\n near = 0\n for dx, dy in [(1,0),(-1,0),(0,1),(0,-1)]:\n c = (x+dx, y+dy)\n if world.in_bounds(c) and c in world.obstacles:\n near += 1\n edge = 1 if (x in [0, world.w-1] or y in [0, world.h-1]) else 0\n return 0.25 * near + 0.15 * edge<\/code><\/pre>\n<\/div>\n<\/div>\nWe implement the A* online planner along with action extraction and local risk evaluation. We compute shortest paths incrementally while respecting a computational budget. We also introduce a lightweight risk model to enable us to override unsafe planned moves during execution.<\/p>\n
\n\n\n\n<\/div>\n<\/div>\n<\/div>\n<\/div>\nCopy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n@dataclass\nclass AgentConfig:\n horizon: int = 6\n replan_on_target_move: bool = True\n replan_on_obstacle_change: bool = True\n max_steps: int = 120\n think_latency: float = 0.02\n act_latency: float = 0.01\n risk_gate: float = 0.85\n alt_search_depth: int = 2\n\n\n@dataclass\nclass StreamingDecisionAgent:\n cfg: AgentConfig\n world: DynamicGridWorld\n start_time: float = field(init=False, default_factory=time.time)\n step_id: int = field(init=False, default=0)\n current_plan: List[Coord] = field(init=False, default_factory=list)\n current_actions: List[str] = field(init=False, default_factory=list)\n last_snapshot: Dict[str, Any] = field(init=False, default_factory=dict)\n stats: Dict[str, Any] = field(init=False, default_factory=lambda: defaultdict(int))\n\n\n def _now(self) -> float:\n return time.time() - self.start_time\n\n\n def _emit(self, kind: str, msg: str, data: Optional[Dict[str, Any]] = None) -> StreamEvent:\n return StreamEvent(t=self._now(), kind=kind, step=self.step_id, msg=msg, data=data or {})\n\n\n def _need_replan(self, obs: Dict[str, Any]) -> bool:\n ch = obs[\"changes\"]\n if obs[\"done\"]:\n return False\n if not self.current_plan or len(self.current_plan) <= 1:\n return True\n if self.cfg.replan_on_target_move and ch.get(\"target_moved\"):\n return True\n if self.cfg.replan_on_obstacle_change and (ch.get(\"obstacles_added\") or ch.get(\"obstacles_cleared\")):\n return True\n if len(self.current_plan) > 1 and self.current_plan[1] in self.world.obstacles:\n return True\n return False\n\n\n def _plan(self) -> PlanResult:\n time.sleep(self.cfg.think_latency)\n self.stats[\"replans\"] += 1\n return astar(self.world, self.world.agent, self.world.target)\n\n\n def _choose_action(self, planned_action: str) -> Tuple[str, str]:\n ax, ay = self.world.agent\n action_to_delta = {\"R\": (1,0), \"L\": (-1,0), \"D\": (0,1), \"U\": (0,-1), \"S\": (0,0)}\n dx, dy = action_to_delta[planned_action]\n nxt = (ax+dx, ay+dy)\n if not self.world.in_bounds(nxt) or not self.world.passable(nxt):\n self.stats[\"overrides\"] += 1\n return \"S\", \"planned_move_invalid -> wait.\"\n r = action_risk(self.world, nxt)\n if r > self.cfg.risk_gate:\n candidates = [\"U\",\"D\",\"L\",\"R\",\"S\"]\n best = (planned_action, float(\"inf\"), \"keep_plan\")\n for a in candidates:\n dx, dy = action_to_delta[a]\n p = (ax+dx, ay+dy)\n if not self.world.in_bounds(p) or not self.world.passable(p):\n continue\n score = action_risk(self.world, p) + 0.05 * self.world.manhattan(p, self.world.target)\n if score < best[1]:\n best = (a, score, \"risk_avoidance_override\")\n if best[0] != planned_action:\n self.stats[\"overrides\"] += 1\n return best[0], best[2]\n return planned_action, \"follow_plan\"\n\n\n def run(self) -> Generator[StreamEvent, None, None]:\n yield self._emit(\"observe\", \"Initialize: reading initial state.\", {\"agent\": self.world.agent, \"target\": self.world.target})\n yield self._emit(\"world\", \"Initial world snapshot.\", {\"grid\": self.world.render()})\n for self.step_id in range(1, self.cfg.max_steps + 1):\n if self.step_id == 1 or self._need_replan(self.last_snapshot):\n pr = self._plan()\n self.current_plan = pr.path\n self.current_actions = path_to_actions(pr.path)\n if pr.reason != \"found_path\":\n yield self._emit(\"plan\", \"Planner could not find a path within budget; switching to reactive exploration.\", {\"reason\": pr.reason, \"expanded\": pr.expanded})\n self.current_actions = []\n else:\n horizon_path = pr.path[: max(2, min(len(pr.path), self.cfg.horizon + 1))]\n yield self._emit(\"plan\", f\"Plan updated (online A*). Commit to next {len(horizon_path)-1} moves, then re-evaluate.\", {\"reason\": pr.reason, \"path_len\": len(pr.path), \"expanded\": pr.expanded, \"commit_horizon\": self.cfg.horizon, \"horizon_path\": horizon_path, \"grid_with_path\": self.world.render(path=horizon_path)})\n if self.current_actions:\n planned_action = self.current_actions[0]\n else:\n ax, ay = self.world.agent\n tx, ty = self.world.target\n options = []\n if tx > ax: options.append(\"R\")\n if tx < ax: options.append(\"L\")\n if ty > ay: options.append(\"D\")\n if ty < ay: options.append(\"U\")\n options += [\"S\",\"U\",\"D\",\"L\",\"R\"]\n planned_action = options[0]\n action, why = self._choose_action(planned_action)\n yield self._emit(\"decide\", f\"Intermediate decision: action={action} ({why}).\", {\"planned_action\": planned_action, \"chosen_action\": action, \"agent\": self.world.agent, \"target\": self.world.target})\n time.sleep(self.cfg.act_latency)\n obs = self.world.step(action)\n self.last_snapshot = obs\n if self.current_actions:\n if action == planned_action:\n self.current_actions = self.current_actions[1:]\n if len(self.current_plan) > 1:\n self.current_plan = self.current_plan[1:]\n ch = obs[\"changes\"]\n surprise = []\n if ch.get(\"target_moved\"): surprise.append(\"target_moved\")\n if ch.get(\"obstacles_added\"): surprise.append(f\"obstacles_added={len(ch['obstacles_added'])}\")\n if ch.get(\"obstacles_cleared\"): surprise.append(f\"obstacles_cleared={len(ch['obstacles_cleared'])}\")\n surprise_msg = (\"Surprises: \" + \", \".join(surprise)) if surprise else \"No major surprises.\"\n self.stats[\"steps\"] += 1\n if obs[\"moved\"]: self.stats[\"moves\"] += 1\n if ch.get(\"target_moved\"): self.stats[\"target_moves\"] += 1\n if ch.get(\"obstacles_added\") or ch.get(\"obstacles_cleared\"): self.stats[\"world_shifts\"] += 1\n yield self._emit(\"observe\", f\"Observed outcome. {surprise_msg}\", {\"moved\": obs[\"moved\"], \"agent\": obs[\"agent\"], \"target\": obs[\"target\"], \"done\": obs[\"done\"], \"changes\": ch, \"grid\": self.world.render(path=self.current_plan[: min(len(self.current_plan), 10)])})\n if obs[\"done\"]:\n yield self._emit(\"done\", \"Goal reached. Stopping execution.\", {\"final_agent\": obs[\"agent\"], \"final_target\": obs[\"target\"], \"stats\": dict(self.stats)})\n return\n yield self._emit(\"done\", \"Max steps reached without reaching the goal.\", {\"final_agent\": self.world.agent, \"final_target\": self.world.target, \"stats\": dict(self.stats)})<\/code><\/pre>\n<\/div>\n<\/div>\nWe design the Streaming Decision Agent that integrates planning, monitoring, and reactive overrides. We implement receding-horizon control, committing only to near-term steps and replanning when surprises occur. We stream structured events at every stage, planning, deciding, acting, and observing, to demonstrate incremental reasoning in action.<\/p>\n
\n\n\n\n<\/div>\n<\/div>\n<\/div>\n<\/div>\nCopy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\ndef run_and_print(agent: StreamingDecisionAgent, throttle: float = 0.0):\n last_kind = None\n for ev in agent.run():\n header = f\"[t={ev.t:6.2f}s | step={ev.step:03d} | {ev.kind.upper():7}]\"\n print(header, ev.msg)\n if ev.kind in {\"plan\", \"observe\", \"world\"}:\n if \"grid_with_path\" in ev.data:\n print(ev.data[\"grid_with_path\"])\n elif \"grid\" in ev.data:\n print(ev.data[\"grid\"])\n if throttle > 0:\n time.sleep(throttle)\n\n\nworld = DynamicGridWorld(w=18, h=10, obstacle_ratio=0.18, seed=10, move_obstacles_every=6)\ncfg = AgentConfig(horizon=6, replan_on_target_move=True, replan_on_obstacle_change=True, max_steps=120, think_latency=0.01, act_latency=0.01, risk_gate=0.85, alt_search_depth=2)\nagent = StreamingDecisionAgent(cfg=cfg, world=world)\nrun_and_print(agent, throttle=0.0)<\/code><\/pre>\n<\/div>\n<\/div>\nWe build the streaming runner and execute the full agent loop inside the dynamic environment. We print structured reasoning updates in real time to simulate a live decision stream. We finally observe the agent adapting continuously, replanning when needed, and completing the task under changing conditions.<\/p>\n
In conclusion, we have an agent that demonstrates incremental reasoning, online planning, and reactive behavior, and it is easy to run and inspect in Colab. We saw how streaming structured events makes the agent\u2019s decision process observable while still keeping reasoning summaries concise and user-safe. Also, we showed how continuous monitoring and replanning turn a static planner into a responsive system that can handle surprises, moving targets, changing obstacles, and step-level risk without stopping execution.<\/p>\n
\nCheck out\u00a0Full Codes here<\/a>.\u00a0<\/strong>Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0120k+ ML SubReddit<\/a><\/strong>\u00a0and Subscribe to\u00a0our Newsletter<\/a><\/strong>. Wait! are you on telegram?\u00a0now you can join us on telegram as well.<\/a><\/strong><\/p>\nThe post How to Design a Streaming Decision Agent with Partial Reasoning, Online Replanning, and Reactive Mid-Execution Adaptation in Dynamic Environments<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"In this tutorial, we build a S…<\/p>\n","protected":false},"author":1,"featured_media":29,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-540","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/540","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=540"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/540\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/media\/29"}],"wp:attachment":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=540"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=540"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=540"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}