Python LangGraph Send로 작업 나누고 동시 실행 수 제한하기
입력 목록의 항목마다 Send를 만들고, graph.ainvoke의 config 최상위에 max_concurrency를 넣으면 돼요. 결과를 모으는 상태에는 reducer도 필요해요. 예를 들어 입력이 20개이고 제한이 3이면 작업은 20개를 만들되, 이 예제의 worker는 동시에 최대 3개까지만 실행하도록 설정할 수 있어요. configurable 안에 넣는 설정이 아니라는 점이 중요해요. LangGraph 공식 가이드
Send 동시성 검증 전체 코드
# Python 3.11 이상 환경에서 사용하세요.
# 설치: python -m pip install -U langgraph
# 저장: send_limit_demo.py
# 실행: python send_limit_demo.py
# 외부 API, API 키, 계정이 필요하지 않아요.
# 공식 API를 조합한 예제이며 실측 로그를 제시하는 코드는 아니에요.
import asyncio
import operator
from importlib.metadata import version
from typing import Annotated, TypedDict
from langgraph.graph import END, START, StateGraph
from langgraph.types import Send
class Job(TypedDict):
item_id: int
value: int
class Result(TypedDict):
item_id: int
squared: int
class State(TypedDict):
items: list[int]
results: Annotated[list[Result], operator.add]
def dispatch(state: State) -> list[Send] | str:
if not state["items"]:
return END
return [
Send("worker", {"item_id": i, "value": value})
for i, value in enumerate(state["items"])
]
async def run_case(items: list[int], limit: int) -> None:
if type(limit) is not int or limit < 1:
raise ValueError("limit는 1 이상의 정수여야 해요")
active = 0
peak = 0
async def worker(job: Job) -> dict[str, list[Result]]:
nonlocal active, peak
# 단일 이벤트 루프에서 이 갱신 구간에는 await가 없어요.
active += 1
peak = max(peak, active)
try:
print(f"start id={job['item_id']} active={active}")
# 외부 I/O 대신 항목별로 다른 대기 시간을 줘요.
await asyncio.sleep(0.05 * (1 + job["item_id"] % 3))
result: Result = {
"item_id": job["item_id"],
"squared": job["value"] ** 2,
}
return {"results": [result]}
finally:
active -= 1
print(f"end id={job['item_id']} active={active}")
builder = StateGraph(State)
builder.add_node("worker", worker)
builder.add_conditional_edges(START, dispatch)
builder.add_edge("worker", END)
graph = builder.compile()
output = await graph.ainvoke(
{"items": items, "results": []},
config={"max_concurrency": limit},
)
ordered = sorted(output["results"], key=lambda row: row["item_id"])
# 누락, 중복, 잘못된 항목 대응을 각각 확인해요.
assert [row["item_id"] for row in ordered] == list(range(len(items)))
assert [row["squared"] for row in ordered] == [x ** 2 for x in items]
assert active == 0
if items:
assert 1 <= peak <= min(limit, len(items))
else:
assert peak == 0
print(f"PASS items={len(items)} limit={limit} peak={peak}")
print(ordered)
async def main() -> None:
print("langgraph:", version("langgraph"))
print("langchain-core:", version("langchain-core"))
await run_case(list(range(8)), limit=3)
await run_case([7, 7, 9], limit=1)
await run_case([42], limit=3)
await run_case([], limit=3)
if __name__ == "__main__":
asyncio.run(main())Send는 작업 개수를, 실행 설정은 동시성을 정해요
고정된 노드 여러 개를 미리 연결하기 어려운 상황이 있어요. 실행할 때마다 문서가 2개일 수도, 200개일 수도 있는 경우죠. Send는 조건부 엣지에서 목적지 노드와 그 노드에 전달할 입력을 지정해요. 동일한 worker를 여러 번 호출하면서 각 호출에 서로 다른 데이터를 넘길 수 있고, 전달하는 데이터 구조가 전체 그래프 상태와 같을 필요도 없어요. Send API 레퍼런스
이 차이를 코드에서는 State와 Job으로 나눴어요. State는 전체 입력과 누적 결과를 담고, Job은 한 항목의 위치와 값만 담아요. worker에 전체 목록을 넘겨 다시 반복문을 돌리면 항목을 나눈 의미가 없어져요. 분배 함수는 작업을 배정하고, worker는 배정받은 한 항목만 처리하게 두는 편이 이해하기 쉬워요.
max_concurrency는 RunnableConfig에 정의된 병렬 호출 상한이에요. 입력 데이터에 limit라는 필드를 추가하는 것만으로 실행기가 그 값을 읽지는 않아요. 예제의 limit 인자는 호출 직전에 config={"max_concurrency": limit}로 전달하기 때문에 실제 실행 설정이 됩니다. RunnableConfig 레퍼런스
결과 개수보다 항목의 대응 관계를 확인해요
Send 동시성 검증 전체 코드에서는 같은 값이 두 번 들어오는 경우도 검사해요. [7, 7, 9]의 첫 번째 7과 두 번째 7은 서로 다른 작업이에요. 값을 식별자로 쓰면 둘을 구분하지 못하므로 enumerate로 입력 위치를 붙였어요. 실제 문서를 처리할 때는 이 위치 대신 고유한 문서 ID를 사용할 수 있어요.
결과 수만 맞는다고 모두 처리됐다고 판단하면 안 돼요. 입력 ID가 0, 1, 2인데 결과 ID가 0, 0, 2여도 개수는 같거든요. 예제는 정렬한 ID 목록을 기대 목록과 비교하므로 누락과 중복을 함께 잡아요. 이어서 각 ID의 계산 결과도 대조해, 다른 항목의 결과를 잘못 붙인 경우까지 확인합니다.
여러 worker가 results에 쓴 업데이트는 operator.add로 합쳐요. worker마다 한 항목짜리 결과 목록만 반환하는 이유예요. 전체 누적 목록을 매번 돌려주면 이미 모인 결과까지 다시 더하는 구조가 될 수 있어요. 병렬 업데이트의 순서에 기대지 않도록 출력 직전에 ID로 정렬합니다. 공식 가이드도 일정한 순서가 필요하면 정렬 기준을 함께 저장하라고 안내해요. LangGraph 공식 가이드
peak는 실행 속도가 아니라 제한 준수 여부예요
active는 worker 본문에 진입했지만 아직 빠져나오지 않은 작업 수이고, peak는 그 최댓값이에요. 예제의 계측 구간에는 sleep뿐 아니라 결과 생성도 포함돼요. 따라서 이 값은 HTTP 연결 수나 CPU 코어 사용량을 뜻하지 않아요. 어떤 구간의 동시성을 측정했는지부터 정해야 숫자를 잘못 해석하지 않습니다.
await asyncio.sleep은 현재 작업을 잠시 멈추고 다른 작업이 실행될 기회를 줘요. 여기서는 외부 요청 없이 대기 시간이 겹치는 상황을 만들기 위해 썼어요. 반대로 비동기 함수 안에서 time.sleep 같은 차단 호출을 사용하면 이벤트 루프 진행을 막을 수 있어요. 그런 상태에서 peak가 낮게 나왔다고 제한 설정이 잘됐다고 결론 내리면 곤란해요. Python asyncio 문서
판정식은 peak == limit가 아니라 peak <= limit예요. 상한은 반드시 그 수만큼 동시에 실행한다는 약속이 아니고, 입력이 한 개면 제한을 3으로 줘도 작업은 하나뿐이니까요. 특히 peak가 1인 실행 하나만으로는 병렬 실행이 제대로 일어났는지 알 수 없어요. 작업이 충분한 사례의 시작 로그에서 active가 2 이상으로 올라가는지도 함께 살펴보세요.
계측 변수는 run_case 호출 내부에 두었어요. 예제처럼 단일 이벤트 루프에서 await 없는 짧은 구간으로 카운터를 갱신하는 구성을 전제로 합니다. worker를 동기 함수와 스레드 기반 계측으로 바꾸려면 카운터 보호 방식도 다시 설계해야 해요. 비동기 함수라는 선언 자체가 모든 공유 데이터의 안전성을 보장하는 것은 아니에요. Python asyncio 문서
설정 위치와 제한 범위를 헷갈리지 마세요
예전 예제를 옮겨 왔다면 configurable 안에 max_concurrency가 들어 있는지 먼저 확인하세요. 공식 저장소에는 그 위치 때문에 제한이 적용되지 않는 문서 예제가 보고됐고, 2026년 10월 16일에는 최상위 설정으로 문서를 수정했다는 안내와 함께 이슈가 닫혔어요. 현재 글은 최상위 키를 사용해요. 공식 저장소의 문서 수정 이슈
적용 범위도 구분해야 해요. 확인한 비동기 실행기 구현은 실행기 인스턴스가 설정을 읽어 세마포어를 만들어요. 따라서 이 설정을 서비스 전체가 공유하는 요청 제한으로 간주하면 안 돼요. 여러 그래프 호출이나 여러 서버를 합쳐 하나의 상한을 지켜야 한다면, 공통 자원을 사용하는 지점에 별도 제어를 설계하세요. LangGraph 비동기 실행기 소스
서버 랙에 여러 서버가 들어 있는 배포 환경을 생각해 볼게요. 서로 독립적인 서버 두 대가 이 예제를 각각 limit=3으로 실행한다면, 두 실행의 worker가 합쳐 최대 6개까지 겹칠 수 있는 구성이에요. 서버 랙 참고 사진은 이런 물리적 배치의 예시이며, 사진 속 장비가 LangGraph를 실행한다거나 특정 처리량을 낸다는 뜻은 아니에요. 서버 랙 참고 사진
worker 내부에서 asyncio.gather로 요청 여러 개를 직접 시작하는 경우도 같은 문제예요. 그래프 작업 하나와 외부 요청 하나가 더 이상 일대일이 아니에요. 예를 들어 worker 3개가 각각 요청 4개를 동시에 시작하도록 직접 작성했다면, 그래프의 상한 3을 외부 요청 상한 3으로 해석할 수 없어요. 이때는 요청 함수 쪽에도 공유 제한을 두는 방법을 검토하세요.
또한 동시 실행 수는 초당 요청 수와 달라요. 가령 세 작업이 모두 0.1초 만에 끝나고 즉시 다음 작업을 시작한다는 단순 가정에서는, 동시성이 3이어도 초당 처리량은 약 30건이 될 수 있어요. 이는 측정 결과가 아니라 차이를 설명하는 계산이에요. 초당 요청 한도가 요구사항이라면 시간 기준의 별도 제한이 필요합니다.
빈 입력과 큰 입력에서는 무엇이 달라질까요
빈 목록은 worker를 만들지 않고 END로 보내도록 명시했어요. 처리할 항목이 없는 것도 정상 결과로 다루기 위한 선택이에요. 한편 limit가 0일 때 이를 일시정지처럼 해석하지 않도록 함수 입구에서 양의 정수만 받게 했어요. 설치한 패키지 버전도 출력하므로, 프로젝트에 옮긴 뒤 동작이 다르면 실행 환경부터 비교할 수 있어요.
동시 실행 수를 줄여도 이 코드가 만드는 Send 목록 자체는 입력 길이만큼 커져요. 결과도 전부 메모리에 모아요. 입력이 아주 크다면 동시성 설정만 조정하지 말고, 입력을 여러 묶음으로 읽고 결과를 묶음별로 저장하는 구조를 고려하세요. 이는 예제의 리스트 생성과 누적 방식을 바탕으로 한 설계 제안이에요.
이 예제는 정상 완료한 작업의 대응 관계를 검사하며, 실패 복구나 재시도를 추가하지 않았어요. finally는 작업이 빠져나갈 때 카운터를 되돌리는 역할이지, 실패를 성공 결과로 바꾸는 장치가 아니에요. 실제 처리 함수를 넣을 때는 한 항목의 실패로 전체 호출을 실패시킬지, 실패 ID와 오류를 결과로 남길지 먼저 정하세요. 공식 가이드는 노드 내부 예외 처리와 RetryPolicy를 별도의 선택지로 설명해요. LangGraph 공식 가이드
프로젝트에 적용할 때는 먼저 샘플 계산 부분만 실제 비동기 처리 함수로 바꾸고, ID 대조와 peak 검사를 유지해 보세요. 처리 함수가 바뀐 뒤에도 항목이 빠지지 않는지, 원하는 구간에서 동시성 상한을 지키는지 확인하는 것이 다음 단계예요.