09b - 核心 Agent 循环模式

def react_loop(goal: str, tools: list[Tool], max_steps: int = 10) -> str:
    """ReAct 循环的核心实现"""
    messages = [{"role": "user", "content": goal}]
    
    for step in range(max_steps):
        # 1. LLM 推理：生成 Thought + Action（或 Final Answer）
        response = llm.generate(
            system_prompt=REACT_SYSTEM_PROMPT,
            messages=messages,
            tools=tools
        )
        
        # 2. 检查是否已得出最终答案
        if response.has_final_answer:
            return response.final_answer
        
        # 3. 执行工具调用
        tool_name = response.action.tool
        tool_input = response.action.input
        observation = execute_tool(tool_name, tool_input)
        
        # 4. 将观察结果加入上下文
        messages.append({"role": "assistant", "content": response.thought})
        messages.append({"role": "tool", "content": observation})
    
    return "达到最大步数限制，任务未完成"

// 使用 Vercel AI SDK 实现 ReAct 模式
import { generateText, tool } from 'ai';
import { anthropic } from '@ai-sdk/anthropic';
import { z } from 'zod';
 
const searchTool = tool({
  description: '搜索网络获取实时信息',
  parameters: z.object({
    query: z.string().describe('搜索关键词'),
  }),
  execute: async ({ query }) => {
    // 实际搜索实现
    return await webSearch(query);
  },
});
 
const calculatorTool = tool({
  description: '执行数学计算',
  parameters: z.object({
    expression: z.string().describe('数学表达式'),
  }),
  execute: async ({ expression }) => {
    return String(eval(expression)); // 生产环境应使用安全的表达式解析器
  },
});
 
async function reactAgent(goal: string): Promise<string> {
  const { text } = await generateText({
    model: anthropic('claude-sonnet-4-20250514'),
    system: `你是一个能使用工具的 AI 助手。
每一步先思考（Thought），再决定行动（Action），
观察结果后继续推理，直到能给出最终答案。`,
    prompt: goal,
    tools: { search: searchTool, calculator: calculatorTool },
    maxSteps: 10, // ReAct 循环的最大步数
  });
  
  return text;
}

你是一个能使用工具的 AI 助手。请按照以下模式解决问题：

## 可用工具
[工具列表及描述]

## 执行模式
每一步按以下格式输出：

Thought: [分析当前状态，思考下一步该做什么]
Action: [选择要使用的工具]
Action Input: [工具的输入参数，JSON 格式]

观察到工具返回结果后，继续推理：

Thought: [基于观察结果的分析]
...（重复直到能给出最终答案）

当你确定可以回答时：
Thought: [最终推理]
Final Answer: [完整的最终回复]

## 约束
- 每步只调用一个工具
- 如果工具调用失败，分析原因后尝试替代方案
- 最多执行 [N] 步，超过则总结当前进展并说明未完成的部分

def plan_act_verify_loop(requirement: str, tools: list[Tool]) -> Result:
    """Plan-Act-Verify 循环的核心实现"""
    
    # ===== Phase 1: Plan =====
    plan = llm.generate(
        prompt=f"分析以下需求，分解为有序的子任务列表：\n{requirement}",
        output_format="json_list"
    )
    task_queue = plan.tasks  # [{id, description, dependencies, verification}]
    
    # ===== Phase 2: Act + Verify 循环 =====
    results = {}
    for task in task_queue:
        max_retries = 3
        for attempt in range(max_retries):
            # Act: 执行子任务
            execution = llm.generate(
                prompt=f"执行以下子任务：\n{task.description}\n已完成的上下文：{results}",
                tools=tools
            )
            
            # Verify: 验证执行结果
            verification = verify_result(
                task=task,
                result=execution,
                method=task.verification  # "run_tests" | "lint" | "type_check" | "manual"
            )
            
            if verification.passed:
                results[task.id] = execution
                break
            else:
                # 诊断失败原因，调整策略后重试
                diagnosis = llm.generate(
                    prompt=f"任务失败：{verification.error}\n分析原因并调整方案"
                )
        else:
            # 超过最大重试次数，修订计划
            plan = revise_plan(plan, task, results)
    
    # ===== Phase 3: 最终验证 =====
    final_check = run_full_verification(results)
    if not final_check.passed:
        # 补充遗漏任务
        additional_tasks = llm.generate(
            prompt=f"最终验证失败：{final_check.errors}\n生成补充任务"
        )
        # 递归处理补充任务...
    
    return results

// Plan-Act-Verify 模式的编码 Agent 实现
import { generateText, generateObject, tool } from 'ai';
import { anthropic } from '@ai-sdk/anthropic';
import { z } from 'zod';
 
// 定义计划结构
const PlanSchema = z.object({
  tasks: z.array(z.object({
    id: z.string(),
    description: z.string(),
    files: z.array(z.string()),
    verification: z.enum(['test', 'lint', 'typecheck', 'build']),
  })),
});
 
async function planActVerifyAgent(requirement: string) {
  // Phase 1: Plan — 生成结构化计划
  const { object: plan } = await generateObject({
    model: anthropic('claude-sonnet-4-20250514'),
    schema: PlanSchema,
    prompt: `分析需求并生成执行计划：\n${requirement}`,
  });
 
  console.log(`📋 计划生成完毕，共 ${plan.tasks.length} 个子任务`);
 
  // Phase 2: Act + Verify 循环
  for (const task of plan.tasks) {
    console.log(`⚡ 执行: ${task.description}`);
    
    let verified = false;
    for (let attempt = 0; attempt < 3 && !verified; attempt++) {
      // Act: 执行子任务
      const { text } = await generateText({
        model: anthropic('claude-sonnet-4-20250514'),
        prompt: `执行以下编码任务：\n${task.description}`,
        tools: { editFile, readFile, runCommand },
        maxSteps: 5,
      });
 
      // Verify: 运行验证
      const result = await runVerification(task.verification, task.files);
      
      if (result.success) {
        console.log(`✅ 验证通过: ${task.id}`);
        verified = true;
      } else {
        console.log(`❌ 验证失败 (尝试 ${attempt + 1}/3): ${result.error}`);
      }
    }
    
    if (!verified) {
      throw new Error(`任务 ${task.id} 在 3 次尝试后仍未通过验证`);
    }
  }
 
  // Phase 3: 最终验证
  console.log('🎯 运行最终验证...');
  await runCommand({ command: 'npm run test && npm run lint && npm run build' });
}

你是一个高级编码 Agent，使用 Plan-Act-Verify 模式完成任务。

## 任务
[具体需求描述]

## 执行流程

### Phase 1: Plan
分析需求，输出结构化的执行计划：
- 列出所有需要修改/创建的文件
- 将任务分解为有序的子步骤
- 为每个步骤指定验证方法（测试/lint/类型检查/构建）

### Phase 2: Act + Verify
对每个子步骤：
1. 执行代码修改
2. 运行指定的验证命令
3. 如果验证失败，分析错误并修正（最多重试 3 次）

### Phase 3: Final Verification
所有子步骤完成后：
1. 运行完整测试套件
2. 运行 lint 和类型检查
3. 确认构建成功

## 约束
- 遵循项目现有的代码风格
- 每次修改后立即验证，不要积累未验证的改动
- 如果计划需要调整，先说明原因再修改

def ooda_loop(agent_config: AgentConfig) -> None:
    """OODA 循环 — 持续运行的事件驱动 Agent"""
    mental_model = load_mental_model(agent_config)  # 历史经验和知识
    
    while agent_config.is_running:
        # ===== Observe: 观察环境 =====
        observations = collect_observations(
            sources=agent_config.data_sources,  # 日志、指标、事件流
            timeframe="last_5_minutes"
        )
        
        # ===== Orient: 定向分析 =====
        situation = llm.generate(
            prompt=f"""基于以下观察数据，分析当前态势：
            观察数据：{observations}
            历史模式：{mental_model.recent_patterns}
            已知问题：{mental_model.known_issues}
            
            输出：
            1. 当前状态评估（正常/警告/危险）
            2. 异常检测结果
            3. 根因分析（如有异常）""",
        )
        
        # ===== Decide: 决策 =====
        if situation.status == "normal":
            continue  # 无需行动，继续观察
        
        decision = llm.generate(
            prompt=f"""基于态势分析，选择行动方案：
            态势：{situation}
            可用行动：{agent_config.available_actions}
            风险容忍度：{agent_config.risk_tolerance}
            
            输出：
            1. 推荐行动及理由
            2. 风险评估
            3. 是否需要人工审批""",
        )
        
        # ===== Act: 执行 =====
        if decision.requires_approval:
            await request_human_approval(decision)
        
        result = execute_action(decision.action)
        
        # 更新心智模型（学习）
        mental_model.update(observations, situation, decision, result)

// OODA 模式的安全监控 Agent
import { generateObject } from 'ai';
import { anthropic } from '@ai-sdk/anthropic';
import { z } from 'zod';
 
const SituationSchema = z.object({
  status: z.enum(['normal', 'warning', 'critical']),
  anomalies: z.array(z.string()),
  rootCause: z.string().optional(),
  confidence: z.number().min(0).max(1),
});
 
const DecisionSchema = z.object({
  action: z.string(),
  reason: z.string(),
  risk: z.enum(['low', 'medium', 'high']),
  requiresApproval: z.boolean(),
});
 
async function oodaSecurityAgent() {
  const mentalModel = new MentalModel(); // 历史经验存储
 
  while (true) {
    // Observe: 收集安全事件
    const observations = await collectSecurityEvents({
      sources: ['waf_logs', 'auth_logs', 'network_flows'],
      timeframe: '5m',
    });
 
    // Orient: 态势分析
    const { object: situation } = await generateObject({
      model: anthropic('claude-sonnet-4-20250514'),
      schema: SituationSchema,
      prompt: `分析安全态势：\n${JSON.stringify(observations)}\n历史模式：${mentalModel.summary()}`,
    });
 
    if (situation.status === 'normal') {
      await sleep(30_000); // 30 秒后再次观察
      continue;
    }
 
    // Decide: 选择响应方案
    const { object: decision } = await generateObject({
      model: anthropic('claude-sonnet-4-20250514'),
      schema: DecisionSchema,
      prompt: `基于态势选择响应：\n${JSON.stringify(situation)}`,
    });
 
    // Act: 执行响应
    if (decision.requiresApproval) {
      await notifySecurityTeam(decision);
      await waitForApproval(decision);
    }
    
    await executeResponse(decision.action);
    mentalModel.learn(observations, situation, decision);
    
    await sleep(10_000); // 10 秒后继续循环
  }
}

你是一个持续运行的安全监控 Agent，使用 OODA 循环模式。

## 当前观察数据
[最近 5 分钟的日志/指标/事件]

## 历史上下文
[最近 24 小时的趋势和已知问题]

## 你的任务

### Observe（已完成，数据如上）

### Orient — 态势分析
基于观察数据和历史上下文：
1. 评估当前状态（正常/警告/危险）
2. 识别异常模式
3. 如有异常，分析可能的根因

### Decide — 决策
如果状态非正常：
1. 列出可选的响应方案（最多 3 个）
2. 评估每个方案的风险和收益
3. 推荐最佳方案
4. 判断是否需要人工审批（高风险操作必须审批）

### Act — 行动建议
输出具体的执行命令或 API 调用。

## 约束
- 风险等级为"高"的操作必须等待人工审批
- 同一问题 1 小时内最多自动响应 3 次
- 所有行动必须记录审计日志

def tool_calling_loop(
    messages: list[Message],
    tools: list[ToolDefinition],
    max_turns: int = 10
) -> str:
    """Tool-Calling 循环的核心实现"""
    
    for turn in range(max_turns):
        # 1. 调用 LLM，传入消息历史和工具定义
        response = llm.chat(
            messages=messages,
            tools=tools,           # 结构化工具定义（JSON Schema）
            tool_choice="auto",    # auto | required | none | specific_tool
        )
        
        # 2. 检查响应类型
        if response.stop_reason == "end_turn":
            # LLM 决定直接回复文本
            return response.text
        
        if response.stop_reason == "tool_use":
            # 3. 处理工具调用（可能有多个并行调用）
            tool_results = []
            for tool_call in response.tool_calls:
                # 4. Schema 验证
                validated = validate_params(tool_call.name, tool_call.arguments)
                
                # 5. 执行工具
                try:
                    result = execute_tool(tool_call.name, validated)
                    tool_results.append({
                        "tool_call_id": tool_call.id,
                        "content": json.dumps(result)
                    })
                except Exception as e:
                    tool_results.append({
                        "tool_call_id": tool_call.id,
                        "content": f"Error: {str(e)}",
                        "is_error": True
                    })
            
            # 6. 将工具结果注入消息历史
            messages.append(response.message)  # assistant 消息（含 tool_use）
            messages.append({"role": "tool", "content": tool_results})
    
    return "达到最大轮次限制"

// 使用 OpenAI SDK 的 Tool-Calling 模式
import OpenAI from 'openai';
 
const openai = new OpenAI();
 
// 定义工具（JSON Schema）
const tools: OpenAI.ChatCompletionTool[] = [
  {
    type: 'function',
    function: {
      name: 'get_weather',
      description: '获取指定城市的天气信息',
      parameters: {
        type: 'object',
        properties: {
          city: { type: 'string', description: '城市名称' },
          unit: { type: 'string', enum: ['celsius', 'fahrenheit'] },
        },
        required: ['city'],
      },
    },
  },
  {
    type: 'function',
    function: {
      name: 'search_database',
      description: '查询产品数据库',
      parameters: {
        type: 'object',
        properties: {
          query: { type: 'string' },
          limit: { type: 'number', default: 10 },
        },
        required: ['query'],
      },
    },
  },
];
 
// 工具执行映射
const toolHandlers: Record<string, (args: any) => Promise<string>> = {
  get_weather: async ({ city, unit }) => {
    const data = await fetchWeatherAPI(city, unit);
    return JSON.stringify(data);
  },
  search_database: async ({ query, limit }) => {
    const results = await db.search(query, limit);
    return JSON.stringify(results);
  },
};
 
async function toolCallingAgent(userMessage: string): Promise<string> {
  const messages: OpenAI.ChatCompletionMessageParam[] = [
    { role: 'user', content: userMessage },
  ];
 
  for (let turn = 0; turn < 10; turn++) {
    const response = await openai.chat.completions.create({
      model: 'gpt-4o',
      messages,
      tools,
      tool_choice: 'auto',
    });
 
    const choice = response.choices[0];
    messages.push(choice.message);
 
    // 如果没有工具调用，返回文本回复
    if (!choice.message.tool_calls?.length) {
      return choice.message.content ?? '';
    }
 
    // 执行所有工具调用（支持并行）
    const toolResults = await Promise.all(
      choice.message.tool_calls.map(async (tc) => {
        const handler = toolHandlers[tc.function.name];
        const args = JSON.parse(tc.function.arguments);
        const result = await handler(args);
        return {
          role: 'tool' as const,
          tool_call_id: tc.id,
          content: result,
        };
      })
    );
 
    messages.push(...toolResults);
  }
 
  return '达到最大轮次限制';
}

你是一个客户服务 Agent，可以使用以下工具帮助客户解决问题。

## 可用工具
工具定义已通过 API 的 tools 参数提供，包括：
- query_account: 查询客户账户信息
- modify_order: 修改订单状态
- create_ticket: 创建工单
- search_knowledge_base: 搜索知识库

## 使用规则
1. 先查询相关信息，再执行操作
2. 修改类操作前，先向客户确认
3. 如果工具调用失败，向客户解释并提供替代方案
4. 涉及退款超过 [金额] 的操作，告知客户需要人工处理

## 回复风格
- 使用友好、专业的语气
- 每次回复简洁明了
- 操作完成后总结所做的事情

用户输入：为 /api/users 端点添加游标分页功能

===== Phase 1: Plan =====

Claude Code 分析需求，生成计划：
1. 读取现有的 users 路由和数据库查询代码
2. 修改数据库查询，添加游标分页参数（cursor, limit）
3. 修改 API 路由，解析分页查询参数
4. 添加分页响应格式（data, nextCursor, hasMore）
5. 编写单元测试覆盖分页逻辑
6. 运行测试验证

===== Phase 2: Act + Verify =====

--- 子任务 1: 读取现有代码 ---
Act:  读取 src/routes/users.ts 和 src/db/queries.ts
Verify: ✅ 文件存在，理解了现有结构

--- 子任务 2: 修改数据库查询 ---
Act:  在 queries.ts 中添加 getUsersWithCursor() 函数
Verify: 运行 tsc --noEmit → ✅ 类型检查通过

--- 子任务 3: 修改 API 路由 ---
Act:  更新 users.ts 路由处理函数
Verify: 运行 tsc --noEmit → ✅ 类型检查通过

--- 子任务 4: 添加响应格式 ---
Act:  定义 PaginatedResponse<T> 类型
Verify: 运行 tsc --noEmit → ✅ 类型检查通过

--- 子任务 5: 编写测试 ---
Act:  创建 src/routes/users.test.ts
Verify: 运行 npm test -- users.test → ❌ 测试失败
  → 诊断：游标解码逻辑有 bug
  → 修复：修正 base64 解码
  → 重新验证：运行 npm test -- users.test → ✅ 通过

===== Phase 3: Final Verification =====
运行 npm test → ✅ 全部 47 个测试通过
运行 npm run lint → ✅ 无警告
运行 npm run build → ✅ 构建成功

📦 交付：3 个文件修改，1 个新文件，47/47 测试通过