Agent工作流在生活场景的落地复盘：从Demo到生产的关键突破

一、Demo中的Agent很聪明，生产中的Agent容易"死循环"

Agent工作流在Demo中表现惊艳——一个"今日计划Agent"能够自动查询天气、检查日历、调用AI生成简报、将结果推送到通知。演示视频中6个步骤一气呵成。但当同一套逻辑部署到生产环境24小时运行时，三类问题持续出现。

循环决策：Agent在"生成待办优先级列表"步骤中，发现自己排出的前三项在日历中已有安排，于是重新生成。因为未对"重新生成"设置上限，Agent在3项任务间反复调整循环了7次才因Token耗尽而终止。每次循环消耗约500 Token，7次循环浪费了3500 Token且未产出有效结果。

工具调用失败链：天气API短暂超时（3秒）时，Agent判断为"天气数据不可用"并跳过该步骤。但后续的"穿搭建议"工具发现缺少天气数据，向主Agent报告错误。主Agent重新尝试调用天气API——同样因未设置重试上限而陷入"调用→失败→重试"的循环。

状态丢失：运行在第4步时服务器重启（部署新版本），Agent的内存状态全部丢失。从零开始重新执行，但第1-3步的中间结果（已生成的部分简报内容）无法恢复，导致第4步的时间线分析使用了不完整的数据产生错误建议。

二、Agent可靠性的三个保障层：循环防护、状态持久化与人工检查点

从Demo到生产，Agent需要三层可靠性保障。

第一层：循环防护。每次步骤执行后检查是否与前2步的动作完全一致。如果连续3次执行相同的工具调用，强制中断当前路径并降级为预设的兜底方案。这是最低成本的防护——不依赖任何外部推理，只在执行记录上做模式匹配。

第二层：状态持久化。每一步执行完成后，将当前状态（已完成步骤、中间结果、剩余步骤列表）序列化到持久化存储（PostgreSQL或Redis）。服务器重启后能从断点恢复，而非从头开始。持久化的粒度是步骤级而非操作级——因为单个工具调用（如API请求）的恢复代价太低，不值得持久化开销。

第三层：人工检查点。对于影响用户可感知结果的关键步骤（如最终输出推送、数据修改），在执行前设置确认检查点。检查点的策略是可配置的：生产环境中默认为"必须确认"，Demo模式中可设为"自动通过"。这层保障不是对Agent能力的质疑，而是对非确定性系统（LLM推理）的工程预期管理。

三、Agent工作流引擎的核心实现：循环检测与断点恢复

"""
Agent工作流引擎：循环检测、断点恢复与检查点控制
设计意图：为LLM驱动的Agent添加确定性保障机制，
确保即使在非确定性推理环境中也能安全运行在生产场景
"""

import json
import hashlib
from datetime import datetime
from typing import Optional
from enum import Enum

class StepStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    SKIPPED = "skipped"  # 非关键步骤降级时跳过
    FAILED = "failed"

class LoopDetector:
    """循环检测器：通过动作指纹识别Agent重复行为"""
    
    def __init__(self, max_consecutive_same: int = 3):
        self.max_same = max_consecutive_same
        self.recent_actions: list[str] = []  # 保留最近N个动作的指纹
    
    def record_action(self, tool_name: str, params: dict) -> None:
        """记录当前动作并生成指纹"""
        # 动作指纹：工具名+参数哈希，忽略无关变化（如timestamp）
        stable_params = {k: v for k, v in params.items() if k != 'timestamp'}
        fingerprint = f"{tool_name}:{hashlib.md5(json.dumps(stable_params, sort_keys=True).encode()).hexdigest()}"
        self.recent_actions.append(fingerprint)
        # 只保留最近max_same*2个指纹，控制内存占用
        if len(self.recent_actions) > self.max_same * 2:
            self.recent_actions = self.recent_actions[-self.max_same * 2:]
    
    def is_looping(self) -> bool:
        """检查最近max_same个动作是否完全相同（循环检测）"""
        if len(self.recent_actions) < self.max_same:
            return False
        last_n = self.recent_actions[-self.max_same:]
        return len(set(last_n)) == 1  # 全部相同即为循环
    
    def get_break_suggestion(self) -> str:
        """循环发生时提供跳出建议"""
        return '连续重复执行相同操作，触发循环保护，建议跳过当前步骤或调用不同工具'

class WorkflowStateManager:
    """工作流状态管理器：支持步骤级断点恢复"""
    
    def __init__(self, db_client):
        self.db = db_client
    
    async def save_checkpoint(self, run_id: str, step_index: int, state: dict) -> None:
        """保存当前步骤的完整状态快照"""
        checkpoint = {
            'run_id': run_id,
            'step_index': step_index,
            'completed_steps': state.get('completed', []),
            'intermediate_results': state.get('results', {}),
            'remaining_steps': state.get('remaining', []),
            'saved_at': datetime.utcnow().isoformat(),
        }
        # 使用UPSERT确保同一步骤多次保存时不创建重复记录
        await self.db.execute("""
            INSERT INTO agent_checkpoints (run_id, step_index, state, saved_at)
            VALUES ($1, $2, $3, $4)
            ON CONFLICT (run_id, step_index) 
            DO UPDATE SET state = $3, saved_at = $4
        """, run_id, step_index, json.dumps(checkpoint), checkpoint['saved_at'])
    
    async def restore_checkpoint(self, run_id: str) -> Optional[dict]:
        """从最近的检查点恢复状态"""
        row = await self.db.fetchrow("""
            SELECT state FROM agent_checkpoints 
            WHERE run_id = $1 
            ORDER BY step_index DESC LIMIT 1
        """, run_id)
        if row:
            return json.loads(row['state'])
        return None
    
    async def needs_human_review(self, step: dict) -> bool:
        """判断当前步骤是否需要人工审核"""
        # 写操作步骤、用户可见输出步骤、高风险操作步骤需要审核
        critical_categories = {'data_write', 'user_output', 'external_action'}
        return step.get('category') in critical_categories
    
    async def cleanup_old_checkpoints(self, older_than_days: int = 7) -> int:
        """清理过期检查点，释放存储空间"""
        result = await self.db.execute("""
            DELETE FROM agent_checkpoints 
            WHERE saved_at < NOW() - INTERVAL '1 day' * $1
        """, older_than_days)
        return result

循环检测器通过动作指纹（工具名+去时间戳的参数哈希）识别重复行为。当连续N次（默认3次）的指纹相同时触发循环保护，强制中断并降级。这是一个完全确定性的防护机制，不依赖LLM推理。

状态管理器在每步骤完成后执行持久化。选择UPSERT而非INSERT应对同一检查点的多次保存。断点恢复从最大step_index的行读取最新状态，确保重启后从前一步骤而非起始步骤恢复。

四、Agent可靠性保障的代价：延迟与复杂度增长

循环检测和状态持久化都是有成本的。每步持久化增加50-100ms的数据库写入延迟。对工作流仅有3步的简单Agent来说，这意味着约15%的额外延迟。人工检查点更是将执行时间的不确定性从秒级拉到分钟级（取决于人工响应速度）。

更结构性的代价是循环防护的误判问题。对于确实需要反复执行相同操作的工作流（如"持续监控天气直到放晴"），循环检测可能将合法行为误判为死循环。解决方案是引入步骤意图标记——标记为loop_allowed的步骤不对其进行循环检测，或使用更高的阈值。

适用判断：工作流≥5步、运行在生产环境、涉及用户数据写入的Agent，必须配备循环防护和状态持久化。Demo和工作流≤3步的快速任务可不引入这些机制。

五、总结

Agent从Demo到生产需要在确定性不足的LLM推理上叠加确定性保障层：

1. 循环防护：通过动作指纹检测重复行为，连续≥3次相同动作时强制中断降级。
2. 状态持久化：步骤级保存中间状态，支持断点恢复，避免从头执行。
3. 人工检查点：关键步骤（数据写入、外部输出）执行前须人工确认，控制非确定性风险。
4. 成本权衡：每步持久化增加50-100ms，3步以上的工作流收益大于成本。
5. 循环误判处理：需要重复执行的工作流添加loop_allowed标记，豁免循环检测。
6. 部署分级：生产环境全开启（循环检测+持久化+检查点），Demo环境可简化配置。