Flow Engineering: LLM 워크플로우부터 조직 아키텍처까지, 흐름을 설계하는 방법

from langgraph.graph import StateGraph, END
from typing import TypedDict
 
# 실행 전 초기화 필요:
# from langchain_openai import ChatOpenAI
# from langchain_community.tools import DuckDuckGoSearchRun
# llm = ChatOpenAI(model="gpt-4o")
# search_web = DuckDuckGoSearchRun().invoke
 
class ResearchState(TypedDict):
    query: str
    search_results: list[str]
    draft: str
    review_feedback: str
    final_answer: str
 
def search_node(state: ResearchState) -> ResearchState:
    results = search_web(state["query"])
    return {**state, "search_results": results}
 
def draft_node(state: ResearchState) -> ResearchState:
    draft = llm.invoke(f"다음 자료로 답변을 작성하세요: {state['search_results']}")
    return {**state, "draft": draft.content}
 
def review_node(state: ResearchState) -> ResearchState:
    # 자기 검증 — 초안을 비판적으로 검토
    feedback = llm.invoke(f"이 답변의 문제점을 찾아주세요: {state['draft']}")
    return {**state, "review_feedback": feedback.content}
 
def revise_node(state: ResearchState) -> ResearchState:
    final = llm.invoke(
        f"피드백을 반영해 개선하세요.\n초안: {state['draft']}\n피드백: {state['review_feedback']}"
    )
    return {**state, "final_answer": final.content}
 
graph = StateGraph(ResearchState)
graph.add_node("search", search_node)
graph.add_node("draft", draft_node)
graph.add_node("review", review_node)
graph.add_node("revise", revise_node)
 
graph.set_entry_point("search")
graph.add_edge("search", "draft")
graph.add_edge("draft", "review")
graph.add_edge("review", "revise")
graph.add_edge("revise", END)
 
app = graph.compile()

Flow Items:
├── Feature  → 새로운 비즈니스 가치 창출
├── Defect   → 품질 문제 수정
├── Risk     → 보안, 컴플라이언스, 아키텍처 부채
└── Debt     → 기술 부채 해소

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
 
# 초기화 예시:
# from langchain_openai import ChatOpenAI
# from langchain_community.tools import DuckDuckGoSearchRun
# llm = ChatOpenAI(model="gpt-4o")
# search_tool = DuckDuckGoSearchRun()
 
class MultiAgentState(TypedDict):
    task: str
    task_analysis: str      # 오케스트레이터가 생성한 작업 분석
    search_result: str
    code_result: str
    validation_result: str
    final_output: str
    errors: Annotated[list[str], operator.add]  # 병렬 노드에서 안전하게 누적
 
def orchestrator(state: MultiAgentState) -> dict:
    """작업을 분석하고 서브에이전트가 공유할 컨텍스트를 생성"""
    analysis = llm.invoke(
        f"다음 작업을 분석하고 핵심 요구사항을 요약하세요: {state['task']}"
    )
    return {"task_analysis": analysis.content}
 
def search_agent(state: MultiAgentState) -> dict:
    """검색 전문 에이전트 — 오케스트레이터 분석을 컨텍스트로 활용"""
    result = search_tool.invoke(
        f"{state['task_analysis']}\n\n검색 쿼리: {state['task']}"
    )
    return {"search_result": result}
 
def code_agent(state: MultiAgentState) -> dict:
    """코드 생성 전문 에이전트"""
    result = llm.invoke(
        f"분석: {state['task_analysis']}\n\n요구사항에 맞는 코드를 작성하세요: {state['task']}"
    )
    return {"code_result": result.content}
 
def validation_agent(state: MultiAgentState) -> dict:
    """검색과 코드 결과를 모두 받은 뒤 실행 — 두 브랜치가 완료되어야 진입"""
    combined = f"검색: {state['search_result']}\n코드: {state['code_result']}"
    validation = llm.invoke(f"다음 결과를 검증하세요: {combined}")
    return {"validation_result": validation.content}
 
def editor_agent(state: MultiAgentState) -> dict:
    """최종 통합"""
    final = llm.invoke(
        f"""다음 결과를 통합하여 최종 답변을 작성하세요:
        - 작업 분석: {state['task_analysis']}
        - 검색 결과: {state['search_result']}
        - 코드 결과: {state['code_result']}
        - 검증 결과: {state['validation_result']}"""
    )
    return {"final_output": final.content}
 
builder = StateGraph(MultiAgentState)
builder.add_node("orchestrator", orchestrator)
builder.add_node("search", search_agent)
builder.add_node("code", code_agent)
builder.add_node("validation", validation_agent)
builder.add_node("editor", editor_agent)
 
builder.set_entry_point("orchestrator")
 
# 팬아웃: orchestrator → search, code 두 노드가 같은 슈퍼스텝에서 실행
builder.add_edge("orchestrator", "search")
builder.add_edge("orchestrator", "code")
 
# 팬인: search, code 모두 완료된 슈퍼스텝 이후에 validation 진입
builder.add_edge("search", "validation")
builder.add_edge("code", "validation")
builder.add_edge("validation", "editor")
builder.add_edge("editor", END)
 
multi_agent_app = builder.compile()

interface FlowItem {
  id: string;
  type: "feature" | "defect" | "risk" | "debt";
  requestedAt: Date;
  startedAt: Date | null;
  completedAt: Date | null;
}
 
interface FlowMetrics {
  flowTime: number;       // 요청부터 완료까지 (일)
  waitTime: number;       // 요청부터 시작까지 (일)
  activeTime: number;     // 실제 작업 시간 (일)
  flowEfficiency: number; // activeTime / flowTime × 100 (%)
}
 
function calculateFlowMetrics(item: FlowItem): FlowMetrics | null {
  if (!item.startedAt || !item.completedAt) return null;
 
  const msPerDay = 1000 * 60 * 60 * 24;
  const flowTime = (item.completedAt.getTime() - item.requestedAt.getTime()) / msPerDay;
  const waitTime = (item.startedAt.getTime() - item.requestedAt.getTime()) / msPerDay;
  const activeTime = (item.completedAt.getTime() - item.startedAt.getTime()) / msPerDay;
  const flowEfficiency = (activeTime / flowTime) * 100;
 
  return { flowTime, waitTime, activeTime, flowEfficiency };
}
 
function analyzeBottleneck(items: FlowItem[]): void {
  const metrics = items
    .map((item) => ({ item, metrics: calculateFlowMetrics(item) }))
    .filter((r) => r.metrics !== null);
 
  const avgWaitTime =
    metrics.reduce((sum, r) => sum + r.metrics!.waitTime, 0) / metrics.length;
  const avgEfficiency =
    metrics.reduce((sum, r) => sum + r.metrics!.flowEfficiency, 0) / metrics.length;
 
  console.log(`평균 대기 시간: ${avgWaitTime.toFixed(1)}일`);
  console.log(`평균 흐름 효율성: ${avgEfficiency.toFixed(1)}%`);
 
  // Mik Kersten의 연구에서 지식 노동(knowledge work)의 Flow Efficiency는
  // 통상 15~40% 범위이며, 15% 미만은 대기 시간이 압도적으로 많다는 신호다.
  if (avgEfficiency < 15) {
    console.log("⚠️  병목 감지: 실제 작업보다 대기 시간이 압도적으로 많습니다.");
    console.log("→ 기획·QA·리뷰 프로세스를 점검해보시면 좋습니다.");
  }
}