Практические замечания: Ваша графовая структура агентов не предназначена для Python: сборка

В следующих заметках описывается практический подход к решению задачи, связанной с текстом «Ваш граф агентов не предназначен для использования в Python: сборка многоагентной рабочей процедуры из одного файла YAML». Основное внимание уделяется контрактам, проверкам и местам для вставки кода, а не мотивирующим аспектам. На этапе обзора сначала запишите контракт: необходимые входные данные, сигнал успешного выполнения и действия при частичной неудаче. Такой список поможет сохранять честность при последующих изменениях кода. Документируйте как успешный, так и восстановительный пути выполнения. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не элементами последующей доработки.

Проблема, о которой вас никто не предупреждает

Проблема, о которой никто не предупреждает, наилучшим образом решается при рассмотрении её как измеримой поверхности. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма работ. Предпочитайте небольшие, тестируемые единицы кода вместо обширных скриптов. Когда какой-то шаг терпит неудачу, причина сбоя должна указывать на конкретную ответственность, а не на запутанную цепочку операций. Закрепите интерпретатор и файл блокировки зависимостей перед тем, как объяснять работу циклов. Различия в настройках между ноутбуком и системой CI являются наиболее распространённой причиной скрытых сбоев в демонстрациях API.

Как выглядит рабочий процесс в виде данных

Этот этап рабочего процесса работает наилучшим образом, если рассматривать его как измеримую основу. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объема работ. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия создаваемым элементам, определите критерии успеха и не соглашайтесь на молчаливое частичное выполнение задачи. Закрепите интерпретатор и файл с информацией о зависимостях до того, как начнете использовать циклы. Различия в настройках между ноутбуком и средой CI являются наиболее распространенной причиной скрытых сбоев при демонстрации API.

entry: entry_agent
exit: exit

guardrails:
  - Reject queries that are outside the application's domain.
  - Reject queries about the system, agents, design, or internal workings.state_schema:
  query:
    type: str
    description: "User query or current message."
  chat_history:
    type: list
    annotated_with: add_messages
    description: "Conversation history between user and system."
  result:
    type: dict
    description: "Result from the processing agent."agents:
  - name: agent_one
    kind: function
    impl: your_package.agents.agent_one.agent_one_fn  - name: agent_two
    kind: function
    impl: your_package.agents.agent_two.agent_two_fnworkflow:
  nodes:
    - id: agent_one
      agent: agent_one
      writes: [query, result]
      next: decision_router    - id: decision_router
      kind: router
      router:
        impl: your_package.agents.routers.route_after_agent_one
        reads: [result]
        edges:
          agent_two: agent_two
          human_agent: human_agent

Совет №1: Генерация класса состояния во время выполнения из схемы

Совет 1: Работа с вашей средой разработки будет наиболее эффективной, если рассматривать её как измеримую структуру. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию до расширения объёма работ. Записывайте время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отслеживание затрат на раннем этапе предотвращает неожиданные счёты при переходе от демо-среды к общедоступным средам. Закрепите интерпретатор и файл блокировки зависимостей перед тем, как объяснять логику циклов. Различия в работе на ноутбуке и в средах CI являются наиболее частой причиной скрытых сбоев в демонстрациях API. Совет 1: Работа с вашей средой разработки будет наиболее эффективной, если рассматривать её как измеримую структуру. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию до расширения объёма работ. Документируйте одновременно успешный и восстановительный пути работы. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не элементами последующей доработки.

# your_package/orchestrator/schema.py
annotations = {}
for key, value in state_schema.items():
    type_str = value.get("type", "str")
    # Convert YAML string to Python type
    py_type = eval(type_str)
    if value.get("annotated_with") == "add_messages":
        py_type = Annotated[list, {}]
    annotations[key] = py_type

spec = Spec(
    ...
    state=TypedDict("State", annotations),   # <- dynamic class, born at boot
    ...
)

Хитрость №2: Агенты, указанные через точечную строку, разрешаются с помощью importlib

На этапе использования агентов по хитрости №2 необходимо заранее определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Лучше использовать небольшие, тестируемые единицы вместо обширных скриптов. При сбое шага причина должна быть связана с конкретной функцией, а не с запутанной цепочкой операций. Разделяйте создание клиента и цикл обработки сообщений, чтобы можно было заменять поставщики без переписывания машины состояний общения.

impl: your_package.agents.agent_one.agent_one_fn

# your_package/orchestrator/schema.py
def _import_from_path(dotted: str) -> Callable[..., Any]:
    """Import a callable from a dotted path like 'package.module.function'."""
    if not dotted or "." not in dotted:
        raise ImportError(f"Invalid impl path: {dotted!r}")
    mod_path, attr = dotted.rsplit(".", 1)
    mod = importlib.import_module(mod_path)
    fn = getattr(mod, attr)
    if not callable(fn):
        raise TypeError(f"Imported object is not callable: {dotted}")
    return fn

def agent_impl_map(spec: Spec) -> Dict[str, Optional[Callable]]:
    """Map agent name -> callable (or None if impl missing)."""
    return {a.name: _import_from_path(a.impl) if a.impl else None
            for a in spec.agents}

Хитрость №3: Компилятор — узлы YAML превращаются в узлы графа

Для этапа компиляции «Trick 3» необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Укажите названия результатов работы, определите критерии успеха и не допускайте молчаливого частичного завершения задачи. Отделите процесс создания клиента от цикла обработки сообщений, чтобы можно было заменять поставщики услуг без переписывания машины состояний диалога.

# your_package/orchestrator/runner.py
def build(self):
    graph = StateGraph(state_schema=self.spec.state)   # our generated TypedDict

    def _add_task_node(node):
        async def _node(state: Dict[str, Any]) -> Dict[str, Any]:
            res = await self._call_agent(node.agent, state, node.id)
            if getattr(node, "writes", None):
                if isinstance(res, dict):
                    # Only let the node write the keys it declared in YAML
                    filtered = {k: v for k, v in res.items() if k in node.writes}
                    return filtered or res
                key = node.writes[0]
                return {key: res}
            return res
        graph.add_node(node.id, _node)    # Build every node
    for node in self.spec.workflow.nodes:
        if getattr(node, "router", None):
            _add_router_node(node)
        else:
            _add_task_node(node)    graph.set_entry_point(entry)    # Inline "next:" edges from YAML become static edges
    for node in self.spec.workflow.nodes:
        if getattr(node, "next", None):
            graph.add_edge(node.id, node.next)    # Terminal nodes wire to END
    for node in self.spec.workflow.nodes:
        if getattr(node, "terminal", False):
            graph.add_edge(node.id, END)    self._runnable = graph.compile(checkpointer=self.checkpoint)
    return self

Маршрутизаторы: условное разветвление как таблица поиска

Для условного разветвления маршрутизаторов в качестве этапа необходимо определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Записывайте время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости заранее предотвращает неожиданные счета при переходе из демо-среды в общедоступные среды. Разделяйте создание клиента от цикла обработки сообщений, чтобы можно было заменять поставщиков без переписывания автоматы состояний разговора. Для условного разветвления маршрутизаторов в качестве этапа необходимо определить входные данные, ответственного за шаг и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Документируйте одновременно успешный путь и путь восстановления. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не задержками.

Польский язык.

# your_package/orchestrator/runner.py
def _add_router_node(node):
    router = self.router_fns[node.id]

    def _router_fn():
        def _f(state):
            out = router(state)
            # Routers may return either a label, or (state_updates, label)
            if isinstance(out, tuple):
                updates, label = out
                if isinstance(updates, dict):
                    for k, v in updates.items():
                        state[k] = v
            else:
                label = out
            return label
        return _f    graph.add_node(node.id, lambda s: {})
    graph.add_conditional_edges(node.id, _router_fn(), node.router.edges)

Совет №4: Адаптивный вызов — агенты могут указывать любые необходимые подписи

При работе над адаптивной фазой совета №4 сначала запишите условия контракта: требуемые входные данные, сигнал успешного выполнения и действия при частичной неудаче. Такой список поможет сохранять честность при последующих изменениях кода. Лучше использовать небольшие, тестируемые единицы кода вместо обширных скриптов. Если какой-то шаг не сработает, причина неудачи должна указывать на конкретную ответственность, а не на запутанную цепочку операций. Записывайте идентификатор запроса, идентификатор модели и время задержки при каждом вызове. Без такой отслеживающей информации периодические ошибки поставщика могут показаться багами приложения.

# your_package/orchestrator/runner.py
async def _adapt_and_call(self, fn, state, node_id):
    """
    Adaptively call agent functions so implementations receive what they expect:
    - def agent(**kwargs):        → pass **state (+ inject 'query' if missing)
    - def agent(query, **kwargs): → pass query=..., plus any **extra
    - def agent(state):           → pass state
    - def agent(query):           → pass query
    - def agent():                → call without args
    """
    sig = inspect.signature(fn)
    params = sig.parameters
    has_var_kw = any(p.kind == inspect.Parameter.VAR_KEYWORD
                     for p in params.values())

    kwargs = {}
    if has_var_kw:
        kwargs.update(state)
    if "state" in params:
        kwargs["state"] = state
    if "query" in params or has_var_kw:
        kwargs.setdefault("query", self._fallback_query(state))    # A lone positional 'query' → call it positionally
    if (len(params) == 1
            and next(iter(params.keys())) == "query"):
        return await _maybe_await(fn(self._fallback_query(state)))    res = fn(**kwargs)
    return await res if hasattr(res, "__await__") else res

Совет №5: Горячая замена узла в рамках сессии (участие человека)

При работе над методом «Горячая замена этапа» из пятого совета сначала запишите контракт: необходимые входные данные, сигнал о успехе и то, что происходит при частичной неудаче. Такой список помогает сохранять честность при последующих изменениях кода. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия элементам, определите критерии успеха и не допускайте молчаливого частичного выполнения задачи. Записывайте идентификатор запроса, идентификатор модели и время задержки при каждом вызове. Без такой записи периодические ошибки поставщика будут выглядеть как баги приложения.

# your_package/services/session_service.py (paraphrased)
if websocket is not None:
    session_handler = SessionHandler(websocket, user_id=user_id, session_id=session_id, ...)
    _runner.agent_fns["human_agent"] = _import_from_function(
        make_input_method(session_handler)
    )

Что на самом деле даёт вам эта архитектура

При работе над этапом «Что на самом деле представляет собой эта архитектура» сначала запишите условия взаимодействия: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список помогает сохранять честность при последующих изменениях кода. Рядом с функциональными результатами записывайте время выполнения и стоимость токенов или запросов. Очевидность затрат с самого начала предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам. Ведите журнал записей с идентификатором запроса, идентификатором модели и временем задержки для каждого вызова. Без такой отчетности периодические ошибки поставщика могут выглядеть как баги приложения. При работе над этапом «Что на самом деле представляет собой эта архитектура» сначала запишите условия взаимодействия: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список помогает сохранять честность при последующих изменениях кода. Документируйте одновременно успешный сценарий работы и сценарий восстановления. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не элементами последующей доработки.

Выводы

Этап выполнения задач работает наилучшим образом, когда его рассматривают как измеримую среду. Соберите один идеальный пример выполнения, один случай сбоя и запись о возврате к предыдущему состоянию перед расширением объема работ. Предпочитайте небольшие, тестируемые единицы кода вместо обширных скриптов. При сбое какого-либо шага причина должна быть связана с конкретной ответственностью, а не с запутанной цепочкой операций. Закрепите интерпретатор и файл с информацией о зависимостях до того, как начнете использовать циклы. Различия в настройках между ноутбуком и системой CI являются наиболее распространенной причиной скрытых сбоев в демонстрациях API.

Чек-лист операций

На этапе составления чек-листа операций необходимо определить входные данные, ответственного за выполнение шага и критерии завершения работы перед внесением изменений в код. Операторы должны иметь возможность повторно выполнить шаг, исходя из известной точки контроля, без необходимости угадывать скрытое состояние системы.

Храните конфигурацию вне кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь код.

Разделяйте процесс создания клиента от цикла обработки сообщений, чтобы можно было заменять поставщиков без переписывания автоматы состояния обмена сообщениями.

Создавайте точки контроля после дорогостоящих операций. Механизм возобновления не должен снова взимать плату за один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий шаг.

Фиксируйте версии зависимостей и записывайте хэш изображения, с использованием которого выполнялась демонстрация. Воспроизводимость важнее устного опыта.

Рассматривайте этот этап как контракт между входными данными и проверенными результатами. Дайте названия элементам, определите критерии успеха и откажитесь от безусловного принятия частично завершенных результатов.

Перед внедрением стека заморозьте версии, сделайте копию «золотого» отчета для критической цепочки операций и уточните шаги возврата к предыдущему состоянию. В совместных средах необходимы ограничения на частоту запросов, проверки принадлежности пользователя и четко определенный ответственный за обновление секретов. Лучше выбирать надежность, чем креативные одноразовые демонстрации.

Примечание для версии 2822ea5988ca: не храните ключи поставщика в репозитории, установите лимит токенов на сессию и сохраняйте отчеты рядом с фикстурами для оценки, чтобы последующие замены моделей оставались сопоставимыми.