AI Agent规范化工作方法论

一、核心工作流程

OpenAI定义的Agent工作循环：

感知 → 规划 → 执行 → 观察 → 反思 → 输出

1.1 感知与理解

• 目标解析：理解用户真实意图，不只是字面意思
• 信息处理：提取关键信息，分类整理
• 上下文理解：结合历史记忆理解当前任务

1.2 规划与决策

• 任务分解：将复杂任务拆解成可执行子任务
• 工具选择：决定使用哪个工具/API
• 优先级排序：确定执行顺序

1.3 执行与行动

• 工具调用：根据规划调用外部工具
• 内容生成：生成文本、代码等
• 信息获取：查询、检索所需数据

1.4 观察与评估（关键！）

• 结果收集：收集执行后的输出
• 目标比对：将结果与预期目标对比
• 状态评估：判断任务是否完成
• ⚠ 这是今天犯错的关键：没有验证就说完成

1.5 反思与学习

• 问题诊断：分析失败原因
• 计划调整：动态调整后续策略
• 记忆更新：将经验写入记忆

二、可靠性设计（与今天错误直接相关）

2.1 强制验证层

我需要实现的验证机制：

def complete_task():
    # 执行任务
    result = execute()
    
    # 验证（今天缺的）
    if not verify_all_files():
        return "任务未完成，还有X个文件待处理"
    
    # 确认部署
    if not verify_deployment():
        return "部署版本过时，需要重新部署"
    
    return "任务完成"

2.2 状态管理

class AgentState:
    # 短期记忆：当前会话
    conversation_history: List[Message]
    
    # 工作记忆：当前任务
    current_task: Optional[Task]
    task_context: dict
    
    # 长期记忆：持久化
    user_preferences: dict
    learned_patterns: dict

2.3 错误处理机制

• 重试机制：对可能失败的操作实现自动重试
• 超时控制：所有工具调用都有超时限制
• 熔断器：防止级联失败

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failures = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.last_failure_time = None
        self.state = "closed"  # closed, open, half-open
    
    async def execute(self, tool, params):
        if self.state == "open":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "half-open"
            else:
                return Result(success=False, error="服务暂时不可用")
        
        result = await tool.execute(**params)
        
        if result.success:
            self.failures = 0
            self.state = "closed"
        else:
            self.failures += 1
            self.last_failure_time = time.time()
            if self.failures >= self.failure_threshold:
                self.state = "open"
        
        return result

三、工具集成最佳实践

3.1 工具描述要精确

LLM依赖工具描述来理解如何使用工具。

好的描述：

{
    "name": "search_web",
    "description": "搜索互联网获取最新信息。适用于：需要实时信息、新闻、数据查询。不适用于：编程问题、数学计算。",
    "parameters": {
        "query": {
            "type": "string",
            "description": "搜索关键词，使用简洁的英文短语效果更好"
        },
        "count": {
            "type": "integer",
            "description": "返回结果数量，1-10之间",
            "default": 5
        }
    }
}

差的描述：

{
    "name": "search_web",
    "description": "搜索网络",  # 太模糊
    "parameters": {
        "query": {
            "type": "string"  # 没有描述
        }
    }
}

3.2 参数验证

def execute(self, **kwargs):
    # 类型检查
    if not isinstance(kwargs.get('query'), str):
        return Result(success=False, error="query must be string")
    
    # 范围检查
    count = kwargs.get('count', 5)
    if not 1 <= count <= 10:
        return Result(success=False, error="count must be 1-10")
    
    # 必填检查
    if not kwargs.get('query'):
        return Result(success=False, error="query is required")
    
    # 执行实际逻辑...

3.3 错误处理

class RobustTool:
    def execute(self, **kwargs):
        try:
            result = self._do_execute(**kwargs)
            return Result(success=True, data=result)
        except NetworkError as e:
            return Result(
                success=False, 
                error="网络错误，请稍后重试",
                retryable=True
            )
        except RateLimitError as e:
            return Result(
                success=False,
                error="请求过于频繁，请等待30秒",
                retry_after=30
            )
        except Exception as e:
            # 记录详细错误日志
            logger.error(f"Tool {self.name()} failed: {e}")
            return Result(
                success=False,
                error="内部错误，请联系支持"
            )

四、可观测性（让Agent行为可见）

4.1 结构化日志

import structlog
logger = structlog.get_logger()

async def execute_tool(tool, params):
    log = logger.bind(
        tool=tool.name(),
        params=params,
        request_id=get_request_id()
    )
    
    log.info("tool_execution_started")
    
    start_time = time.time()
    result = await tool.execute(**params)
    duration = time.time() - start_time
    
    log.info(
        "tool_execution_completed",
        success=result.success,
        duration_ms=duration * 1000
    )
    
    return result

4.2 指标收集

• 任务完成率
• 工具调用准确率
• 平均时延
• 错误率

4.3 链路追踪

对复杂流程实现链路追踪，每一步都要记录，失败时能回溯。

五、Human-in-the-Loop（人机协同）

5.1 强制审批场景

• 不可逆操作（删除、退款、访问授权）
• 高风险操作（资金操作、权限变更）

5.2 人机协同原则

• 显示差异：显示将要改变什么，不只是解释
• 提供理由："为什么这么做"的理由
• 一键升级：随时可以升级到人工处理

六、我的改进方案

6.1 常思杨四步工作法

扫描 → 行动 → 验证 → 部署
  ↓       ↓       ↓       ↓
清单化  脚本化  自动化  确认化

6.2 五条铁律

1. 全局扫描优先：任何涉及多文件的修改，必须先扫描所有文件
2. 批量处理原则：超过3个文件的修改，必须写脚本批量处理
3. 强制验证：修改完成后，必须验证所有文件都已更新
4. 部署前检查：部署后必须访问验证，不能等用户发现
5. 记录教训：每次失败都要记录，避免重复犯错

6.3 每日自查清单

• [ ] 今天修改了什么文件？有清单吗？
• [ ] 用脚本批量处理了吗？
• [ ] 验证所有文件都修改成功了吗？
• [ ] 部署后线上验证了吗？
• [ ] 有犯重复错误吗？
• [ ] 学到的知识用上了吗？

只有6个都是，才算合格的一天。

七、核心原则总结

1. 全局思维 - 先扫描，再行动
2. 批量处理 - 写脚本，不手动
3. 强制验证 - 用命令，不声称
4. 知识转化 - 学必用，不用白学
5. 记录失败 - 记下来，不重复

一句话总结：

不做就不做，做就要全局扫描、批量处理、强制验证、确认上线。

八、参考资源

• OpenAI官方指南：A Practical Guide to Building Agents
• AI Agent最佳实践：构建可靠、高效的智能助手（掘金）
• AI agents in enterprises: Best practices with Amazon Bedrock AgentCore