LangGraph Send 사용법: 입력 개수만큼 작업을 나누고 결과 검증까지
입력 문서 수가 실행할 때마다 달라진다면, 조건부 엣지에서 문서마다 Send("worker", 작업별_입력)를 만들어 반환하면 돼요. 각 worker가 돌려준 결과는 reducer로 하나의 목록에 모으고, 마지막 노드에서 입력 ID와 대조하면 됩니다. Send는 같은 노드를 서로 다른 상태로 여러 번 호출하는 용도로 제공돼요. 여기서는 외부 API 없이 문서 길이를 계산하며, 입력 1개·여러 개·빈 목록까지 확인하는 완전한 예제를 만들어요. Send 공식 레퍼런스
문서별 분배와 집계 검증 전체 코드
# send_documents.py
# 준비: python -m pip install -U langgraph
# 실행: python send_documents.py
# Python 3.11+ 기준. 외부 API나 모델을 호출하지 않습니다.
import operator
from collections import Counter
from typing import Annotated, TypedDict
from langgraph.graph import END, START, StateGraph
from langgraph.types import Send
class Document(TypedDict):
id: str
text: str
class Result(TypedDict):
id: str
char_count: int
class State(TypedDict):
documents: list[Document]
results: Annotated[list[Result], operator.add]
ordered_results: list[Result]
class Task(TypedDict):
doc: Document
def prepare(state: State) -> dict:
ids = [doc['id'] for doc in state['documents']]
if len(ids) != len(set(ids)):
raise ValueError('입력 문서 ID는 서로 달라야 합니다.')
return {}
def distribute(state: State) -> list[Send] | str:
if not state['documents']:
return 'collect'
return [
Send('worker', {'doc': doc})
for doc in state['documents']
]
def worker(task: Task) -> dict:
doc = task['doc']
result: Result = {
'id': doc['id'],
'char_count': len(doc['text']),
}
# 전체 상태 대신 이번 작업에서 만든 갱신만 반환합니다.
return {'results': [result]}
def collect(state: State) -> dict:
expected = Counter(doc['id'] for doc in state['documents'])
actual = Counter(row['id'] for row in state['results'])
if actual != expected:
raise ValueError(
f'결과 누락·중복 또는 알 수 없는 ID: '
f'expected={expected}, actual={actual}'
)
by_id = {row['id']: row for row in state['results']}
ordered = [by_id[doc['id']] for doc in state['documents']]
# results에 반환하면 reducer가 다시 붙이므로 별도 필드를 씁니다.
return {'ordered_results': ordered}
builder = StateGraph(State)
builder.add_node('prepare', prepare)
builder.add_node('worker', worker)
builder.add_node('collect', collect)
builder.add_edge(START, 'prepare')
builder.add_conditional_edges(
'prepare', distribute, ['worker', 'collect']
)
builder.add_edge('worker', 'collect')
builder.add_edge('collect', END)
graph = builder.compile()
def run(documents: list[Document]) -> list[Result]:
output = graph.invoke({
'documents': documents,
'results': [],
'ordered_results': [],
})
return output['ordered_results']
def main() -> None:
cases = [
('one', [{'id': 'a', 'text': 'hello'}],
[{'id': 'a', 'char_count': 5}]),
('many', [
{'id': 'b', 'text': 'cat'},
{'id': 'a', 'text': ''},
{'id': 'c', 'text': 'hello'},
], [
{'id': 'b', 'char_count': 3},
{'id': 'a', 'char_count': 0},
{'id': 'c', 'char_count': 5},
]),
('empty', [], []),
]
for name, documents, expected in cases:
actual = run(documents)
assert actual == expected, (name, actual, expected)
print(name, actual)
try:
run([
{'id': 'same', 'text': 'first'},
{'id': 'same', 'text': 'second'},
])
except ValueError as exc:
assert str(exc) == '입력 문서 ID는 서로 달라야 합니다.'
print('duplicate: rejected')
else:
raise AssertionError('중복 ID를 거절해야 합니다.')
if __name__ == '__main__':
main()
노드를 늘리는 대신 작업별 입력을 만들어요
동적으로 달라지는 것은 worker의 정의가 아니라 호출할 작업의 개수예요. 그래프에는 worker를 한 번만 등록하고, 분배 함수가 입력 목록을 순회하며 Send 객체를 만들어요. 이때 두 번째 인자로 보내는 상태는 전체 그래프 상태와 달라도 됩니다. 그래서 전체 문서 목록을 가진 State와 문서 하나만 가진 Task를 분리할 수 있어요. Send 공식 레퍼런스
문서별 분배와 집계 검증 전체 코드에서는 Task에 doc만 넣었어요. 이 구조에서 worker가 접근할 입력은 task["doc"]이고, 전체 목록인 task["documents"]는 전달하지 않았어요. 작업 함수에 언어나 처리 옵션이 필요하다면 해당 값을 Task와 Send의 인자에 함께 추가하세요. 필요한 값이 어디서 오는지 입력 계약에 드러내면, 작업 함수만 따로 읽어도 처리 범위를 이해하기 쉬워요.
이 예제의 char_count는 원문에 len을 적용하는 단순 계산이에요. LLM 토큰 수나 요약 품질을 측정하는 값으로 해석하면 안 돼요. 먼저 결정적인 계산으로 분배 구조를 확인한 뒤, worker 내부의 계산만 실제 문서 분석 함수로 교체하는 출발점으로 사용하세요.
결과를 합치는 규칙과 최종 가공을 분리해요
여러 작업이 같은 results 키에 값을 쓰므로 해당 필드에는 operator.add reducer를 지정했어요. reducer가 없는 동일 키에 병렬 갱신이 들어오면 INVALID_CONCURRENT_GRAPH_UPDATE 오류가 발생할 수 있어요. 각 작업은 자신이 만든 결과 하나를 목록으로 감싸 반환하고, 누적된 전체 목록을 다시 반환하지 않는 편이 맞아요. 동시 상태 갱신 오류 안내
여기서 reducer와 collect는 역할이 달라요. reducer는 작업 결과 목록을 연결하고, collect는 연결된 결과가 입력과 맞는지 검사한 뒤 표시 순서를 정해요. 특히 collect가 정렬한 목록을 results로 다시 반환하면 덧셈 규칙에 따라 기존 결과에 붙게 됩니다. 예제에서 ordered_results라는 별도 필드를 둔 이유예요. 누적용 필드와 최종 출력용 필드를 구분하면 이 실수를 피하기 쉬워요.
State 전체를 수정해서 반환할 필요도 없어요. worker는 results만, collect는 ordered_results만 반환해요. 공식 문서도 상태를 직접 변경하기보다 필요한 갱신 값을 반환하는 방식을 안내해요. Graph API 사용 가이드
빈 목록도 집계 경로를 명시해요
이 예제의 일반 경로는 prepare, 여러 worker, collect 순서예요. 같은 단계에서 실행되는 병렬 작업들이 끝난 뒤 다음 단계가 진행되는 LangGraph의 super-step 모델을 이용해요. 따라서 각 worker가 한 단계로 끝나는 현재 구조에서는 worker에서 collect로 엣지를 연결해 결과를 모을 수 있어요. Graph API 실행 모델
입력이 비었을 때는 분배 함수가 collect를 직접 반환하도록 설계했어요. 아무 작업도 만들지 않는 것과 최종 집계 노드를 실행하는 것은 다른 요구이기 때문이에요. 이 분기 덕분에 빈 입력에도 동일한 출력 필드와 검증 절차를 적용할 수 있어요. 빈 목록을 오류로 취급해야 하는 서비스라면 prepare에서 거절하는 정책으로 바꾸면 됩니다.
작업 내부를 여러 그래프 노드로 쪼개고 분기마다 경로 길이까지 달라지면 합류 조건을 다시 검토해야 해요. 현재의 한 단계 worker 구조에 대한 설명을 그대로 일반화하면 안 됩니다. 공식 가이드는 길이가 다른 분기들의 최종 처리를 위해 defer=True로 노드 실행을 미루는 방법을 설명해요. 이런 확장은 기본 예제의 검증을 통과한 뒤 별도로 적용하는 편이 좋아요. Graph API 사용 가이드
개수만 맞는 결과를 성공으로 판단하지 않아요
결과가 세 개라는 사실만으로 문서 세 개가 모두 처리됐다고 단정할 수는 없어요. 예를 들어 기대한 ID가 a, b, c인데 실제 결과가 a, a, c라면 개수는 같아도 b가 빠져 있어요. collect는 입력과 출력의 ID별 등장 횟수를 비교해 이런 누락과 중복을 함께 거절하도록 작성했어요. 이는 LangGraph의 자동 보장이 아니라 이 예제에서 추가한 애플리케이션 검증이에요.
입력 단계에서 ID 중복도 거절해요. 여기서는 ID 하나를 처리 대상 하나로 정의했기 때문이에요. 같은 문서를 여러 설정으로 처리해야 한다면 문서 ID와 별개로 작업 ID를 만들고, 결과 검증에도 그 작업 ID를 사용하세요. 예를 들어 문서 a의 한국어 처리와 영어 처리는 a:ko, a:en처럼 구별하는 방식이 가능해요.
책을 페이지별 텍스트로 나눠 처리하는 경우도 생각해 볼 수 있어요. 참고 사진인 File:Page number.jpg는 책에 인쇄된 페이지 번호를 보여 줘요. 페이지 번호를 ID에 활용한다면 서로 다른 책의 같은 쪽이 충돌하지 않도록 book-a:978처럼 책 식별자도 붙이세요. 이때 예제에 넣는 text는 이미 추출한 문자열이어야 해요. 사진에서 글자를 읽는 OCR은 이 코드에 포함하지 않았어요. 책의 페이지 번호 참고 사진
결과 목록의 위치를 입력 위치로 간주하지도 마세요. 공식 가이드는 병렬 단계의 갱신 순서가 일관적이지 않을 수 있다고 안내해요. 예제는 ID로 결과를 찾은 뒤 원래 입력 순서로 재배열해요. ID의 사전순으로 정렬하는 방식과 달리, 호출자가 지정한 문서 순서를 보존할 수 있어요. Graph API 사용 가이드
실행 확인 뒤에는 검증 코드를 일부러 깨뜨려 보세요
새 가상환경에서 python -m pip install -U langgraph로 설치하고, 문서별 분배와 집계 검증 전체 코드를 send_documents.py로 저장한 뒤 python send_documents.py로 실행하세요. 설치 방식은 공식 안내를 따랐어요. 예제는 Python 3.11 이상을 기준으로 작성했으며, 실제 실행 결과를 관측한 보고서는 아니에요. 각 assert에 코드가 만족해야 할 기대값을 명시했어요. LangGraph 설치 안내
제공한 정상 사례를 확인했다면 worker가 반환하는 ID를 임시로 고정해 보세요. 여러 입력을 처리할 때 collect의 검증이 실패해야 해요. 또 결과 목록을 빈 목록으로 바꾸면 누락을 잡아야 합니다. 정상 출력만 보는 것보다, 검증 장치가 잘못된 출력을 거절하는지도 확인할 수 있는 방법이에요.
이 코드는 checkpointer 없이 한 번의 호출 안에서 분배와 집계를 끝내는 범위예요. 실제 처리에서 일부 실패도 결과로 남기고 싶다면 결과 타입에 성공 여부와 오류 내용을 추가하고, 완료된 입력과 성공한 입력을 따로 집계하도록 설계하세요. 다음 단계에서는 ID 검증은 유지한 채 worker의 계산만 교체하면, 분배 문제와 실제 처리 함수의 문제를 구분하며 확장할 수 있어요.