Agent Loop · 数据流可视化

# CLI 参数解析 → 环境变量加载 → Agent 构建 → 任务执行 def main(): parser = argparse.ArgumentParser(description="Agent Loop v1.3") parser.add_argument("task", nargs="*") parser.add_argument("--model", "") parser.add_argument("--api-key", "") # ... 其他参数 # LLM 配置 config = LLMConfig( api_key=args.api_key or LLMConfig.api_key, model=args.model or "deepseek-v4-flash", ) # 预算控制（三重上限） budget = Budget( max_iterations=args.max_iter or 25, max_duration=args.max_duration or 600, ) # Agent 实例化并运行 agent = Agent(llm, tools, system_prompt=system, verifier=verifier, budget=budget) result = agent.run(task, verbose=not args.quiet, checkpoint=args.checkpoint)

# 主循环：反复执行 Think → Act → Observe → Verify for i in range(1, self.budget.max_iterations + 1): self.state.iteration = i # ── Think: LLM 调用 ── resp = self.llm.chat( self.messages, tools=list(self.tools.values()), ← 注册所有工具 ) # 保留完整消息（含 tool_calls） assistant_msg = {"role": "assistant", "content": resp.get("content", "")} if resp.get("tool_calls"): assistant_msg["tool_calls"] = resp["tool_calls"] self.messages.append(assistant_msg)

# ── Act: 执行工具 ── if resp.get("tool_calls"): for tc in resp.get("_parsed_tool_calls", []): tool = self.tools.get(tc["name"]) ← 按名查找工具 if not tool: result = f"[Unknown tool: {tc['name']}]" else: result = tool.execute(**tc['args']) ← 调用 handler # Observe: 结果注入消息 self.messages.append({ "role": "tool", "tool_call_id": tc["id"], "content": result, })

# 工具结果追加到 messages → 下轮 Think 可见 # 形成闭环：Think → Act → Observe → Think self.messages.append({ "role": "tool", "tool_call_id": tc["id"], "content": result, }) # 循环回到 for 顶部，开始下一轮 Think

# LLM 没有 tool_calls → 直接输出答案 → 验证 if not resp.get("tool_calls"): answer = resp.get("content", "") # ── Verify: 质量检查 ── if self.verifier.verify(answer): self.state.status = "done" self.state.save() return answer ← 完成，输出结果 else: # 验证失败 → 注入反馈重试 self._retry_count += 1 if self._retry_count > self.budget.max_retries: return answer + "\\n⚠️ Verification failed" self.messages.append({ "role": "user", "content": f"质量检查未通过：{self.verifier.summary()}" }) continue ← 回到 Think 重试

# 任务完成 → 输出结果 + 状态持久化 # 状态更新 self.state.status = "done" self.state.summary = f"Completed in {i} iterations" self.state.save() # 返回最终答案 return answer # ── main() 中打印输出 ── print(f"\\n{'='*50}") print(f"📋 FINAL RESULT [{agent.state.status}]") print(result)