Automatización de Flujos de Trabajo con IA: Guía de Automatización de Procesos Empresariales
La automatización de flujos de trabajo con IA es la automatización de procesos empresariales repetitivos utilizando inteligencia artificial. En esta guía, examinamos estrategias de automatización empresarial.
Categorías de Automatización
1. Procesamiento de Documentos
- Procesamiento de facturas
- Análisis de contratos
- Extracción de datos de formularios
2. Automatización de la Comunicación
- Clasificación de correos electrónicos
- Respuestas automáticas
- Enrutamiento de tickets
3. Procesamiento de Datos
- Extracción de datos
- Transformación de datos
- Generación de informes
4. Automatización de Decisiones
- Procesos de aprobación
- Evaluación de riesgos
- Detección de anomalías
Flujo de Procesamiento de Documentos
Procesamiento de Facturas
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)
Análisis de Contratos
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## Automatización de Email 25 26### Clasificación de Email 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)
Generación de Respuestas Automáticas
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
Orquestación de Flujos de Trabajo
Pipeline Secuencial
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"})
Flujo de Trabajo Condicional
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## Extracción de Datos 34 35### Extracción de Tablas 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
Extracción de Entidades
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)
Flujos de Aprobación
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## Manejo de Errores y Monitoreo 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 }
Patrones de Integración
Disparador 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}
Conclusión
La automatización de flujos de trabajo con IA acelera los procesos empresariales y reduce las tasas de error. Se pueden lograr mejoras significativas en eficiencia mediante el procesamiento de documentos, la automatización de correos electrónicos y los sistemas de apoyo a decisiones.
En Veni AI, desarrollamos soluciones de automatización empresarial.