Veni AI
Automação

Automação de Fluxos de Trabalho com IA: Guia de Automação de Processos Empresariais

Automação de processos empresariais com IA, processamento de documentos, automação de e-mails, extração de dados e integração de RPA corporativa.

Veni AI Technical Team29 Aralık 20246 dk okuma
Automação de Fluxos de Trabalho com IA: Guia de Automação de Processos Empresariais

Automação de Fluxos de Trabalho com IA: Guia de Automação de Processos de Negócios

A automação de fluxos de trabalho com IA é a automação de processos de negócios repetitivos usando inteligência artificial. Neste guia, examinamos estratégias de automação corporativa.

Categorias de Automação

1. Processamento de Documentos

  • Processamento de faturas
  • Análise de contratos
  • Extração de dados de formulários

2. Automação de Comunicação

  • Classificação de e-mails
  • Respostas automáticas
  • Roteamento de tickets

3. Processamento de Dados

  • Extração de dados
  • Transformação de dados
  • Relatórios

4. Automação de Decisões

  • Processos de aprovação
  • Avaliação de risco
  • Detecção de anomalias

Pipeline de Processamento de Documentos

Processamento de Faturas

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álise 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## Automação de Email
25
26### Classificação 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)

Geração de Resposta Automática

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

Orquestração de Workflow

Pipeline Sequencial

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"})

Workflow 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## Extração de Dados
34
35### Extração de Tabelas
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

Extração 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)

Fluxos de Aprovação

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## Tratamento de Erros & Monitoramento
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        }

Padrões de Integração

Webhook Trigger

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}

Conclusão

A automação de workflows com IA acelera processos de negócios e reduz taxas de erro. Ganhos significativos de eficiência podem ser alcançados com processamento de documentos, automação de emails e sistemas de suporte à decisão.

Na Veni AI, desenvolvemos soluções de automação corporativa.

İlgili Makaleler

AI Agents

Padrões de Agentes de IA para Empresas: Modelos Operacionais Emergentes no Início de 2026

Daha fazla oku →
AI Governance

Governança de IA Corporativa: Registro de Modelos e Padrões de Avaliação

Daha fazla oku →
RAG

Desenvolvimentos em RAG Multimodal: Combinando Busca Vetorial e Busca por Grafos

Daha fazla oku →