10장: 실전 프로젝트 -- Agentic Workflow 시스템 구축

customer-support-workflow/
  agents/
    triage.py         # 분류 에이전트
    researcher.py     # 조사 에이전트
    resolver.py       # 해결 에이전트
    qa.py             # 품질 검증 에이전트
  tools/
    customer_tools.py # 고객 관련 도구
    kb_tools.py       # 지식 베이스 도구
    ticket_tools.py   # 티켓 관련 도구
  workflow/
    graph.py          # LangGraph 워크플로우 정의
    state.py          # 상태 모델
    gates.py          # 승인 게이트
  infra/
    audit.py          # 감사 로깅
    security.py       # 보안 검증
    monitoring.py     # 모니터링
  config/
    permissions.py    # 에이전트 권한
    thresholds.py     # 임계값 설정
  main.py             # 진입점

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
 
class CustomerSupportState(TypedDict):
    """고객 지원 워크플로우 상태"""
 
    # 입력 데이터
    inquiry_id: str
    inquiry_text: str
    customer_id: str
    channel: str  # email, chat, api
 
    # 고객 컨텍스트
    customer_info: dict | None
    ticket_history: list[dict]
 
    # 분류 결과
    category: str | None
    subcategory: str | None
    priority: str | None
    confidence: float
 
    # 조사 결과
    relevant_docs: list[dict]
    similar_cases: list[dict]
    root_cause: str | None
 
    # 해결
    proposed_response: str | None
    response_approved: bool
    internal_notes: str | None
 
    # 워크플로우 제어
    requires_escalation: bool
    escalation_reason: str | None
    current_phase: str
    messages: Annotated[list, add_messages]
 
    # 감사
    audit_trail: list[dict]

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import SystemMessage, HumanMessage
 
TRIAGE_SYSTEM_PROMPT = """당신은 고객 지원 문의를 분류하는 전문가입니다.
 
역할:
- 고객 문의의 카테고리와 하위 카테고리를 분류합니다
- 우선순위(low, medium, high, urgent)를 판단합니다
- 즉시 에스컬레이션이 필요한 경우를 식별합니다
 
분류 카테고리:
- billing: 결제, 환불, 요금제 관련
- technical: 기술 문제, 버그 보고
- account: 계정 설정, 비밀번호 관련
- feature: 기능 요청, 개선 제안
- general: 일반 문의
 
에스컬레이션 기준:
- 법적 위협이 포함된 문의
- 데이터 유출/보안 관련 문의
- VIP 고객(enterprise 등급)의 긴급 문의
 
반드시 JSON 형식으로 응답하세요."""
 
class TriageAgent:
    def __init__(self):
        self.llm = ChatAnthropic(
            model="claude-sonnet-4-20250514",
            temperature=0,
        )
 
    async def classify(self, state: CustomerSupportState) -> dict:
        """문의 분류 수행"""
        customer_context = ""
        if state.get("customer_info"):
            tier = state["customer_info"].get("tier", "unknown")
            customer_context = f"\n고객 등급: {tier}"
 
        messages = [
            SystemMessage(content=TRIAGE_SYSTEM_PROMPT),
            HumanMessage(content=(
                f"고객 문의 내용:\n{state['inquiry_text']}"
                f"{customer_context}"
                f"\n\n이전 티켓 수: {len(state.get('ticket_history', []))}"
            )),
        ]
 
        response = await self.llm.ainvoke(messages)
        classification = parse_json_response(response.content)
 
        # 감사 로그 항목 생성
        audit_entry = {
            "phase": "triage",
            "action": "classify_inquiry",
            "result": classification,
            "model": "claude-sonnet-4-20250514",
            "timestamp": datetime.utcnow().isoformat(),
        }
 
        return {
            "category": classification["category"],
            "subcategory": classification.get("subcategory"),
            "priority": classification["priority"],
            "confidence": classification["confidence"],
            "requires_escalation": classification.get("requires_escalation", False),
            "escalation_reason": classification.get("escalation_reason"),
            "current_phase": "classified",
            "audit_trail": [audit_entry],
        }

RESEARCH_SYSTEM_PROMPT = """당신은 고객 문의 해결을 위한 조사 전문가입니다.
 
역할:
- 지식 베이스에서 관련 문서를 검색합니다
- 유사한 과거 케이스를 찾아 해결 방법을 참고합니다
- 근본 원인을 분석합니다
 
조사 결과를 구조화하여 제공하세요."""
 
class ResearcherAgent:
    def __init__(self, tools: list):
        self.llm = ChatAnthropic(
            model="claude-sonnet-4-20250514",
            temperature=0,
        ).bind_tools(tools)
        self.tools = {t.name: t for t in tools}
 
    async def investigate(self, state: CustomerSupportState) -> dict:
        """문의 조사 수행"""
        messages = [
            SystemMessage(content=RESEARCH_SYSTEM_PROMPT),
            HumanMessage(content=(
                f"카테고리: {state['category']}\n"
                f"문의 내용: {state['inquiry_text']}\n"
                f"고객 정보: {json.dumps(state.get('customer_info', {}), ensure_ascii=False)}"
            )),
        ]
 
        # 도구를 사용한 조사 루프
        relevant_docs = []
        similar_cases = []
        max_iterations = 5
 
        for _ in range(max_iterations):
            response = await self.llm.ainvoke(messages)
 
            if not response.tool_calls:
                break
 
            for tool_call in response.tool_calls:
                tool = self.tools[tool_call["name"]]
                result = await tool.ainvoke(tool_call["args"])
 
                if tool_call["name"] == "search_knowledge_base":
                    relevant_docs.extend(result.get("documents", []))
                elif tool_call["name"] == "find_similar_cases":
                    similar_cases.extend(result.get("cases", []))
 
                messages.append(response)
                messages.append(ToolMessage(
                    content=json.dumps(result, ensure_ascii=False),
                    tool_call_id=tool_call["id"],
                ))
 
        # 근본 원인 분석
        root_cause_response = await self.llm.ainvoke(messages + [
            HumanMessage(content="수집된 정보를 바탕으로 근본 원인을 분석해 주세요."),
        ])
 
        return {
            "relevant_docs": relevant_docs[:5],
            "similar_cases": similar_cases[:3],
            "root_cause": root_cause_response.content,
            "current_phase": "investigated",
            "audit_trail": [{
                "phase": "research",
                "action": "investigate",
                "docs_found": len(relevant_docs),
                "cases_found": len(similar_cases),
                "timestamp": datetime.utcnow().isoformat(),
            }],
        }

RESOLVER_SYSTEM_PROMPT = """당신은 고객 문의에 대한 답변을 작성하는 전문가입니다.
 
역할:
- 조사 결과를 바탕으로 정확하고 친절한 답변을 작성합니다
- 기술 용어를 고객이 이해하기 쉽게 설명합니다
- 해결 불가능한 경우 솔직히 안내하고 대안을 제시합니다
 
답변 작성 원칙:
- 공감 표현으로 시작합니다
- 문제의 원인을 간결하게 설명합니다
- 해결 방법을 단계별로 안내합니다
- 추가 도움이 필요하면 연락할 수 있음을 안내합니다"""
 
class ResolverAgent:
    def __init__(self):
        self.llm = ChatAnthropic(
            model="claude-sonnet-4-20250514",
            temperature=0.3,  # 약간의 창의성 허용
        )
 
    async def resolve(self, state: CustomerSupportState) -> dict:
        context = (
            f"고객 문의: {state['inquiry_text']}\n\n"
            f"카테고리: {state['category']}\n"
            f"근본 원인: {state.get('root_cause', '분석 중')}\n\n"
            f"관련 문서:\n"
        )
 
        for doc in state.get("relevant_docs", []):
            context += f"- {doc.get('title', '')}: {doc.get('excerpt', '')}\n"
 
        if state.get("similar_cases"):
            context += "\n유사 사례 해결 방법:\n"
            for case in state["similar_cases"]:
                context += f"- {case.get('resolution', '')}\n"
 
        messages = [
            SystemMessage(content=RESOLVER_SYSTEM_PROMPT),
            HumanMessage(content=context),
        ]
 
        response = await self.llm.ainvoke(messages)
 
        return {
            "proposed_response": response.content,
            "current_phase": "resolved",
            "audit_trail": [{
                "phase": "resolution",
                "action": "generate_response",
                "response_length": len(response.content),
                "timestamp": datetime.utcnow().isoformat(),
            }],
        }

QA_SYSTEM_PROMPT = """당신은 고객 답변의 품질을 검증하는 전문가입니다.
 
검증 항목:
1. 정확성: 기술적 내용이 정확한가?
2. 완전성: 고객의 문의에 충분히 답변했는가?
3. 톤: 공감적이고 전문적인 톤인가?
4. 안전성: 민감 정보 노출이나 부적절한 약속이 없는가?
 
각 항목을 1-5점으로 평가하고, 총점 16점 이상이면 승인, 미만이면 수정 요청을 하세요."""
 
class QAAgent:
    def __init__(self):
        self.llm = ChatAnthropic(
            model="claude-sonnet-4-20250514",
            temperature=0,
        )
 
    async def review(self, state: CustomerSupportState) -> dict:
        messages = [
            SystemMessage(content=QA_SYSTEM_PROMPT),
            HumanMessage(content=(
                f"원래 문의: {state['inquiry_text']}\n\n"
                f"제안된 답변:\n{state['proposed_response']}\n\n"
                f"참고한 근본 원인: {state.get('root_cause', '없음')}"
            )),
        ]
 
        response = await self.llm.ainvoke(messages)
        review_result = parse_json_response(response.content)
 
        total_score = sum(review_result.get("scores", {}).values())
        approved = total_score >= 16
 
        return {
            "response_approved": approved,
            "internal_notes": review_result.get("feedback", ""),
            "current_phase": "qa_complete" if approved else "qa_rejected",
            "audit_trail": [{
                "phase": "qa",
                "action": "review_response",
                "scores": review_result.get("scores", {}),
                "total_score": total_score,
                "approved": approved,
                "timestamp": datetime.utcnow().isoformat(),
            }],
        }

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
 
def create_workflow(db_url: str):
    """고객 지원 워크플로우 그래프 생성"""
 
    # 에이전트 인스턴스 생성
    triage = TriageAgent()
    researcher = ResearcherAgent(tools=research_tools)
    resolver = ResolverAgent()
    qa = QAAgent()
 
    # 노드 함수 정의
    async def lookup_customer(state: CustomerSupportState) -> dict:
        """고객 정보 조회"""
        info = await customer_db.lookup(state["customer_id"])
        history = await ticket_db.get_history(state["customer_id"], limit=10)
        return {"customer_info": info, "ticket_history": history}
 
    async def triage_node(state: CustomerSupportState) -> dict:
        return await triage.classify(state)
 
    async def research_node(state: CustomerSupportState) -> dict:
        return await researcher.investigate(state)
 
    async def resolve_node(state: CustomerSupportState) -> dict:
        return await resolver.resolve(state)
 
    async def qa_node(state: CustomerSupportState) -> dict:
        return await qa.review(state)
 
    async def send_response(state: CustomerSupportState) -> dict:
        """답변 발송"""
        await response_sender.send(
            channel=state["channel"],
            customer_id=state["customer_id"],
            response=state["proposed_response"],
        )
        return {"current_phase": "completed"}
 
    async def escalate(state: CustomerSupportState) -> dict:
        """사람에게 에스컬레이션"""
        await escalation_handler.escalate(
            inquiry_id=state["inquiry_id"],
            reason=state.get("escalation_reason", "에이전트 판단에 의한 에스컬레이션"),
            context={
                "inquiry": state["inquiry_text"],
                "customer": state.get("customer_info"),
                "category": state.get("category"),
                "audit_trail": state.get("audit_trail", []),
            },
        )
        return {"current_phase": "escalated"}
 
    # 라우팅 함수
    def route_after_triage(state: CustomerSupportState) -> str:
        if state.get("requires_escalation"):
            return "escalate"
        return "research"
 
    def route_after_qa(state: CustomerSupportState) -> str:
        if state.get("response_approved"):
            return "send_response"
        # 3회 이상 반려 시 에스컬레이션
        rejection_count = sum(
            1 for entry in state.get("audit_trail", [])
            if entry.get("phase") == "qa" and not entry.get("approved", True)
        )
        if rejection_count >= 3:
            return "escalate"
        return "resolve"
 
    # 그래프 구성
    graph = StateGraph(CustomerSupportState)
 
    graph.add_node("lookup_customer", lookup_customer)
    graph.add_node("triage", triage_node)
    graph.add_node("research", research_node)
    graph.add_node("resolve", resolve_node)
    graph.add_node("qa", qa_node)
    graph.add_node("send_response", send_response)
    graph.add_node("escalate", escalate)
 
    graph.set_entry_point("lookup_customer")
    graph.add_edge("lookup_customer", "triage")
    graph.add_conditional_edges("triage", route_after_triage, {
        "research": "research",
        "escalate": "escalate",
    })
    graph.add_edge("research", "resolve")
    graph.add_edge("resolve", "qa")
    graph.add_conditional_edges("qa", route_after_qa, {
        "send_response": "send_response",
        "resolve": "resolve",
        "escalate": "escalate",
    })
    graph.add_edge("send_response", END)
    graph.add_edge("escalate", END)
 
    # 체크포인터 연결
    checkpointer = PostgresSaver.from_conn_string(db_url)
    return graph.compile(checkpointer=checkpointer)

from langgraph.prebuilt import create_react_agent
 
class ApprovalGateManager:
    """승인 게이트 관리"""
 
    async def check_approval_needed(
        self, state: CustomerSupportState
    ) -> bool:
        """승인 필요 여부 판단"""
        # 고위험 카테고리
        if state.get("category") in ["billing", "account"]:
            if state.get("priority") in ["high", "urgent"]:
                return True
 
        # 신뢰도 낮은 경우
        if state.get("confidence", 0) < 0.80:
            return True
 
        # 엔터프라이즈 고객
        if state.get("customer_info", {}).get("tier") == "enterprise":
            return True
 
        return False
 
    async def request_approval(
        self, state: CustomerSupportState
    ) -> dict:
        """승인 요청 생성"""
        return {
            "inquiry_id": state["inquiry_id"],
            "proposed_response": state["proposed_response"],
            "category": state["category"],
            "priority": state["priority"],
            "confidence": state.get("confidence"),
            "customer_tier": state.get("customer_info", {}).get("tier"),
            "audit_summary": self._summarize_audit(state.get("audit_trail", [])),
        }

class WorkflowAuditLogger:
    """워크플로우 감사 로거"""
 
    def __init__(self, audit_store: ImmutableAuditLog):
        self.store = audit_store
 
    async def log_workflow_event(
        self,
        workflow_id: str,
        phase: str,
        action: str,
        detail: dict,
        actor: str = "agent",
    ) -> None:
        entry = AuditEntry(
            entry_id=str(uuid.uuid4()),
            workflow_id=workflow_id,
            trace_id=get_current_trace_id(),
            span_id=get_current_span_id(),
            timestamp=datetime.utcnow(),
            actor_type=actor,
            actor_id=f"cs-{phase}-agent",
            action_type=action,
            action_detail=detail,
            outcome="success",
        )
 
        await self.store.append(entry)
 
    async def log_human_decision(
        self,
        workflow_id: str,
        reviewer_id: str,
        decision: str,
        reason: str,
    ) -> None:
        entry = AuditEntry(
            entry_id=str(uuid.uuid4()),
            workflow_id=workflow_id,
            trace_id=get_current_trace_id(),
            span_id=get_current_span_id(),
            timestamp=datetime.utcnow(),
            actor_type="human",
            actor_id=reviewer_id,
            action_type="human_review",
            action_detail={"decision": decision, "reason": reason},
            outcome=decision,
        )
 
        await self.store.append(entry)

# 에이전트별 권한 설정
AGENT_PERMISSIONS = {
    "triage": AgentPermissionSet(
        agent_id="cs-triage-agent",
        role="triage",
        allowed_tools=["customer_lookup", "ticket_history"],
        resource_permissions=[
            ResourcePermission("customer_db", {Permission.READ}),
            ResourcePermission("ticket_db", {Permission.READ}),
        ],
        max_token_budget=50_000,
        max_tool_calls=20,
        allowed_models=["claude-sonnet-4-20250514"],
    ),
    "researcher": AgentPermissionSet(
        agent_id="cs-researcher-agent",
        role="researcher",
        allowed_tools=[
            "search_knowledge_base", "find_similar_cases",
            "customer_lookup",
        ],
        resource_permissions=[
            ResourcePermission("knowledge_base", {Permission.READ}),
            ResourcePermission("ticket_db", {Permission.READ}),
            ResourcePermission("customer_db", {Permission.READ}),
        ],
        max_token_budget=100_000,
        max_tool_calls=30,
        allowed_models=["claude-sonnet-4-20250514"],
    ),
    "resolver": AgentPermissionSet(
        agent_id="cs-resolver-agent",
        role="resolver",
        allowed_tools=["draft_response"],
        resource_permissions=[
            ResourcePermission("response_drafts", {Permission.READ, Permission.WRITE}),
        ],
        max_token_budget=80_000,
        max_tool_calls=10,
        allowed_models=["claude-sonnet-4-20250514"],
    ),
}

import asyncio
from workflow.graph import create_workflow
 
async def process_inquiry(inquiry: InquiryInput):
    """고객 문의 처리 진입점"""
    workflow = create_workflow(db_url=DATABASE_URL)
 
    initial_state = {
        "inquiry_id": inquiry.id,
        "inquiry_text": inquiry.text,
        "customer_id": inquiry.customer_id,
        "channel": inquiry.channel,
        "customer_info": None,
        "ticket_history": [],
        "category": None,
        "subcategory": None,
        "priority": None,
        "confidence": 0.0,
        "relevant_docs": [],
        "similar_cases": [],
        "root_cause": None,
        "proposed_response": None,
        "response_approved": False,
        "internal_notes": None,
        "requires_escalation": False,
        "escalation_reason": None,
        "current_phase": "started",
        "messages": [],
        "audit_trail": [],
    }
 
    config = {
        "configurable": {
            "thread_id": f"inquiry-{inquiry.id}",
        },
    }
 
    result = await workflow.ainvoke(initial_state, config)
    return result
 
async def main():
    # Kafka 소비자로 문의 이벤트 수신
    consumer = WorkflowEventConsumer(process_inquiry)
    await consumer.start()
 
if __name__ == "__main__":
    asyncio.run(main())