AI 워크플로 자동화: 비즈니스 프로세스 자동화 가이드
AI 워크플로 자동화는 인공지능을 활용해 반복적인 비즈니스 프로세스를 자동화하는 것입니다. 이 가이드에서는 엔터프라이즈 자동화 전략을 살펴봅니다.
자동화 범주
1. 문서 처리
- 송장 처리
- 계약 분석
- 양식 데이터 추출
2. 커뮤니케이션 자동화
- 이메일 분류
- 자동 응답
- 티켓 라우팅
3. 데이터 처리
- 데이터 추출
- 데이터 변환
- 리포팅
4. 의사결정 자동화
- 승인 프로세스
- 리스크 평가
- 이상 탐지
문서 처리 파이프라인
송장 처리
1from openai import OpenAI 2import base64 3 4client = OpenAI() 5 6def process_invoice(image_path: str) -> dict: 7 with open(image_path, "rb") as f: 8 image_data = base64.b64encode(f.read()).decode() 9 10 response = client.chat.completions.create( 11 model="gpt-4-vision-preview", 12 response_format={"type": "json_object"}, 13 messages=[ 14 { 15 "role": "system", 16 "content": """Extract information from the invoice: 17 { 18 "vendor": "string", 19 "invoice_number": "string", 20 "date": "YYYY-MM-DD", 21 "total": number, 22 "tax": number, 23 "items": [{"description": "string", "quantity": number, "price": number}] 24 }""" 25 }, 26 { 27 "role": "user", 28 "content": [ 29 {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_data}"}} 30 ] 31 } 32 ] 33 ) 34 35 return json.loads(response.choices[0].message.content)
계약 분석
1def analyze_contract(contract_text: str) -> dict: 2 response = client.chat.completions.create( 3 model="gpt-4-turbo", 4 response_format={"type": "json_object"}, 5 messages=[ 6 { 7 "role": "system", 8 "content": """Analyze the contract and extract the following information: 9 { 10 "parties": ["string"], 11 "effective_date": "YYYY-MM-DD", 12 "termination_date": "YYYY-MM-DD", 13 "key_obligations": ["string"], 14 "payment_terms": "string", 15 "risks": ["string"], 16 "unusual_clauses": ["string"] 17 }""" 18 }, 19 {"role": "user", "content": contract_text} 20 ] 21 ) 22 23 return json.loads(response.choices[0].message.content) 24## 이메일 자동화 25 26### 이메일 분류 27 28```python 29def classify_email(email_content: str, categories: list) -> dict: 30 response = client.chat.completions.create( 31 model="gpt-4-turbo", 32 response_format={"type": "json_object"}, 33 messages=[ 34 { 35 "role": "system", 36 "content": f"""Classify the email: 37 Categories: {categories} 38 39 Response format: 40 {{ 41 "category": "string", 42 "confidence": 0.0-1.0, 43 "priority": "high|medium|low", 44 "sentiment": "positive|neutral|negative", 45 "requires_response": boolean, 46 "summary": "string" 47 }}""" 48 }, 49 {"role": "user", "content": email_content} 50 ] 51 ) 52 53 return json.loads(response.choices[0].message.content) 54 55# Usage 56categories = ["sales_inquiry", "support_request", "complaint", "partnership", "spam"] 57result = classify_email(email_body, categories)
자동 응답 생성
1def generate_auto_reply(email: dict, context: str = "") -> str: 2 response = client.chat.completions.create( 3 model="gpt-4-turbo", 4 messages=[ 5 { 6 "role": "system", 7 "content": f"""Generate a professional auto-reply. 8 Tone: Friendly but professional 9 Language: English 10 11 Company info: {context} 12 13 Email category: {email['category']} 14 Priority: {email['priority']} 15 """ 16 }, 17 {"role": "user", "content": f"Email content:\n{email['content']}\n\nSummary: {email['summary']}"} 18 ] 19 ) 20 21 return response.choices[0].message.content
워크플로 오케스트레이션
순차 파이프라인
1class WorkflowPipeline: 2 def __init__(self): 3 self.steps = [] 4 5 def add_step(self, name: str, function: callable, config: dict = None): 6 self.steps.append({ 7 "name": name, 8 "function": function, 9 "config": config or {} 10 }) 11 12 def execute(self, input_data: dict) -> dict: 13 result = input_data 14 15 for step in self.steps: 16 print(f"Executing: {step['name']}") 17 try: 18 result = step["function"](result, **step["config"]) 19 except Exception as e: 20 return {"error": str(e), "failed_step": step["name"]} 21 22 return result 23 24# Pipeline definition 25pipeline = WorkflowPipeline() 26pipeline.add_step("extract", extract_data) 27pipeline.add_step("validate", validate_data) 28pipeline.add_step("transform", transform_data) 29pipeline.add_step("load", load_to_database) 30 31result = pipeline.execute({"file_path": "invoice.pdf"})
조건부 워크플로
1class ConditionalWorkflow: 2 def __init__(self): 3 self.routes = {} 4 self.default_route = None 5 6 def add_route(self, condition: callable, handler: callable): 7 self.routes[condition] = handler 8 9 def set_default(self, handler: callable): 10 self.default_route = handler 11 12 def execute(self, data: dict) -> dict: 13 for condition, handler in self.routes.items(): 14 if condition(data): 15 return handler(data) 16 17 if self.default_route: 18 return self.default_route(data) 19 20 raise ValueError("No matching route found") 21 22# Usage 23workflow = ConditionalWorkflow() 24workflow.add_route( 25 lambda x: x["category"] == "support", 26 create_support_ticket 27) 28workflow.add_route( 29 lambda x: x["category"] == "sales", 30 forward_to_sales 31) 32workflow.set_default(archive_email) 33## 데이터 추출 34 35### 테이블 추출 36 37```python 38def extract_table_from_document(document_path: str) -> list: 39 # Convert document to image or parse PDF 40 images = convert_to_images(document_path) 41 42 all_tables = [] 43 for img in images: 44 response = client.chat.completions.create( 45 model="gpt-4-vision-preview", 46 response_format={"type": "json_object"}, 47 messages=[ 48 { 49 "role": "system", 50 "content": "Extract tables from image as JSON array. Each table should be a list of dicts." 51 }, 52 { 53 "role": "user", 54 "content": [ 55 {"type": "image_url", "image_url": {"url": img}} 56 ] 57 } 58 ] 59 ) 60 61 tables = json.loads(response.choices[0].message.content) 62 all_tables.extend(tables.get("tables", [])) 63 64 return all_tables
엔티티 추출
1def extract_entities(text: str, entity_types: list) -> dict: 2 response = client.chat.completions.create( 3 model="gpt-4-turbo", 4 response_format={"type": "json_object"}, 5 messages=[ 6 { 7 "role": "system", 8 "content": f"""Extract the following entities from text: {entity_types} 9 10 Format: 11 {{ 12 "entities": [ 13 {{"type": "string", "value": "string", "confidence": 0.0-1.0}} 14 ] 15 }}""" 16 }, 17 {"role": "user", "content": text} 18 ] 19 ) 20 21 return json.loads(response.choices[0].message.content) 22 23# Usage 24entities = extract_entities( 25 "John Doe paid $5,000 on January 15, 2024.", 26 ["PERSON", "DATE", "MONEY"] 27)
승인 워크플로우
1class ApprovalWorkflow: 2 def __init__(self, rules: list): 3 self.rules = rules 4 5 def evaluate(self, request: dict) -> dict: 6 for rule in self.rules: 7 if rule["condition"](request): 8 return { 9 "decision": rule["action"], 10 "rule_name": rule["name"], 11 "approvers": rule.get("approvers", []), 12 "reason": rule.get("reason", "") 13 } 14 15 return {"decision": "manual_review", "reason": "No matching rule"} 16 17 def ai_evaluate(self, request: dict) -> dict: 18 """Complex decisions with LLM""" 19 response = client.chat.completions.create( 20 model="gpt-4-turbo", 21 response_format={"type": "json_object"}, 22 messages=[ 23 { 24 "role": "system", 25 "content": f"""Approval rules: {json.dumps(self.rules)} 26 27 Evaluate the request and decide: 28 {{ 29 "decision": "approve|reject|escalate", 30 "confidence": 0.0-1.0, 31 "reasoning": "string", 32 "risk_level": "low|medium|high" 33 }}""" 34 }, 35 {"role": "user", "content": json.dumps(request)} 36 ] 37 ) 38 39 return json.loads(response.choices[0].message.content) 40 41# Rules 42rules = [ 43 { 44 "name": "auto_approve_small", 45 "condition": lambda x: x["amount"] < 1000, 46 "action": "approve", 47 "reason": "Amount below threshold" 48 }, 49 { 50 "name": "manager_approval", 51 "condition": lambda x: x["amount"] < 10000, 52 "action": "route", 53 "approvers": ["manager"] 54 } 55] 56 57workflow = ApprovalWorkflow(rules) 58## 오류 처리 & 모니터링 59 60```python 61class AutomationMonitor: 62 def __init__(self): 63 self.executions = [] 64 65 def log_execution(self, workflow_name: str, input_data: dict, 66 result: dict, duration: float): 67 self.executions.append({ 68 "workflow": workflow_name, 69 "timestamp": datetime.now(), 70 "success": "error" not in result, 71 "duration": duration, 72 "input_size": len(str(input_data)), 73 "error": result.get("error") 74 }) 75 76 def get_metrics(self) -> dict: 77 if not self.executions: 78 return {} 79 80 successful = [e for e in self.executions if e["success"]] 81 82 return { 83 "total_executions": len(self.executions), 84 "success_rate": len(successful) / len(self.executions), 85 "avg_duration": sum(e["duration"] for e in self.executions) / len(self.executions), 86 "error_types": self._group_errors() 87 }
통합 패턴
Webhook 트리거
1from fastapi import FastAPI, Request 2 3app = FastAPI() 4 5@app.post("/webhook/email") 6async def handle_email_webhook(request: Request): 7 data = await request.json() 8 9 # Email processing 10 classification = classify_email(data["body"], CATEGORIES) 11 12 if classification["requires_response"]: 13 reply = generate_auto_reply(data, COMPANY_CONTEXT) 14 send_email(data["from"], reply) 15 16 # Route to appropriate workflow 17 workflow = get_workflow(classification["category"]) 18 result = workflow.execute(data) 19 20 return {"status": "processed", "result": result}
결론
AI 워크플로 자동화는 비즈니스 프로세스를 가속화하고 오류율을 줄입니다. 문서 처리, 이메일 자동화, 의사결정 지원 시스템을 통해 상당한 효율성을 달성할 수 있습니다.
Veni AI는 엔터프라이즈 자동화 솔루션을 개발하고 있습니다.