Практычныя прытамулі: Ваш агентскі граф не паслядае 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 стаюць вузламі графа

Для этапу компіляцыі ў Трык 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: Зміна вузла за сесыяй (участь чалавека)

Калі працюеце над Трыкам 5 «Гарячая замена этапа», спачатку запісайце угоду: неабяжлівыя вхідныя даны, сигнал успеху і тое, што выходзіць у разе частковага абякання. Такі список пераконтроўваець дапамагае залічыцца чыстымі пазнейшыя змены коду. Спрацоўвайце з гэтым этапам як з угодай межа вхіднымі данымі і паверанымі выходнымі даннымі. Дайце назву артыфактам, задаце пераконтроўванні успеху і адмовіцеся ад тыхоўага частковага завершэння. Запісвайце ID запытку, ID модэлю і час адклікання праз кожны вызов. Без гэтага следу періодычныя памылкі прадастаўця выглядаюць як багі прыкладнага програмнага забезпечэння.

# 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.

Чек-ліст для эксплуатацыі

Для этапа чек-ліста для эксплуатацыі неабходна з’явіць вхідныя даны, адпаведальнага за крок і критэрыя завершэння пры зміне коду. Аператары павінны магчымаць перзапуск крока з вядомай точкі контролю, не прабуючы здагадвацца пра схованы стан.

Зберагаеце настройкі зовні коду прыемліка. Файлы сераўнавальнага сэрвісу, хранення секретных дадзенаў і флагі функцыйяй павінны знаходзіцца ў адном месцы, куды аператары можаюць адбавіць аудыт без неабяжнага чытання всіх дадзенаў.

Раздзеляйце стварэнне кліента ад цыклу паведамленняў, каб было можна зменіць прадаўцаў без перапісвання машыны стану размовы.

Ствараеце контрольныя пункты пасля дорогіх крокаў. Система вярнення праблемы не павінна знову стаўляць плату за той самы вызов LLM, калі аператар перапрынцяе выконанне на пазнейшый вузел.

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

Спрыймайце гэты этап як кантракт між вхіднымі дадзеннямі і перакананымі выходнымі рэзультатамі. Даўайце назвы артыфактам, задаюце критэрыя успеху і адмовляйцеся ад мовчанкавага частковага завершэння.

Перш чым запускать стак, заморозьце версіі, зафіксавце «золаты» транскрыпты для критичных шляхоў і паказваце крокі для вярнення да пачатковага стану. У спільных серавысках неабходны ліміты частоты запытоў, перакананні ў правах на выкарыстоўванне ресурсоў і чыста вялікі адпаведальны за змěну секрэтных даных. Валіце простую надзейнасць працы над крэатіўными, але разовымі дамах.

Прыметкі для 2822ea5988ca: не кладзіце ключы прадаўцаў у репазітарый, задаце верхнюю межу токеноў на кожную сесію і зберагачыце транскрыпты празаўсёды поблізу фікстураў для ацэнкі, каб пазнейшыя замены модэляў заставаліся пораўнанымі.