Agent Workflow Builder

Agent 工作流构建器

Agent 工作流构建器技能将引导你设计和实现多 Agent AI 系统，使其能够规划、推理、使用工具并协同合作，以完成复杂任务。现代 AI 应用越来越多地依赖 Agent 架构，其中大模型充当推理引擎，负责编排操作，而不只是生成文本。

此技能涵盖 Agent 设计模式、工具集成、状态管理、错误处理和人在回路工作流。它能帮助你构建稳健的 Agent 系统，在保持安全性和可控性的同时，应对现实世界的复杂情况。

无论你是在构建自主助手、工作流自动化，还是复杂推理系统，此技能都能确保你的 Agent 架构设计完善，可用于生产环境。

核心工作流

工作流 1：设计 Agent 架构

1. 定义 Agent 的范围：
   • 它应自主处理哪些任务？
   • 哪些任务需要人工批准？
   • 明确不在其范围内的事项有哪些？
2. 选择架构模式：

   模式	描述	适用场景
   单 Agent	一个带工具的大模型	简单任务、范围明确
   路由 Agent	分类并委派任务	多个不同领域
   顺序链	按顺序运行的 Agent	流水线处理
   层级式	管理者 + 工作 Agent	复杂且可拆分的任务
   协作式	相互讨论的同级 Agent	需要多种专业知识

3. 设计工具集：
   • Agent 需要哪些能力？
   • 如何定义和记录工具？
   • 安全边界有哪些？
4. 规划状态管理：
   • 对话历史
   • 任务状态和进度
   • 外部系统状态
5. 记录架构决策

工作流 2：实现 Agent 循环

1. 构建核心 Agent 循环：

   class Agent:
       def __init__(self, llm, tools, system_prompt):
           self.llm = llm
           self.tools = {t.name: t for t in tools}
           self.system_prompt = system_prompt
   
       async def run(self, user_input, max_steps=10):
           messages = [
               {"role": "system", "content": self.system_prompt},
               {"role": "user", "content": user_input}
           ]
   
           for step in range(max_steps):
               response = await self.llm.chat(messages, tools=self.tools)
   
               if response.tool_calls:
                   # Execute tools
                   for call in response.tool_calls:
                       result = await self.execute_tool(call)
                       messages.append({"role": "tool", "content": result})
               else:
                   # Final response
                   return response.content
   
           raise MaxStepsExceeded()
   
       async def execute_tool(self, call):
           tool = self.tools[call.name]
           return await tool.execute(call.arguments)
   

2. 实现接口清晰的工具
3. 添加错误处理和重试机制
4. 加入日志记录和可观测性
5. 使用多样化场景进行测试

工作流 3：构建多 Agent 系统

1. 定义 Agent 角色：

   agents = {
       "planner": Agent(
           llm=gpt4,
           tools=[search, create_task],
           system_prompt="You decompose complex tasks into steps..."
       ),
       "researcher": Agent(
           llm=claude,
           tools=[web_search, read_document],
           system_prompt="You gather and synthesize information..."
       ),
       "executor": Agent(
           llm=gpt4,
           tools=[code_interpreter, file_system],
           system_prompt="You execute tasks and produce outputs..."
       ),
       "reviewer": Agent(
           llm=claude,
           tools=[validate, provide_feedback],
           system_prompt="You review work for quality and correctness..."
       )
   }
   

2. 实现编排：
   • Agent 如何通信？
   • 谁决定何时运行什么？
   • 如何在 Agent 之间传递工作？
3. 管理共享状态：
   • 任务看板或工作队列
   • 共享内存或上下文
   • 工件存储
4. 妥善处理故障
5. 在需要时添加人工检查点

快速参考

操作	命令/触发语
设计 Agent	“为 [task] 设计一个 Agent”
添加工具	“[agent type] 需要哪些工具”
构建多 Agent 系统	“为 [goal] 构建多 Agent 系统”
处理错误	“Agent 错误处理模式”
添加人在回路环节	“在 Agent 工作流中添加人工审批”
调试 Agent	“调试 Agent 工作流”

最佳实践

• 从简单开始：先使用带工具的单个 Agent，再考虑多 Agent

  • 以最低复杂度证明价值
  • 仅在必要时添加 Agent
  • 每个 Agent 都应有明确且独立的职责

• 谨慎设计工具：工具是 Agent 的双手

  • 使用清晰、描述性强的名称和文档
  • 明确定义输入／输出架构
  • 妥善处理错误并提供错误消息
  • 尽可能确保操作具有幂等性

• 限制 Agent 的自主性：控制影响范围

  • 明确规定 Agent 不能做什么
  • 对影响重大的操作要求审批
  • 设置支出／速率限制
  • 记录所有操作以供审计

• 明确管理状态：不要只依赖大模型的记忆

  • 持久化对话和任务状态
  • 总结较长的上下文以适配窗口
  • 跟踪已尝试／已完成的内容

• 优雅地处理失败：Agent 会遇到错误

  • 提供清晰的错误消息，供 Agent 推理
  • 使用带退避机制的重试逻辑
  • 制定回退策略
  • 设置人工升级路径

• 全面观察：调试 Agent 很困难

  • 记录所有大模型调用和工具调用
  • 跟踪推理链和决策
  • 按任务类型衡量成功率

高级技巧

ReAct 模式（推理 + 行动）

明确构建 Agent 的思考过程：

REACT_PROMPT = """
You are an agent that solves tasks step by step.

For each step:
1. Thought: Analyze the current situation and decide what to do
2. Action: Choose a tool and provide arguments
3. Observation: Review the tool result

Continue until you can provide a final answer.

Available tools: {tool_descriptions}

Current task: {task}

Begin:
"""

带任务分解的规划 Agent

将复杂任务拆分为易于管理的步骤：

class PlanningAgent:
    async def solve(self, task):
        # Step 1: Create plan
        plan = await self.create_plan(task)

        # Step 2: Execute each step
        results = []
        for step in plan.steps:
            result = await self.execute_step(step, context=results)
            results.append(result)

            # Replan if needed
            if result.status == "blocked":
                plan = await self.replan(task, results)

        # Step 3: Synthesize final output
        return await self.synthesize(task, results)

反思与自我纠正

让 Agent 审查并改进自己的工作：

async def solve_with_reflection(self, task, max_attempts=3):
    for attempt in range(max_attempts):
        # Generate solution
        solution = await self.generate_solution(task)

        # Self-critique
        critique = await self.critique_solution(task, solution)

        if critique.is_acceptable:
            return solution

        # Improve based on critique
        task = f"{task}\n\nPrevious attempt issues: {critique.issues}"

    return solution  # Return best effort

人机协同检查点

将人工审批整合到工作流中：

class HumanApprovalTool:
    async def execute(self, action_description, risk_level):
        if risk_level == "low":
            return {"approved": True, "auto": True}

        # Send to approval queue
        approval_request = await self.create_request(action_description)

        # Wait for human response (with timeout)
        response = await self.wait_for_approval(
            approval_request.id,
            timeout_minutes=30
        )

        return {
            "approved": response.approved,
            "feedback": response.feedback,
            "auto": False
        }

记忆管理

处理较长的对话和上下文：

class AgentMemory:
    def __init__(self, max_tokens=8000):
        self.max_tokens = max_tokens
        self.messages = []
        self.summaries = []

    def add(self, message):
        self.messages.append(message)

        if self.token_count() > self.max_tokens:
            self.compress()

    def compress(self):
        # Summarize older messages
        old_messages = self.messages[:-5]  # Keep recent
        summary = summarize(old_messages)

        self.summaries.append(summary)
        self.messages = self.messages[-5:]

    def get_context(self):
        return {
            "summaries": self.summaries,
            "recent_messages": self.messages
        }

应避免的常见问题

• 在单个 Agent 足以胜任时仍构建多 Agent 系统
• 在缺乏安全边界的情况下赋予 Agent 过多自主权
• 未处理工具故障和边界情况
• 忘记大模型可能会产生幻觉，虚构工具调用
• Agent 陷入困境时进入无限循环
• 没有记录足够的信息来调试 Agent 行为
• 假定 Agent 会完美遵循指令
• 忽视 Agent 循环中的成本（Token 用量）