그래프 실행
그래프 실행 (Graph execution)
노드별 상태와 스트리밍 콘텐츠로 다단계 그래프 파이프라인을 시각화해요.
LangGraph 에이전트는 블랙박스가 아닙니다. 모든 그래프는 순차 또는 병렬로 실행되는 **이름 있는 노드들(named nodes)**로 구성됩니다. 분류(classify), 조사(research), 분석(analyze), 종합(synthesize)처럼요. 그래프 실행 카드(Graph execution cards)는 각 노드에 카드를 렌더링해 이 파이프라인을 보이게 만듭니다. 각 카드는 상태를 보여주고, 콘텐츠를 실시간으로 스트리밍하며, 전체 워크플로우의 완료를 추적합니다. 사용자는 에이전트가 무엇을 하는지, 어떤 단계에 있는지, 각 단계가 무엇을 만들었는지 정확히 볼 수 있어요.
출처: 문서
본문
이 패턴은 그래프 구조를 제품 UX로 바꾸기 때문에 프로덕션 에이전트에 특히 유용합니다. 런을 단일 어시스턴스 응답으로 취급하는 대신, LangGraph가 내부적으로 사용하는 것과 같은 체크포인트, 노드 이름, 상태 키, 스트림 메타데이터를 노출할 수 있어요.
그래프 노드가 UI 카드에 어떻게 매핑되는가 (How graph nodes map to UI cards)
LangGraph 그래프는 각각 특정 작업을 담당하는 일련의 노드를 정의합니다. 예를 들어 조사 파이프라인은 다음과 같을 수 있어요:
- Classify: 사용자 쿼리 분류
- Research: 관련 정보 수집
- Analyze: 조사에서 결론 도출
- Synthesize: 최종 완성 응답 생성
각 노드는 그래프 상태의 특정 키에 출력을 씁니다. 프론트엔드에서는 그 매핑을 하드코딩할 필요가 없어요. useStream이 stream.subgraphs로 실행되는 각 노드를 발견하고, 관찰된 각 단계에 대한 SubgraphDiscoverySnapshot을 노출합니다:
// Nodes are discovered automatically — no hardcoded list needed
const graphNodes = [...stream.subgraphs.values()];
// Each snapshot carries the node name and current status
graphNodes.forEach((node) => {
console.log(node.nodeName, node.status); // "classify", "running"
});
진행 표시줄과 카드 헤더의 라벨에는 node.nodeName을 사용하세요. 각 스냅샷을 useMessages(stream, node)에 전달해, UI를 그래프 상태 키 이름에 결합하지 않고 노드 범위 스트리밍 콘텐츠를 렌더링할 수 있어요.
이 매핑은 그래프와 UI 사이의 계약이 됩니다. 백엔드 작성자는 의도적으로 노드를 추가·이름변경·재정렬할 수 있고, 프론트엔드 작성자는 각 상태 키를 상태 배지, 마크다운 패널, 표, 차트, 트레이스 뷰, 승인 카드 등으로 어떻게 시각화할지 결정합니다.
useStream 설정 (Setting up useStream)
useStream을 평소처럼 연결하세요. 사용할 주요 속성은 messages(대화용)와 subgraphs(현재 실행에서 발견된 그래프 노드용)입니다. 발견된 각 서브그래프 스냅샷을 셀렉터에 전달해 그 노드로 범위가 한정된 메시지를 읽습니다.
React:
import { useStream } from "@langchain/react";
const AGENT_URL = "http://localhost:2024";
export function PipelineChat() {
const stream = useStream<typeof myAgent>({
apiUrl: AGENT_URL,
assistantId: "graph_execution_cards",
});
const graphNodes = [...stream.subgraphs.values()];
return (
<div>
<PipelineProgress nodes={graphNodes} isLoading={stream.isLoading} />
<NodeCardList nodes={graphNodes} stream={stream} isLoading={stream.isLoading} />
</div>
);
}
Vue:
<script setup lang="ts">
import { useStream } from "@langchain/vue";
const AGENT_URL = "http://localhost:2024";
const stream = useStream<typeof myAgent>({
apiUrl: AGENT_URL,
assistantId: "graph_execution_cards",
});
</script>
<template>
<div>
<PipelineProgress
:nodes="[...stream.subgraphs.value.values()]"
:is-loading="stream.isLoading.value"
/>
<NodeCardList
:nodes="[...stream.subgraphs.value.values()]"
:stream="stream"
:is-loading="stream.isLoading.value"
/>
</div>
</template>
Svelte:
<script lang="ts">
import { useStream } from "@langchain/svelte";
const AGENT_URL = "http://localhost:2024";
const stream = useStream<typeof myAgent>({
apiUrl: AGENT_URL,
assistantId: "graph_execution_cards",
});
</script>
<div>
<PipelineProgress nodes={[...stream.subgraphs.values()]} isLoading={stream.isLoading} />
<NodeCardList
nodes={[...stream.subgraphs.values()]}
{stream}
isLoading={stream.isLoading}
/>
</div>
Angular:
import { Component, computed } from "@angular/core";
import { injectStream } from "@langchain/angular";
const AGENT_URL = "http://localhost:2024";
@Component({
selector: "app-pipeline-chat",
template: `
<div>
<app-pipeline-progress
[nodes]="graphNodes()"
[isLoading]="stream.isLoading()"
/>
<app-node-card-list
[nodes]="graphNodes()"
[stream]="stream"
[isLoading]="stream.isLoading()"
/>
</div>
`,
})
export class PipelineChatComponent {
stream = injectStream<typeof myAgent>({
apiUrl: AGENT_URL,
assistantId: "graph_execution_cards",
});
graphNodes = computed(() => [...this.stream.subgraphs().values()]);
}
스트리밍 토큰을 노드로 라우팅 (Routing streaming tokens to nodes)
그래프가 스트리밍되면 발견된 각 서브그래프 스냅샷이 그것이 속한 노드를 식별합니다. 그 스냅샷을 셀렉터 훅 또는 컴포저블에 전달해 그 노드로 범위가 한정된 메시지를 읽으세요:
import { AIMessage } from "langchain";
import { useMessages, type AnyStream, type SubgraphDiscoverySnapshot } from "@langchain/react";
function NodeCard({
node,
stream,
}: {
node: SubgraphDiscoverySnapshot;
stream: AnyStream;
}) {
const messages = useMessages(stream, node);
const lastAIMessage = messages.find(AIMessage.isInstance);
const streamingContent = lastAIMessage?.text ?? "";
return <NodeCardBody node={node} content={streamingContent} />;
}
첫 번째로 마운트된 셀렉터가 그 노드 네임스페이스에 대한 범위 구독(scoped subscription)을 엽니다. 노드 카드가 언마운트되면 구독은 자동으로 해제됩니다.
노드 상태 결정 (Determining node status)
발견된 각 노드는 현재 상태를 담고 있습니다. node.status를 직접 사용하세요. 발견 스냅샷은 "pending", "running", "complete", "error"를 보고합니다:
type NodeStatus = SubgraphDiscoverySnapshot["status"];
const status: NodeStatus = node.status;
파이프라인 진행 표시줄 만들기 (Building the pipeline progress bar)
상단의 가로 진행 표시줄은 전체 파이프라인을 한눈에 보여줍니다. 각 단계는 노드가 완료됨에 따라 채워지는 라벨 있는 세그먼트입니다:
function PipelineProgress({
nodes,
isLoading,
}: {
nodes: SubgraphDiscoverySnapshot[];
isLoading: boolean;
}) {
const firstIncompleteIdx = nodes.findIndex((node) => node.status !== "complete");
return (
<div className="flex items-center gap-1">
{nodes.map((node, i) => {
const isRunning =
isLoading && node.status !== "complete" && firstIncompleteIdx === i;
const colors = {
pending: "bg-gray-200 text-gray-500",
running: "bg-blue-400 text-white animate-pulse",
complete: "bg-green-500 text-white",
error: "bg-red-500 text-white",
};
const status = isRunning ? "running" : node.status;
return (
<div key={node.id} className="flex items-center">
<div
className={`rounded-full px-3 py-1 text-xs font-medium ${colors[status]}`}
>
{node.nodeName}
</div>
{i < nodes.length - 1 && (
<div
className={`mx-1 h-0.5 w-6 ${
status === "complete" ? "bg-green-500" : "bg-gray-200"
}`}
/>
)}
</div>
);
})}
</div>
);
}
접을 수 있는 NodeCard 컴포넌트 만들기 (Building collapsible NodeCard components)
각 노드는 상태 배지, 콘텐츠(스트리밍 또는 최종), 긴 출력을 위한 접을 수 있는 본문을 보여주는 자체 카드를 가집니다:
function NodeCard({
node,
stream,
}: {
node: SubgraphDiscoverySnapshot;
stream: AnyStream;
}) {
const [open, setOpen] = useState(node.status === "running");
const messages = useMessages(stream, node);
const lastAIMessage = messages.find(AIMessage.isInstance);
useEffect(() => {
if (node.status === "running") setOpen(true);
if (node.status === "complete") setOpen(false);
}, [node.status]);
return (
<div className="rounded-lg border bg-white shadow-sm">
<button
onClick={() => setOpen(!open)}
className="flex w-full items-center justify-between p-4"
>
<div className="flex items-center gap-3">
<h3 className="font-semibold">{node.nodeName}</h3>
<StatusBadge status={node.status} />
</div>
<span className={open ? "rotate-90" : ""}>▶</span>
</button>
{open && (
<div className="border-t px-4 py-3">
<div className="prose prose-sm max-w-none">
{lastAIMessage?.text?.trim()
? <Markdown>{lastAIMessage.text}</Markdown>
: <p className="italic text-gray-500">Processing...</p>}
</div>
</div>
)}
</div>
);
}
스트리밍 vs 완료된 콘텐츠 (Streaming vs. completed content)
노드 카드는 스트리밍과 최종 콘텐츠 모두에 범위 메시지를 읽습니다. 이는 그래프 노드 이름이 그 노드가 쓰는 상태 키와 일치한다고 가정하는 것을 피합니다(예: 플레이그라운드 그래프에서 do_research는 research에 씀):
| 소스 | 언제 사용 |
|---|---|
useMessages(stream, node) |
노드 범위 스트리밍·최종 메시지 렌더링 |
stream.values |
실제 상태 키를 사용해 최종 synthesis 필드 같은 전체 그래프 상태 읽기 |
패턴은 이렇습니다: 노드 카드에 가장 최근의 범위 AI 메시지를 보여주고, 의도적으로 그래프 상태 필드가 필요할 때만 stream.values를 사용합니다.
범위 메시지가 생산 노드에 묶여 있으므로, UI는 메시지 순서를 추측하지 않고도 병렬 그래프 경로를 지원할 수 있어요. 각 카드는 자체 노드에 속한 스트림 이벤트에서 갱신되고, 완료된 값은 stream.values를 통해 계속 사용할 수 있습니다.
function NodeContent({ stream, node }: { stream: AnyStream; node: SubgraphDiscoverySnapshot }) {
const messages = useMessages(stream, node);
const content = messages.find(AIMessage.isInstance)?.text ?? "";
return <Markdown>{content}</Markdown>;
}
모두 합치기 (Putting it all together)
라우팅, 상태 감지, 카드 렌더링을 결합한 전체 카드 목록입니다:
function NodeCardList({
nodes,
stream,
isLoading,
}: {
nodes: SubgraphDiscoverySnapshot[];
stream: AnyStream;
isLoading: boolean;
}) {
const firstIncompleteIdx = nodes.findIndex((node) => node.status !== "complete");
return (
<div className="space-y-3">
{nodes.map((node, i) => {
const isComplete = node.status === "complete";
const isRunning = isLoading && !isComplete && firstIncompleteIdx === i;
if (!isComplete && !isRunning) return null;
return <NodeCard key={node.id} node={node} stream={stream} />;
})}
</div>
);
}
사용 사례 (Use cases)
그래프 실행 카드는 가시성이 중요한 모든 다단계 파이프라인에서 잘 작동합니다:
- 조사 파이프라인: classify → 소스 수집 → analyze → 보고서 종합
- 콘텐츠 생성: 개요 → 초안 → 사실 확인 → 편집 → 게시
- 데이터 처리: 수집 → 검증 → 변환 → 집계 → 내보내기
- 코드 생성: 요구사항 이해 → 아키텍처 계획 → 코드 작성 → 리뷰 → 테스트
- 결정 워크플로우: 맥락 수집 → 옵션 평가 → 대안 채점 → 추천
동적 파이프라인 처리 (Handling dynamic pipelines)
모든 그래프가 고정된 노드 집합을 가지는 것은 아닙니다. 일부 파이프라인은 입력에 따라 노드를 추가하거나 건너뜁니다. 발견 맵에는 현재 스레드에서 관찰된 노드만 포함됩니다:
const activeNodes = [...stream.subgraphs.values()];
이렇게 하면 UI가 현재 실행과 관련된 노드에 대한 카드만 표시해, 빈 자리 표시자 카드를 피할 수 있어요.
모범 사례 (Best practices)
- 스트림에서 노드 발견. 예상 노드를 하드코딩하는 대신
stream.subgraphs에서 카드를 렌더링하세요. 조건부 또는 건너뛴 단계는 실행될 때까지 나타나지 않아요. - 상태 키를 UI 계약으로 취급. 프론트엔드가 렌더링할 만큼 안정적인 그래프 출력이 무엇인지 결정하고, 그 키를 그래프 정의 옆에 문서화하세요.
- 노드 카드에는 범위 메시지 사용. 스트리밍 중과 완료 후 모두 동작하며, UI 카드를 상태 키 이름에 결합하지 않습니다.
- 완료된 노드 자동 접기. 긴 파이프라인에서 완료된 카드는 자동으로 접어 사용자가 현재 활성 단계에 집중하게 하세요.
- 예상 시간 표시. 각 노드가 걸리는 시간에 대한 과거 데이터가 있으면 시간 추정을 표시해 사용자 기대를 설정하세요.
- 전역 진행 표시기 추가. 노드별 카드를 보완해 파이프라인 뷰 상단에 전체 진행 표시줄(예: "Step 2 of 4")을 추가하세요.
- 노드별 오류 처리. 노드가 실패하면 전체 파이프라인을 접지 않고 그 카드에 오류를 표시하세요. 다른 노드는 여전히 성공적으로 완료될 수 있어요.
더 알아보기 (Learn more)
- 프론트엔드 개요 — LangGraph를 프론트엔드로 렌더링.
- Custom stream channels — 커스텀 서버 측 데이터 스트리밍.
- LangChain 프론트엔드 패턴 — 모든 그래프에서 동작하는 UI 패턴.