2장: 에이전트 기반 비즈니스 프로세스 설계

interface TaskDefinition {
  id: string
  name: string
  description: string
  inputs: Record<string, SchemaType>
  outputs: Record<string, SchemaType>
  requiredTools: string[]
  estimatedDuration: Duration
  failureStrategy: 'retry' | 'fallback' | 'escalate'
}
 
// 예시: 고객 문의 분류 태스크
const classifyInquiry: TaskDefinition = {
  id: 'classify-inquiry',
  name: '고객 문의 분류',
  description: '고객 문의 내용을 분석하여 카테고리와 우선순위를 결정',
  inputs: {
    inquiryText: 'string',
    customerHistory: 'CustomerHistory',
  },
  outputs: {
    category: 'InquiryCategory',
    priority: 'PriorityLevel',
    confidence: 'number',
  },
  requiredTools: ['customer-db-lookup', 'category-classifier'],
  estimatedDuration: { seconds: 5 },
  failureStrategy: 'retry',
}

from dataclasses import dataclass
 
@dataclass
class AgentRole:
    name: str
    persona: str  # 시스템 프롬프트로 주입
    tools: list[str]  # 허용된 도구 목록
    permissions: PermissionSet
    escalation_target: str  # 에스컬레이션 대상
 
# 고객 지원 워크플로우의 에이전트 역할 정의
triage_agent = AgentRole(
    name="triage",
    persona="당신은 고객 문의를 분류하는 전문가입니다. "
            "문의 내용을 분석하여 카테고리, 우선순위, "
            "긴급도를 판단합니다.",
    tools=["customer_lookup", "ticket_history", "category_classifier"],
    permissions=PermissionSet(read=["customer_db"], write=["ticket_db"]),
    escalation_target="human_supervisor",
)
 
resolver_agent = AgentRole(
    name="resolver",
    persona="당신은 기술 지원 전문가입니다. "
            "분류된 문의에 대해 해결책을 조사하고 "
            "고객에게 답변을 작성합니다.",
    tools=["knowledge_base", "doc_search", "draft_response"],
    permissions=PermissionSet(read=["kb_db", "doc_store"], write=["response_draft"]),
    escalation_target="senior_engineer",
)

// MCP 기반 도구 스키마 정의
const customerLookupTool = {
  name: 'customer_lookup',
  description: '고객 ID 또는 이메일로 고객 정보를 조회합니다',
  inputSchema: {
    type: 'object',
    properties: {
      customerId: { type: 'string', description: '고객 고유 ID' },
      email: { type: 'string', format: 'email', description: '고객 이메일' },
    },
    oneOf: [
      { required: ['customerId'] },
      { required: ['email'] },
    ],
  },
  outputSchema: {
    type: 'object',
    properties: {
      id: { type: 'string' },
      name: { type: 'string' },
      tier: { type: 'string', enum: ['free', 'pro', 'enterprise'] },
      history: { type: 'array', items: { type: 'object' } },
    },
  },
}

from langgraph.graph import StateGraph, END
from typing import TypedDict
 
class WorkflowState(TypedDict):
    inquiry: str
    customer_info: dict | None
    analysis: dict | None
    priority: str | None
    resolution: str | None
    status: str
 
def receive_inquiry(state: WorkflowState) -> WorkflowState:
    """문의 접수 및 초기 상태 설정"""
    return {**state, "status": "received"}
 
def lookup_customer(state: WorkflowState) -> WorkflowState:
    """고객 정보 조회"""
    info = customer_db.lookup(state["inquiry"])
    return {**state, "customer_info": info}
 
def analyze_inquiry(state: WorkflowState) -> WorkflowState:
    """문의 내용 분석 및 분류"""
    analysis = triage_agent.analyze(state["inquiry"])
    return {**state, "analysis": analysis}
 
def route_by_priority(state: WorkflowState) -> str:
    """우선순위에 따른 라우팅"""
    if state["analysis"]["priority"] == "urgent":
        return "escalate"
    return "auto_resolve"
 
# 그래프 구성
graph = StateGraph(WorkflowState)
graph.add_node("receive", receive_inquiry)
graph.add_node("lookup", lookup_customer)
graph.add_node("analyze", analyze_inquiry)
graph.add_node("resolve", auto_resolve)
graph.add_node("escalate", escalate_to_human)
 
graph.set_entry_point("receive")
graph.add_edge("receive", "lookup")
graph.add_edge("receive", "analyze")
graph.add_conditional_edges("analyze", route_by_priority, {
    "auto_resolve": "resolve",
    "escalate": "escalate",
})
graph.add_edge("resolve", END)
graph.add_edge("escalate", END)
 
workflow = graph.compile()

from temporalio import workflow, activity
from datetime import timedelta
 
@activity.defn
async def classify_inquiry(inquiry_text: str) -> Classification:
    """문의 분류 액티비티"""
    return await triage_agent.classify(inquiry_text)
 
@activity.defn
async def generate_resolution(context: ResolutionContext) -> str:
    """해결책 생성 액티비티"""
    return await resolver_agent.resolve(context)
 
@workflow.defn
class CustomerSupportWorkflow:
    @workflow.run
    async def run(self, inquiry: InquiryInput) -> WorkflowResult:
        # 분류 (5초 타임아웃)
        classification = await workflow.execute_activity(
            classify_inquiry,
            inquiry.text,
            start_to_close_timeout=timedelta(seconds=30),
        )
 
        if classification.priority == "urgent":
            # 긴급: 즉시 에스컬레이션
            await workflow.execute_activity(
                escalate_to_human,
                EscalationRequest(inquiry=inquiry, reason="urgent"),
                start_to_close_timeout=timedelta(minutes=5),
            )
            return WorkflowResult(status="escalated")
 
        # 일반: 자동 해결 시도
        resolution = await workflow.execute_activity(
            generate_resolution,
            ResolutionContext(inquiry=inquiry, classification=classification),
            start_to_close_timeout=timedelta(minutes=2),
        )
 
        return WorkflowResult(status="resolved", resolution=resolution)