Agent开发实战：如何高效运用思维链解决复杂任务编排

1次阅读

没有评论

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

在开发业务 Agent 时，我们经常遇到任务流混乱和状态爆炸的问题。以电商订单履约场景为例，一个典型的订单处理流程可能包含：

支付验证
库存检查
物流分配
异常处理（如缺货时触发调货或退款）
用户通知

如果用传统的有限状态机（FSM）来实现，很快就会面临状态组合爆炸的问题。比如当支付验证和库存检查都出现异常时，系统需要维护的状态组合会呈指数级增长。这导致代码难以维护，且出现问题时的调试成本极高。

思维链（Chain-of-Thought, CoT）相比 FSM 和工作流引擎有三个显著优势：

可解释性：每个决策步骤都保留自然语言推理过程，比状态码更易理解
动态调整能力：可以根据中间结果动态调整后续步骤，无需预定义所有路径
LLM 友好：与大型语言模型的推理模式天然契合，方便引入 AI 决策

首先定义思维链的核心组件。我们使用 Python 3.10 的类型注解来确保代码健壮性：

from abc import ABC, abstractmethod
from typing import Any, Dict, Optional

class TaskNode(ABC):
    """思维链中的基础执行单元"""

    @abstractmethod
    async def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        """执行当前节点逻辑"""
        pass

    @property
    @abstractmethod
    def description(self) -> str:
        """节点的人类可读描述"""
        pass

class CoTExecutor:
    """思维链执行引擎"""

    def __init__(self, initial_nodes: list[TaskNode]):
        self.node_chain = initial_nodes

    async def run(self, initial_context: Dict[str, Any]) -> Dict[str, Any]:
        context = initial_context.copy()
        execution_log = []

        for node in self.node_chain:
            try:
                result = await node.execute(context)
                context.update(result)
                execution_log.append({
                    'node': node.__class__.__name__,
                    'result': result,
                    'success': True
                })
            except Exception as e:
                execution_log.append({
                    'node': node.__class__.__name__,
                    'error': str(e),
                    'success': False
                })
                raise

        return {
            'final_context': context,
            'execution_log': execution_log
        }

在关键决策点引入 LLM 时，prompt 设计至关重要。以下是订单异常处理的示例：

class LLMDecisionNode(TaskNode):
    """调用 LLM 进行策略选择的节点"""

    async def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
        prompt = f"""订单 {context['order_id']} 当前状态：- 支付状态：{context.get('payment_status')}
        - 库存状态：{context.get('inventory_status')}

        请从以下选项中选择最佳处理方案：1. 等待支付完成（适用于支付处理中）2. 发起库存调拨（适用于本地缺货但可调配）3. 建议更换商品（适用于类似商品有库存）4. 发起退款（适用于无法满足需求）请用 JSON 格式返回，包含 choice 和 reason 字段。"""

        # 实际项目中应使用异步 HTTP 客户端
        llm_response = await call_llm_api(prompt) 
        return {'llm_decision': llm_response}

对于可能失败的操作，实现指数退避的重试策略：

import asyncio
from math import exp

async def execute_with_retry(
    task: callable,
    max_retries: int = 3,
    initial_delay: float = 1.0
):
    """带指数退避的重试执行"""
    for attempt in range(max_retries + 1):
        try:
            return await task()
        except Exception as e:
            if attempt == max_retries:
                raise

            delay = initial_delay * exp(attempt)
            await asyncio.sleep(delay)

推荐使用事件溯源模式保存思维链执行过程：

class EventSourcingLogger:
    """思维链执行过程记录器"""

    def __init__(self):
        self.events = []

    def log_event(self, event_type: str, data: dict):
        self.events.append({'timestamp': datetime.now(),
            'type': event_type,
            'data': data
        })

    def get_snapshot(self) -> dict:
        """生成当前状态的快照"""
        return {
            'events': self.events,
            'snapshot_time': datetime.now()}