RPA软件+大模型混合架构：什么时候该用RAG，什么时候直接Prompt

┌─────────────────────────────────────────┐
│           任务输入（自然语言描述）          │
└─────────────────┬───────────────────────┘
                  ▼
        ┌─────────────────┐
        │  是否需要外部知识？ │
        └────────┬────────┘
                 │
       ┌─────────┴─────────┐
       ▼                   ▼
      是                  否
       │                   │
       ▼                   ▼
┌─────────────┐    ┌─────────────────┐
│ 知识是否结构化？ │    │ 输入是否实时数据？ │
└──────┬──────┘    └────────┬────────┘
       │                    │
   ┌───┴───┐            ┌───┴───┐
   ▼       ▼            ▼       ▼
  结构化  非结构化       是      否
   │       │            │       │
   ▼       ▼            ▼       ▼
 API查询  RAG检索    直接Prompt  通用推理

class HybridAutomationRouter:
    """
    RPA+大模型混合架构路由决策器
    根据任务特征自动选择RAG或直接Prompt
    """
    
    def __init__(self, knowledge_base, llm_client, db_connector):
        self.kb = knowledge_base          # 向量知识库
        self.llm = llm_client             # 大模型客户端
        self.db = db_connector            # 业务数据库连接
        
    def route_and_execute(self, task_desc, context=None):
        """
        主路由方法
        task_desc: 用户自然语言描述的任务
        context: 附加上下文（如页面截图、当前状态）
        """
        # 第一步：判断是否需要外部知识
        if self._needs_knowledge(task_desc):
            # 需要知识 → 判断知识类型
            if self._is_structured_knowledge(task_desc):
                # 结构化知识 → 直接查数据库/API
                data = self.db.query(self._extract_query_params(task_desc))
                prompt = self._build_data_prompt(task_desc, data)
            else:
                # 非结构化知识 → RAG检索
                docs = self.kb.search(task_desc, top_k=5)
                prompt = self._build_rag_prompt(task_desc, docs)
        else:
            # 不需要外部知识 → 直接Prompt
            prompt = self._build_direct_prompt(task_desc, context)
        
        # 统一调用大模型
        return self.llm.generate(prompt)
    
    def _needs_knowledge(self, task):
        """判断任务是否需要外部知识支撑"""
        knowledge_keywords = [
            '政策', '规定', '手册', '流程', '标准',
            '根据', '按照', '依据', '参考'
        ]
        return any(kw in task for kw in knowledge_keywords)
    
    def _is_structured_knowledge(self, task):
        """判断所需知识是否为结构化数据"""
        structured_keywords = [
            '订单', '库存', '价格', '状态', '余额',
            '数量', '金额', '时间', '等级'
        ]
        return any(kw in task for kw in structured_keywords)
    
    def _build_rag_prompt(self, task, documents):
        """构建RAG增强的Prompt"""
        context = "\n\n".join([
            f"[文档{i+1}] {doc.content}" 
            for i, doc in enumerate(documents)
        ])
        return f"""基于以下参考资料回答问题：

{context}

---
要求：
1. 优先基于上述资料回答
2. 如果资料不足以回答，明确说明"根据现有资料无法确定"
3. 回答中标注引用来源编号

用户问题：{task}"""

    def _build_data_prompt(self, task, data):
        """构建基于实时数据的Prompt"""
        return f"""基于以下实时数据执行任务：

{json.dumps(data, ensure_ascii=False, indent=2)}

---
要求：
1. 基于上述数据做逻辑判断
2. 给出明确的结论和行动建议
3. 如果数据缺失关键字段，指出缺失项

任务：{task}"""

    def _build_direct_prompt(self, task, context):
        """构建直接Prompt"""
        base = f"任务：{task}\n\n"
        if context:
            base += f"当前上下文：{context}\n\n"
        base += "请直接执行任务，给出明确的操作指令或回答。"
        return base

-- 向量检索 + 业务数据关联示例
SELECT 
    d.doc_content,
    d.embedding <-> query_vec AS distance,
    o.order_status,
    o.create_time
FROM document_vectors d
LEFT JOIN orders o ON d.ref_id = o.order_id
WHERE d.embedding <-> query_vec < 0.3
ORDER BY distance
LIMIT 5;

# 混合架构配置
AUTOMATION_CONFIG = {
    # RAG流水线配置
    "rag_pipeline": {
        "embedding_model": "text-embedding-3-small",
        "vector_db": "pgvector",           # 本地PostgreSQL
        "chunk_size": 500,                  # 文档切片大小
        "chunk_overlap": 100,               # 切片重叠
        "top_k": 5,                         # 检索返回数量
        "rerank": True,                     # 是否启用重排序
        "score_threshold": 0.7              # 相关性阈值
    },
    
    # 直接Prompt流水线配置
    "prompt_pipeline": {
        "model": "deepseek-chat",
        "temperature": 0.3,                 # 低温度保证确定性
        "max_tokens": 2048,
        "timeout": 10,                      # 超时时间（秒）
        "retry": 3                          # 失败重试次数
    },
    
    # 路由规则
    "routing_rules": [
        {
            "name": "政策解读",
            "condition": "task.contains(['政策', '规定', '手册'])",
            "pipeline": "rag",
            "knowledge_base": "regulatory_docs"
        },
        {
            "name": "订单处理",
            "condition": "task.contains(['订单', '库存', '价格'])",
            "pipeline": "prompt",
            "data_source": "order_db"
        },
        {
            "name": "通用问答",
            "condition": "default",
            "pipeline": "prompt"
        }
    ],
    
    # 降级策略
    "fallback": {
        "rag_empty": "switch_to_prompt",    # RAG无结果转直接Prompt
        "llm_timeout": "return_cached",     # 超时返回缓存
        "api_error": "notify_human"         # 连续错误通知人工
    }
}