Neste tutorial, construiremos um Agente de Decisão Streaming que pensa e age em um ambiente online e mutável, transmitindo continuamente atualizações seguras de raciocínio parcial. Implementamos um mundo de grade dinâmico com obstáculos em movimento e um objetivo variável, e então usamos um planejador A* online em um loop de horizonte recessivo para se comprometer com apenas alguns movimentos de curto prazo e reavaliar frequentemente. À medida que executamos, tomamos decisões intermediárias que podem anular o plano quando uma etapa se torna inválida ou localmente arriscada, permitindo-nos adaptar-nos no meio da execução, em vez de seguir cegamente uma trajetória obsoleta. Copiar Código Copiado Usar um navegador diferente 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("Por favor, instale pydantic: `!pip -q install pydantic` (depois execute novamente).") class StreamEvent(BaseModel): t: float = Field(..., description="Tempo de relógio (segundos desde o início)") kind: str = Field(..., description="tipo de evento, por exemplo, plan/update/act/observe/done") step: int = Field(..., description="contador de passos do agente") msg: str = Field(..., description="resumo de raciocínio parcial legível por humanos") data: Dict[str, Any] = Field(default_factory=dict, description="carga útil estruturada") Coord = Tuple[int, int] Definimos o esquema de evento de streaming e as estruturas de tipo principais que nos permitem emitir atualizações de raciocínio estruturadas. Usamos Pydantic para formalizar a estrutura de cada decisão ou observação transmitida de forma segura e consistente. Estabelecemos a interface fundamental que alimenta o raciocínio incremental ao longo do ciclo de vida do agente. Copiar Código Copiado Usar um navegador diferente @dataclass class DynamicGridWorld: w: int = 18 h: int = 10 obstacle_ratio: float = 0.18 seed: int = 7 move_obstacles_every: int = 6 spawn_obstacle_prob: float = 0.25 clear_obstacle_prob: float = 0.15 target_jitter_prob: float = 0.35 rng: random.Random = field(init=False) obstacles: set = field(init=False, default_factory=set) agent: Coord = field(init=False, default=(1, 1)) target: Coord = field(init=False, default=(15, 7)) step_count: int = field(init=False, default=0) def __post_init__(self): self.rng = random.Random(self.seed) self.reset() def reset(self): self.step_count = 0 self.obstacles = set() for y in range(self.h): for x in range(self.w): if (x, y) in [(1, 1), (self.w - 2, self.h - 2)]: continue if self.rng.random() < sel
Fonte: MarkTechPost
Publicado em 2026-03-11