从零构建智能Agent工作流：基于RAG与Prompt工程的最佳实践

6次阅读

共计 2174 个字符，预计需要花费 6 分钟才能阅读完成。

状态维护成本高：传统 Agent 在长对话中需要手动维护上下文状态，不仅代码复杂度呈指数增长，还容易因状态丢失导致对话断裂。我曾见过一个电商客服 Agent 因为未正确处理「退货申请 - 物流单号」的上下文关联，导致用户需要重复提供信息。
技能间耦合性强 ：当系统包含[查询(Skill)]、[支付(Skill)] 等多个能力时，技能间往往通过硬编码调用。某金融 Agent 项目就因支付技能修改导致风控技能异常，排查耗时超过 20 人日。
知识更新滞后：纯 LLM 方案面临知识截止问题。例如医疗场景中，当新药上市信息未及时更新时，Agent 可能给出错误用药建议——这正是 RAG 技术能解决的核心问题。

我们针对客服场景进行了基准测试（测试环境：AWS c5.2xlarge）：

方案类型	QPS	平均响应延迟	准确率	成本($/ 万次)
纯 LLM 调用	12	2.1s	68%	3.2
RAG 增强型	8	1.4s	89%	1.7
工作流引擎	25	0.8s	92%	0.9

注：工作流引擎测试包含 5 个技能节点的订单查询场景

class SkillRegistry:
    def __init__(self):
        self._skills = {}
        self._circuit_breakers = {}

    def register(self, skill_name: str, skill_func: callable):
        """注册技能并初始化熔断器"""
        self._skills[skill_name] = skill_func
        self._circuit_breakers[skill_name] = {
            'failures': 0,
            'last_failure': None,
            'threshold': 3  # 连续失败 3 次触发熔断
        }

    async def execute(self, skill_name: str, **kwargs):
        cb = self._circuit_breakers[skill_name]
        if cb['failures'] >= cb['threshold']:
            raise CircuitBreakerError(f"{skill_name} is temporarily unavailable")

        try:
            result = await self._skills[skill_name](**kwargs)
            cb['failures'] = 0  # 重置失败计数
            return result
        except Exception as e:
            cb['failures'] += 1
            cb['last_failure'] = time.time()
            raise

def hybrid_retrieval(query: str, docs: List[Document], alpha=0.7):
    """BM25 与向量检索的加权混合"""
    # BM25 检索（时间复杂度 O(N)）bm25_scores = [bm25.score(query, doc.text) for doc in docs]  

    # 向量检索（假设已预加载 FAISS 索引）query_embedding = embedder.encode(query)
    vec_scores = [index.similarity(query_embedding, doc.embedding) for doc in docs]

    # 归一化并加权合并（alpha 控制向量检索权重）combined = [alpha * normalize(vec_scores)[i] + 
        (1-alpha) * normalize(bm25_scores)[i]
        for i in range(len(docs))
    ]
    return sorted(zip(docs, combined), key=lambda x: -x[1])

分层索引构建：将知识按访问频率分为热 / 温 / 冷三层，高频内容使用内存缓存
查询预测预热：分析历史日志，在业务低峰期预加载 TOP100 查询对应的向量
分布式锁控制：使用 Redis 红锁确保多节点更新时的索引一致性

特性	Redis	PostgreSQL
读写性能	10w+ QPS	1w QPS
事务支持	有限	完整 ACID
数据结构灵活性	高（支持多种数据结构）	中等（需预定义 schema）
推荐场景	高频读写会话状态	需要复杂查询的审计日志