
Python LangGraph Send로 작업 나누고 동시 실행 수 제한하기
입력 목록의 항목마다 Send를 만들고, graph.ainvoke의 config 최상위에 max_concurrency를 넣으면 돼요. 결과를 모으는 상태에는 reducer도 필요해요.
개발·코드·도구을 더 깊이 이해하고, 일상에 필요한 답을 찾아가는 기록.

입력 목록의 항목마다 Send를 만들고, graph.ainvoke의 config 최상위에 max_concurrency를 넣으면 돼요. 결과를 모으는 상태에는 reducer도 필요해요.

오래 걸리는 LangGraph 노드의 진행 상황은 노드 안에서 get_stream_writer()로 보내고, 호출 측에서 custom 스트림으로 받으면 돼요. 최종 결과는 노드의 반환값으로 남기고, 작업 단계와 완료 건수는 실행 도중 별도로 전달하는 방식이에요.

Python LangGraph에서 병렬 노드 하나만 실패했다면, checkpointer에 실행 기록을 저장하고 같은 thread_id로 입력값 None을 전달해 재개하면 돼요.

Python LangGraph에서 반복 한도에 걸리기 전에 부분 결과를 반환하려면, 상태에 RemainingSteps를 선언하고 남은 단계가 적을 때 마무리 노드로 분기하면 돼요. 처리한 데이터는 상태에 쌓고, 반환값에는 완료 여부와 종료 이유를 함께…

저장된 대화를 유지하려면 모델 호출 노드 안에서 trim_messages로 입력 목록을 만들고, 그 목록만 model.invoke에 전달하면 돼요. 노드는 새 응답만 messages 업데이트로 반환합니다. 축소한 목록으로 저장 상태를 교체하지 않는 것이 핵심이에요.

체크포인터를 연결하고, 실패 원인을 해결한 뒤 같은 thread_id로 graph.invoke(None, config)를 호출하면 돼요. 같은 병렬 실행 단계에서 성공한 노드의 결과는 pending writes로 보존되어 재개할 때 다시 계산하지 않아요.

새 대화에서 thread_id가 바뀌어도 사용자 설정을 유지하려면, 설정을 Store의 사용자별 namespace에 저장하고 매 실행마다 같은 user_id로 조회하면 돼요. 대화 상태는 체크포인터에 맡기고, 언어·응답 형식처럼 대화를 넘어 적용할 값만 따로 저장하는…

입력 문서 수가 실행할 때마다 달라진다면, 조건부 엣지에서 문서마다 Send("worker", 작업별_입력)를 만들어 반환하면 돼요. 각 worker가 돌려준 결과는 reducer로 하나의 목록에 모으고, 마지막 노드에서 입력 ID와 대조하면 됩니다.

Task.cancel()은 실행 중인 코드를 강제로 끊는 명령이 아니에요. 작업에 취소를 요청하고, 작업이 그 신호를 확인해 종료하도록 하는 협력적 취소 방식이에요. 반복문에 취소 확인이 없거나 try?로 취소 오류를 삼키면 이후 코드가…

Swift actor 안에서도 await로 실행이 중단되면 다른 호출이 같은 actor의 상태를 바꿀 수 있어요. 이를 재진입(reentrancy)이라고 해요. actor는 격리된 상태의 동시 접근을 막지만, 비동기 메서드 전체를 하나의 작업처럼 보장하지는 않아요.

LangGraph에서 interrupt를 재개하면 멈췄던 노드가 처음부터 다시 실행돼요. 같은 노드에서 interrupt보다 먼저 API를 호출했다면 그 호출도 반복됩니다.

반환값이 있는 async 함수는 테스트 자체를 async로 만들고 await 뒤에 #expect를 붙이면 됩니다.