LangGraphマルチエージェントがプロダクションで崩壊する理由とコードレベルでの復旧手法
makedream26 जुलाई 2026
0
Computing/SoftwareRelated Video
10:33ループエンジニアリングはもう古い！時代はグラフエンジニアリングへ
Chase AI
Comments (0)
Log in to leave a comment
No posts yet
makedream10:33makedream10:33Chase AI
Log in to leave a comment
No posts yet
単一のプロンプトチェーンを超えてマルチエージェントに移行したバックエンドエンジニアが陥りがちな勘違いがあります。プロンプトを改善すればシステムが安定する、という考えです。現場で発生する問題の大部分は、プロンプトとは何の関係もありません。状態汚染、無限ループ、APIのRate Limit、デバッグ不可能な非同期トレースといったシステム構造の問題こそが真の原因です。
LangGraphベースのマルチエージェントを運用環境へ投入するには、プロンプトのリソース集積としてではなく、バックエンドシステムを扱うかのように状態の隔離、コンカレンシー（並行性）制御、トレーシングをコードレベルで抑え込む必要があります。
状態ベースのグラフにおいて、複数のノードが単一の共有オブジェクトを直接変更すると、競合状態（Race Condition）が発生します。同時実行されるfan-outパターンにおいて、適切なリデューサー（Reducer）なしで通常のフィールドを上書きすると、最も遅く終了したノードの結果だけが残り、その他の結果は消失します。
これを防ぐには、上位グラフの状態には明示的なリデューサーを付与し、下位エージェントは独立したスキーマを持つサブグラフとして完全にカプセル化する必要があります。
`python
import operator
from typing import Annotated, List, TypedDict
from langgraph.graph import END, START, StateGraph
class ParentState(TypedDict):
task_id: str
input_query: str
audit_logs: Annotated[List[str], operator.add]
final_response: str
class InternalAgentState(TypedDict):
sub_task: str
scratchpad_messages: List[str]
sub_result: str
def internal_processing_node(state: InternalAgentState) -> dict:
updated_messages = state["scratchpad_messages"] + ["내부 격리 추론 진행 중"]
return {
"scratchpad_messages": updated_messages,
"sub_result": f"하위 작업 완료: {state['sub_task']}"
}
subgraph_builder = StateGraph(InternalAgentState)
subgraph_builder.add_node("internal_processing", internal_processing_node)
subgraph_builder.add_edge(START, "internal_processing")
subgraph_builder.add_edge("internal_processing", END)
compiled_subgraph = subgraph_builder.compile()
def call_isolated_subgraph_wrapper(state: ParentState) -> dict:
subgraph_input: InternalAgentState = {
"sub_task": state["input_query"],
"scratchpad_messages": []
}
subgraph_output = compiled_subgraph.invoke(subgraph_input)
return {
"audit_logs": [f"[서브그래프 결과]: {subgraph_output['sub_result']}"]
}
parent_builder = StateGraph(ParentState)
parent_builder.add_node("isolated_agent", call_isolated_subgraph_wrapper)
parent_builder.add_edge(START, "isolated_agent")
parent_builder.add_edge("isolated_agent", END)
main_graph = parent_builder.compile()
`
上位の ParentState において、並列書き込みが発生する audit_logs フィールドに operator.add リデューサーを指定しました。下位タスクは InternalAgentState という独自のステートを使用するサブグラフへと隔離し、ラッパー関数経由でのみ結果を授受します。データ汚染が遮断されることで、無限ループのデバッグ時間を週あたり5時間以上削減できます。
ReActフィードバックループが終了条件を満たせず、堂々巡りになる問題もよく見られます。ステートのスキーマにカウンターを設け、条件付きエッジでそれをフィルタリングするガードレールルーターが必要です。
`python
from typing import Literal, TypedDict
from langgraph.graph import END, START, StateGraph
class GuardedState(TypedDict):
query: str
draft: str
feedback: str
is_approved: bool
iterations: int
max_iterations: int
def drafting_node(state: GuardedState) -> dict:
return {
"draft": f"작성된 초안 (반복 회차: {state['iterations'] + 1})",
"iterations": state["iterations"] + 1
}
def review_node(state: GuardedState) -> dict:
approved = state["iterations"] >= 3
return {
"is_approved": approved,
"feedback": "승인 완료" if approved else "반려: 내용 수정 필요"
}
def loop_guardrail_router(state: GuardedState) -> Literal["drafting", "fallback_escalation", "end"]:
if state["is_approved"]:
return END
if state["iterations"] >= state["max_iterations"]:
return "fallback_escalation"
return "drafting"
def fallback_escalation_node(state: GuardedState) -> dict:
return {
"draft": "에이전트 검수 피드백 루프 최대 횟수 초과. 담당자 수동 검토 건으로 이관 처리되었습니다."
}
builder = StateGraph(GuardedState)
builder.add_node("drafting", drafting_node)
builder.add_node("review", review_node)
builder.add_node("fallback_escalation", fallback_escalation_node)
builder.add_edge(START, "drafting")
builder.add_edge("drafting", "review")
builder.add_conditional_edges(
"review",
loop_guardrail_router,
{
"drafting": "drafting",
"fallback_escalation": "fallback_escalation",
END: END
}
)
builder.add_edge("fallback_escalation", END)
guarded_graph = builder.compile()
`
カウンター（iterations）が設定値（max_iterations）に達した場合、即座に手動移管ノード（fallback_escalation）へ分岐するように設定しました。無限ループによるトークンの浪費を確実に防止できます。
下位ノードに一斉に並列リクエストを送信すると、OpenAIやAnthropic APIの1分あたりのトークン制限（TPM）に達し、HTTP 429エラーが発生します。バックエンドが揺らぐと、トランザクション全体が麻痺してしまいます。
asyncio.Semaphore で同時リクエスト数を制限し、tenacity の指数バックオフ（Exponential Backoff）を実装することでAPIの崩壊を防ぐことができます。
`python
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
API_SEMAPHORE = asyncio.Semaphore(5)
@retry(
stop=stop_after_attempt(5),
wait=wait_exponential(multiplier=1, min=2, max=10),
retry=retry_if_exception_type(Exception),
reraise=True
)
async def safe_llm_call_with_backoff(llm: ChatOpenAI, prompt: str) -> str:
async with API_SEMAPHORE:
response = await llm.ainvoke([HumanMessage(content=prompt)])
return response.content
async def parallel_worker_node(state: dict) -> dict:
llm = ChatOpenAI(model="gpt-4o", temperature=0)
task_input = state["task_data"]
result_text = await safe_llm_call_with_backoff(llm, f"하위 작업 처리: {task_input}")
return {"results": [result_text]}
`
同時呼び出しを最大5個に制限し、失敗時には2秒から10秒まで待機時間を倍増させながらリトライを行います。外部API呼び出しの失敗に伴うシステムの中断を完全に回避することが可能です。
単一ループの方式では1つのノードがエラーを起こすだけで全体の再推論に14秒以上を要しますが、このようにノードを隔離してバックオフを設定しておけば、失敗からの復旧時間を10ms程度にまで短縮できます。成功したノードの結果は維持されるため、不必要なトークンの再消費も防げます。
非同期で複雑に絡み合ったエージェントの挙動は、コンソール出力だけでは追跡できません。LangfuseなどのOpenTelemetryベースの観測プラットフォームを組み込み、DB操作やバックエンド処理といった非LLMロジックまでトレーシングスパンに含めることで、初めてボトルネックが可視化されます。
`python
import os
from langfuse.decorators import observe, langfuse_context
from langgraph.graph import StateGraph, START, END
@observe(name="vector_store_retrieval")
def query_vector_store(query: str) -> list:
langfuse_context.update_current_observation(
input={"query": query},
metadata={"top_k": 3, "database": "pgvector"}
)
return ["문서 1: 보안 규정 예시", "문서 2: 서비스 약관"]
def retrieval_node(state: dict) -> dict:
docs = query_vector_store(state["query"])
return {"context": docs}
def execute_graph_with_tracing(app, user_query: str, session_id: str, user_id: str):
from langfuse.callback import CallbackHandler
langfuse_handler = CallbackHandler()
config = {
"configurable": {"thread_id": session_id},
"callbacks": [langfuse_handler]
}
return app.invoke({"query": user_query}, config=config)
`
@observe デコレータを使用することで、vector search などの通常の関数もトレースに組み込むことができます。グラフの実行時に CallbackHandler を渡すことで、セッションごとのエージェント全体の挙동とトークン消費量を一目で確認できるようになります。
処理の途中で、例えば10番目のノードでエラーが発生した際、最初からやり直すのはコストと時間の双方で大きな損失となります。PostgresSaver チェックポインタを使用すれば、ノード実行のスナップショットがDBにそのまま保存されます。
`python
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import StateGraph
DATABASE_URL = "postgresql://postgres:postgres@localhost:5432/agent_checkpoints"
def build_app_graph():
builder = StateGraph(dict)
return builder
def resume_execution_from_failure(app, thread_id: str, fixed_payload: dict, last_valid_node: str):
config = {"configurable": {"thread_id": thread_id}}
app.update_state(
config,
values=fixed_payload,
as_node=last_valid_node
)
resumed_output = app.invoke(None, config)
return resumed_output
`
get_state_history で最後の正常な状態を確認した後、update_state で問題のあったデータを修正し、app.invoke(None, config) を呼び出すことで、正確に停止した時点から処理を再開できます。
すべてのエージェントノードにGPT-4oを適用するのは予算の無駄遣いです。メインのプランニングには高パフォーマンスなモデルを使用し、単純な分類や検証ノードにはClaude 3.5 Haikuのような軽量モデルを組み合わせるティアリング技法が基本となります。
さらにRedisVLベースのセマンティックキャッシュを導入すれば、同一または類似の検証リクエストに対してLLMを呼び出すことなく、わずか50msでレスポンスを返却できます。
`python
from redisvl.extensions.llmcache import SemanticCache
from langchain_community.chat_models import ChatAnthropic
audit_semantic_cache = SemanticCache(
name="audit_nodes_cache",
redis_url="redis://localhost:6379",
distance_threshold=0.1,
ttl=86400
)
def audit_verification_node(state: dict) -> dict:
prompt_query = f"다음 최종 결과물의 정책 준수 여부를 검수하세요: {state['final_response']}"
cached_response = audit_semantic_cache.check(prompt=prompt_query)
if cached_response:
return {
"audit_passed": cached_response[0]["response"] == "PASSED",
"audit_logs": ["[Audit Node]: 시맨틱 캐시 데이터 활용 (LLM 호출 스킵)"]
}
audit_llm = ChatAnthropic(model="claude-3-5-haiku-20241022", temperature=0)
eval_result = audit_llm.invoke(prompt_query).content
audit_semantic_cache.store(
prompt=prompt_query,
response=eval_result,
metadata={"node": "audit_verification"}
)
return {
"audit_passed": eval_result == "PASSED",
"audit_logs": [f"[Audit Node]: 신규 모델 검수 완료 ({eval_result})"]
}
`
distance_threshold を 0.1 と厳格に設定して誤検知を防ぎ、キャッシュに存在しない場合にのみ軽量モデルである Haiku を呼び出します。この構成を整えるだけでも、検証クオリティを維持しながらAPIトークンコストを最大40%削減することが可能です。
エージェントシステムをプロダクション環境へ展開する際に必要なのは、奇抜なプロンプトのテクニックではありません。状態隔離、コンカレンシー制御、チェックポイントからの復旧、モデルのティアリングといったバックエンドの堅牢な基礎設計こそが、システムの崩壊を防ぐカギとなります。