LLM 에이전트 백엔드 설계 패턴: ReAct, LangGraph, Temporal로 프로덕션 버티기

# agent/core.py — ReAct 루프의 단순화된 개념 구조
class MaxIterationsExceeded(Exception):
    pass
 
async def react_loop(
    user_input: str,
    tools: list[Tool],
    max_iterations: int = 10
) -> str:
    messages = [{"role": "user", "content": user_input}]
    
    for _ in range(max_iterations):
        # Thought: LLM이 다음 행동을 추론
        response = await llm.invoke(messages, tools=tools)
        
        if response.finish_reason == "stop":
            return response.content
        
        # Action: 도구 호출
        for call in response.tool_calls:
            # Observation: 도구 실행 결과를 컨텍스트에 추가
            result = await execute_tool(call.name, call.args)
            messages.append({"role": "tool", "content": result})
        
        messages.append(response)
    
    raise MaxIterationsExceeded(f"{max_iterations}회 반복 초과, 강제 종료")

# agent/memory.py
class AgentMemory:
    def __init__(self):
        # 실제 환경에서는 연결 설정 필요:
        # Redis(host="localhost", port=6379, decode_responses=True)
        self.short_term = Redis()        # 세션 컨텍스트, 빠른 읽기
        self.long_term = PGVector()      # 임베딩 기반 장기 기억
        self.checkpoint = Postgres()     # 실패 복구용 내구성 저장소
    
    async def retrieve_context(self, query: str, session_id: str) -> dict:
        # 단기: 현재 세션의 최근 메시지
        recent = await self.short_term.get(f"session:{session_id}")
        
        # 장기: 쿼리와 의미적으로 유사한 과거 기억 k=5개 검색
        # 검색 결과는 시스템 프롬프트 직후, 사용자 메시지 앞에 주입
        relevant = await self.long_term.similarity_search(query, k=5)
        
        return {"recent": recent, "relevant": relevant}

# mcp_server/search_server.py — FastMCP로 간단한 MCP 서버 구현 예시
from mcp.server.fastmcp import FastMCP
 
mcp = FastMCP("search-server")
 
@mcp.tool()
async def web_search(query: str) -> str:
    """웹에서 최신 정보를 검색합니다."""
    results = await search_api.search(query)
    return results.format_as_text()

# agent/api.py — FastAPI 엔드포인트
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from redis.asyncio import Redis
import asyncio, json
 
app = FastAPI()
redis = Redis(host="localhost", port=6379, decode_responses=True)
 
@app.post("/agent/task")
async def submit_task(request: TaskRequest) -> dict:
    task_id = generate_task_id()
    
    # 즉시 큐에 푸시하고 task_id만 반환 — 에이전트 실행은 Worker가 담당
    await redis.lpush("agent:queue", json.dumps({
        "task_id": task_id,
        "input": request.input,
        "tools": request.tools
    }))
    
    return {"task_id": task_id, "status": "queued"}
 
@app.get("/agent/task/{task_id}/stream")
async def stream_result(task_id: str):
    # SSE로 에이전트 진행 상황 스트리밍
    # 프로덕션에서는 타임아웃(30s 등)과 연결 종료 처리도 추가 필요
    async def event_generator():
        pubsub = redis.pubsub()
        await pubsub.subscribe(f"agent:result:{task_id}")
        try:
            async for message in pubsub.listen():
                if message["type"] == "message":
                    yield f"data: {message['data']}\n\n"
                    if json.loads(message["data"]).get("done"):
                        break
        finally:
            await pubsub.unsubscribe(f"agent:result:{task_id}")
    
    return StreamingResponse(event_generator(), media_type="text/event-stream")

# agent/worker.py — 실제 에이전트 루프 실행 프로세스
async def agent_worker():
    while True:
        # brpop: 큐가 비어있으면 블로킹, 태스크 도착 시 즉시 처리
        _, task_data = await redis.brpop("agent:queue")
        task = json.loads(task_data)
        
        async for step_result in run_agent_loop(task):
            await redis.publish(
                f"agent:result:{task['task_id']}",
                json.dumps(step_result)
            )

# agent/graph.py — LangGraph 멀티 에이전트 파이프라인
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
 
# TypedDict: 상태 타입 명시, Annotated[list, operator.add]는 각 노드의 반환값을 리스트에 추가(append)하는 리듀서 선언
class AgentState(TypedDict):
    query: str
    retrieved_docs: list[str]
    draft_answer: str
    quality_score: float
    messages: Annotated[list, operator.add]
 
async def retriever_node(state: AgentState) -> AgentState:
    docs = await vector_db.similarity_search(state["query"])
    return {"retrieved_docs": docs}
 
async def summarizer_node(state: AgentState) -> AgentState:
    draft = await llm.invoke(
        f"다음 문서를 요약해 답변을 작성하세요:\n{state['retrieved_docs']}"
    )
    return {"draft_answer": draft}
 
async def critic_node(state: AgentState) -> AgentState:
    # LLM-as-judge 패턴: 별도 LLM 호출로 답변 품질을 0~1 점수로 평가
    # 프롬프트에 평가 기준(관련성, 완성도, 사실 정확성)을 명시하고 JSON으로 파싱
    score = await evaluate_quality(state["draft_answer"], state["query"])
    return {"quality_score": score}
 
def route_after_critic(state: AgentState) -> str:
    if state["quality_score"] < 0.7:
        return "retriever"  # 품질 미달 → 재질의
    return END
 
workflow = StateGraph(AgentState)
workflow.add_node("retriever", retriever_node)
workflow.add_node("summarizer", summarizer_node)
workflow.add_node("critic", critic_node)
 
workflow.set_entry_point("retriever")
workflow.add_edge("retriever", "summarizer")
workflow.add_edge("summarizer", "critic")
workflow.add_conditional_edges("critic", route_after_critic)
 
# redis_checkpointer: 각 노드 실행 결과를 Redis에 저장해 중간 실패 시 재개 가능
agent = workflow.compile(checkpointer=redis_checkpointer)

# agent/temporal_workflow.py
from temporalio import workflow, activity
from temporalio.common import RetryPolicy
from datetime import timedelta
from dataclasses import dataclass
 
@dataclass
class LLMResponse:
    content: str
    is_final: bool
    tool_name: str | None = None
    tool_args: dict | None = None
 
@activity.defn
async def call_llm(prompt: str) -> LLMResponse:
    raw = await llm.invoke(prompt)
    # LLM 응답을 파싱해 구조화된 LLMResponse로 변환
    return parse_llm_response(raw)
 
@activity.defn
async def execute_tool(tool_name: str, args: dict) -> str:
    return await tool_registry.execute(tool_name, args)
 
@workflow.defn
class AgentWorkflow:
    @workflow.run
    async def run(self, user_input: str) -> str:
        messages = [user_input]
        
        for _ in range(10):  # 무한 루프 방지
            # 각 Activity는 실패 시 자동 재시도, 결과는 영속적으로 저장
            response: LLMResponse = await workflow.execute_activity(
                call_llm,
                args=[str(messages)],
                start_to_close_timeout=timedelta(seconds=30),
                retry_policy=RetryPolicy(maximum_attempts=3)
            )
            
            if response.is_final:
                return response.content
            
            tool_result = await workflow.execute_activity(
                execute_tool,
                args=[response.tool_name, response.tool_args],
                start_to_close_timeout=timedelta(seconds=60)
            )
            messages.append(tool_result)
        
        return "최대 반복 횟수 초과"