LangGraph одобрување: кодот пред interrupt() може да се изврши двапати

11 октомври 2026 г., во 13:00|Автор: Уредувачки тим на QUASA|5 мин. читање

За човечко одобрување во LangGraph, повикајте interrupt() пред чувствителната алатка и продолжете со Command(resume=...) и истиот thread_id. Упатството на LangGraph за прекини објаснува дека при продолжување јазолот почнува од почеток: кодот пред interrupt() повторно се извршува. За паузата да преживее рестартирање на сервисот, графот треба да користи траен checkpointer.

Практичниот распоред е барањето да се подготви и зачува пред јазолот за одобрување, а испраќањето, наплатата или запишувањето да се изврши по позитивната одлука. На чувствителната операција ѝ треба и стабилен идентификатор: одобрувањето спречува дејство пред одлуката, но само по себе не го решава повторниот обид по прекин на врската или пад на процесот.

Што смее да стои пред interrupt()

Во јазолот за одобрување, пред interrupt() ставете читање од зачуваната состојба, составување на барањето за преглед и други пресметки што безбедно се повторуваат. Не создавајте таму нов случаен идентификатор и не повикувајте алатка што испраќа порака, наплатува или додава запис. При продолжување, новиот идентификатор би можел да го претстави истиот обид како нова операција.

Зачувајте ги примачот, содржината и operation_id пред паузата. Лицето што одобрува треба да ги види вредностите што подоцна ќе ги добие алатката. Ако интерфејсот дозволува промена на примачот или текстот, измената бара нова јасно определена верзија за одобрување; во спротивно, претходната одлука може да се примени на дејство што човекот не го видел. interrupt() прима податоци што може да се претворат во JSON, па за прегледот испратете едноставни вредности од состојбата.

Минимален Python образец со трајна состојба

Следниот условен образец е за е-пошта. send_once е адаптер на апликацијата, а DB_URI и workflow_id доаѓаат од нејзината конфигурација и евиденција за барања. Алатката не се повикува во јазолот што паузира; така границата меѓу одлуката и дејството останува видлива во графот.

Увезете TypedDict и NotRequired од typing, PostgresSaver од langgraph.checkpoint.postgres, StateGraph, START и END од langgraph.graph, како и Command и interrupt од langgraph.types. Состојбата State нека има задолжителни operation_id, recipient и body, и незадолжителни approved и result.
Јазолот за одобрување нека биде: def approve(s): return {'approved': interrupt({'recipient': s['recipient'], 'body': s['body']}) is True}. Проверувањето со is True пропушта само булова вредност True; текстуален одговор како 'yes' не ја активира алатката.
Јазолот за дејство нека биде: def execute(s): return {'result': send_once(s['recipient'], s['body'], s['operation_id'])}. send_once мора да го препознае истиот operation_id и истата содржина при повторен обид и да го врати зачуваниот исход кога тој е познат.
Создајте builder = StateGraph(State), додајте ги јазлите approve и execute и поврзете го START со approve. Со builder.add_conditional_edges('approve', lambda s: 'execute' if s['approved'] else END) одобрениот тек оди кон алатката, а одбиениот завршува. Поврзете го execute со END.

За продукциска употреба, упатството за трајна состојба прикажува PostgresSaver.from_conn_string(DB_URI), пакети за PostgreSQL и повик на checkpointer.setup() при првото поставување. Отворете го checkpointer како контекст, компајлирајте со builder.compile(checkpointer=cp) и извршувајте го графот додека контекстот е отворен. По рестартирање, отворете нова врска кон истата база и компајлирајте го истиот граф.

Поставете config = {'configurable': {'thread_id': workflow_id}}. Првиот graph.invoke(initial_state, config=config) запира на одобрувањето; барањето за човекот е во резултатот под __interrupt__. По одлуката, повикајте graph.invoke(Command(resume=True), config=config) за одобрување или Command(resume=False) за одбивање. Зачувајте го workflow_id во серверската евиденција за барањето: нов thread_id упатува на нова состојба, а не на паузираното извршување.

Идемпотентност на чувствителната операција

Операцијата по одобрувањето може да се повтори ако надворешниот сервис ја прифати, а процесот падне пред графот да го зачува резултатот. Затоа operation_id треба да се создаде еднаш за конкретното барање, пред одобрувањето, и да се пренесува непроменет при секој обид. Адаптерот треба да одбие повторна употреба на истиот идентификатор со изменет примач, износ или содржина.

Кај плаќањата, правилата на Stripe за идемпотентни барања предвидуваат ист клуч за повторен обид и споредување на параметрите; клучевите може да се отстранат откако ќе бидат стари најмалку 24 часа. Затоа заштитата што ја дава давателот не треба да се третира како неограничена евиденција на вашата апликација. За сопствено запишување во база, единствено ограничување врз operation_id и запишување на исходот во трансакција се посигурен образец од одделна проверка „дали постои“ пред внесување.

Кај давател на е-пошта без идемпотентен повик, локален единствен запис може да спречи два работника истовремено да ја преземат истата задача. Останува нејасен случајот кога давателот ја прифатил пораката, а процесот паднал пред да го зачува успехот. Тогаш адаптерот треба да може да го утврди исходот кај давателот или да го означи барањето за разрешување, наместо слепо да испрати повторно.

Одобрувањето мора да биде врзано за овластено лице

thread_id служи за наоѓање на зачуваното извршување; сам по себе не докажува право на одобрување. Серверската рута што го прима одговорот треба да го провери идентитетот на лицето, неговото овластување за конкретната операција и фактот дека барањето чека одлука. Клиентски доставен thread_id не треба да биде единствениот услов за повик на Command(resume=...).

Зачувајте кој ја донел одлуката и кои вредности му биле прикажани. При одобрување пред испраќање е-пошта, ова ја врзува одлуката за конкретен примач и текст. Ако нацртот се промени по прегледот, новата верзија треба повторно да помине низ одобрување.

Тест за повторно продолжување

Во тест заменете го send_once со адаптер што ги брои обидите и ги бележи operation_id и стварните дејства одделно. Извршете го графот до __interrupt__ и проверете дека нема дејство. Продолжете со False и проверете го истото. Со нов workflow_id повторете ја паузата, продолжете со True и проверете дека има едно одобрено дејство. Тест-бројач во јазолот за одобрување може да покаже дека неговиот код пред interrupt() бил повикан и при паузата и при продолжувањето.

Потоа симулирајте пад откако лажната алатка го евидентирала дејството, но пред јазолот да го врати резултатот. По повторниот обид, бројот на повици до адаптерот може да порасне, додека бројот на стварни дејства за истиот operation_id треба да остане еден. Токму оваа разлика го проверува ризикот што не се гледа во едноставен тест на одобрување: алатката мора да препознае повторување и кога паузата и овластувањето се правилно поставени.

Прочитајте и:

Сподели: