В этом руководстве мы создаём агента для потоковых решений, который может мыслить и действовать в онлайн-среде, постоянно изменяющейся, и непрерывно транслировать обновления частичного анализа.
Основные шаги
1. Создание динамической сетки с движущимися препятствиями и изменяемой целью.
2. Использование онлайн-планировщика A* в цикле с уменьшающимся горизонтом для принятия решений о нескольких ближайших шагах и их частой переоценке.
3. Принятие промежуточных решений, которые могут переопределить план, когда шаг становится недействительным или локально рискованным.
Код и структуры данных
Мы определяем схему потокового события и основные типы структур, которые позволяют нам выдавать структурированные обновления анализа. Используем Pydantic для формализации структуры каждого потокового решения или наблюдения.
«`python
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»)
«`
Динамическая сетка
«`python
@dataclass
class DynamicGridWorld:
w: int = 18
h: int = 10
obstacle_ratio: float = 0.18
seed: int = 7
moveobstaclesevery: int = 6
spawnobstacleprob: float = 0.25
clearobstacleprob: float = 0.15
targetjitterprob: 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)
«`
Агент для потоковых решений
«`python
@dataclass
class StreamingDecisionAgent:
cfg: AgentConfig
world: DynamicGridWorld
starttime: float = field(init=False, defaultfactory=time.time)
step_id: int = field(init=False, default=0)
currentplan: List[Coord] = field(init=False, defaultfactory=list)
currentactions: List[str] = field(init=False, defaultfactory=list)
lastsnapshot: Dict[str, Any] = field(init=False, defaultfactory=dict)
stats: Dict[str, Any] = field(init=False, default_factory=lambda: defaultdict(int))
«`
Запуск агента
«`python
def runandprint(agent: StreamingDecisionAgent, throttle: float = 0.0):
last_kind = None
for ev in agent.run():
header = f»[t={ev.t:6.2f}s | step={ev.step:03d} | {ev.kind.upper():7}]»
print(header, ev.msg)
if ev.kind in {«plan», «observe», «world»}:
if «gridwithpath» in ev.data:
print(ev.data[«gridwithpath»])
elif «grid» in ev.data:
print(ev.data[«grid»])
if throttle > 0:
time.sleep(throttle)
«`
Заключение
Мы создали агента, который демонстрирует инкрементальный анализ, онлайн-планирование и реактивное поведение. Агент легко запускается и проверяется в Colab. Потоковая передача структурированных событий делает процесс принятия решений наблюдаемым, сохраняя при этом краткость и безопасность сводок анализа для пользователя.
1. Какие основные шаги включает в себя проектирование агента для потоковых решений с частичным анализом, онлайн-перепланированием и адаптацией в динамических средах?
Ответ: проектирование такого агента включает в себя создание динамической сетки с движущимися препятствиями и изменяемой целью, использование онлайн-планировщика A* в цикле с уменьшающимся горизонтом для принятия решений о нескольких ближайших шагах и их частой переоценке, а также принятие промежуточных решений, которые могут переопределить план, когда шаг становится недействительным или локально рискованным.
2. Какие структуры данных используются для определения схемы потокового события и основных типов структур, которые позволяют выдавать структурированные обновления анализа?
Ответ: для определения схемы потокового события и основных типов структур используются классы и структуры данных Python, такие как `dataclasses`, `deque`, `defaultdict` и `BaseModel` из библиотеки `pydantic`.
3. Какие компоненты включает в себя класс `StreamingDecisionAgent`?
Ответ: класс `StreamingDecisionAgent` включает в себя следующие компоненты: `cfg` (конфигурация агента), `world` (динамическая сетка), `starttime` (время начала работы агента), `stepid` (идентификатор шага), `currentplan` (текущий план действий), `currentactions` (текущие действия), `last_snapshot` (последний снимок состояния) и `stats` (статистика).
4. Какие функции используются для запуска агента и вывода результатов его работы?
Ответ: для запуска агента используется функция `runandprint`, которая запускает агента и выводит результаты его работы в виде структурированных событий. Функция `runandprint` принимает агента и параметр `throttle`, который определяет скорость вывода результатов.
5. Какие возможности предоставляет агент для потоковых решений, созданный в статье?
Ответ: агент для потоковых решений, созданный в статье, демонстрирует инкрементальный анализ, онлайн-планирование и реактивное поведение. Он легко запускается и проверяется в Colab, а потоковая передача структурированных событий делает процесс принятия решений наблюдаемым, сохраняя при этом краткость и безопасность сводок анализа для пользователя.