构建真正能工作的自主 AI 代理

Source: Dev.to

“AI 驱动”工具的问题

现在每个 SaaS 产品的着陆页上都贴上“AI 驱动”的标签，但其中 99 % 只是在单次 LLM 调用之上加了一个薄薄的 UI。你仍然需要：

1. 决定要做什么
2. 编写提示词
3. 审核输出
4. 决定下一步要做什么
5. 永久循环

这并不是自动化——而是你在做所有思考，机器只负责敲键盘。

实际的代理是什么样子的

一个自主代理遵循 循环，而不是单一的提示‑响应。 我使用的模式基于 ReAct 框架：

┌─────────────────────────────────────┐
│           AGENT LOOP                │
│                                     │
│  ┌──────────┐                       │
│  │ PERCEIVE │◄── APIs, databases,   │
│  └────┬─────┘    live data sources  │
│       │                             │
│  ┌────▼─────┐                       │
│  │  REASON  │◄── LLM + context      │
│  └────┬─────┘    from RAG store     │
│       │                             │
│  ┌────▼─────┐                       │
│  │   ACT    │──► API calls, DB      │
│  └────┬─────┘    writes, alerts     │
│       │                             │
│       └──────────► loop back ◄──────│
└─────────────────────────────────────┘

每一次迭代，代理会：

1. 感知其环境（API、数据库、实时数据）。
2. 使用 LLM 加上任何检索到的上下文进行 推理。
3. 通过调用 API、写入数据库、发送警报等方式 行动。

关键洞见：决定 何时停止 的是代理，而不是你。

架构：多代理而非单体

单个“大”代理很快会产生幻觉并丢失上下文。相反，我将职责划分为 专门的子代理。

from dataclasses import dataclass
from enum import Enum

class AgentRole(Enum):
    RESEARCHER = "researcher"
    AUDITOR    = "auditor"
    STRATEGIST = "strategist"
    EXECUTOR   = "executor"

@dataclass
class AgentTask:
    role: AgentRole
    objective: str
    constraints: list[str]
    context: dict

class AgentOrchestrator:
    def __init__(self, agents: dict[AgentRole, "Agent"]):
        self.agents = agents
        self.shared_state: dict = {}

    async def run_pipeline(self, trigger: dict):
        # 1️⃣ Researcher gathers data
        research = await self.agents[AgentRole.RESEARCHER].execute(
            AgentTask(
                role=AgentRole.RESEARCHER,
                objective="Analyze SERP changes for target keywords",
                constraints=["Use DataForSEO API", "Max 500 queries"],
                context=trigger,
            )
        )

        # 2️⃣ Auditor validates against guidelines
        audit = await self.agents[AgentRole.AUDITOR].execute(
            AgentTask(
                role=AgentRole.AUDITOR,
                objective="Check findings against brand guidelines",
                constraints=["Flag confidence  0.8:
            await self.agents[AgentRole.EXECUTOR].execute(
                AgentTask(
                    role=AgentRole.EXECUTOR,
                    objective="Implement approved changes",
                    constraints=["Dry‑run first", "Log all mutations"],
                    context={"strategy": strategy},
                )
            )

每个子代理的范围都很窄：

• Researcher 从不编写内容。
• Executor 从不决定策略。

这种分离可以将幻觉限制在局部范围。

每个子代理内部的 ReAct 循环

每个子代理都使用 ReAct 模式运行自己的推理循环——在每次 Action 之前生成一个 Thought。

import json
import openai
from typing import Callable

class Agent:
    def __init__(self, role: AgentRole, tools: list[Callable]):
        self.role = role
        self.tools = {t.__name__: t for t in tools}
        self.client = openai.AsyncOpenAI()

    async def execute(self, task: AgentTask, max_steps: int = 10):
        messages = [
            {"role": "system", "content": self._build_system_prompt(task)},
        ]

        for step in range(max_steps):
            response = await self.client.chat.completions.create(
                model="gpt-4o",
                messages=messages,
                tools=self._tool_schemas(),
            )
            message = response.choices[0].message

            # If the LLM decides to call a tool, run it and feed the result back
            if message.tool_calls:
                for call in message.tool_calls:
                    tool_name = call.function.name
                    args = json.loads(call.function.arguments)
                    result = await self.tools[tool_name](**args)
                    messages.append({"role": "tool", "content": str(result)})
                continue

            # No tool call → final answer
            messages.append({"role": "assistant", "content": message.content})
            break

        return messages[-1]["content"]

    # -----------------------------------------------------------------
    # Helper methods (implementation details omitted for brevity)
    # -----------------------------------------------------------------
    def _build_system_prompt(self, task: AgentTask) -> str:
        return (
            f"You are a {task.role.value} agent.\n"
            f"Objective: {task.objective}\n"
            f"Constraints: {', '.join(task.constraints)}\n"
            f"Context: {json.dumps(task.context)}"
        )

    def _tool_schemas(self) -> list[dict]:
        """Return OpenAI‑compatible tool specifications for self.tools."""
        # Placeholder – in a real implementation you would generate JSON schema
        # objects for each callable in self.tools.
        return []

每个代理：

1. 接收一个 system prompt，其中定义了它的角色、目标、约束和上下文。
2. 生成一个 thought（LLM 的下一条消息）。
3. 如果该 thought 包含 tool call，则执行相应工具，并将结果作为 tool 消息反馈回去。
4. 循环重复，直到 LLM 给出最终答案（没有工具调用）。

要点

• 基于循环的代理（感知 → 推理 → 行动）远比单提示 UI 更强大。
• 专用子代理使整体系统保持稳固并减少幻觉。
• ReAct 模式为每个代理提供了严格的推理循环，使其能够决定何时调用外部工具以及何时停止。

有了这种架构，你可以让自主的 SEO 助手整夜运行，持续监控 SERP，审计品牌合规性，制定策略并执行更改——全部无需人工干预。 🚀

示例工作流片段

整个流程每 24 小时运行一次。我会收到一条 Slack 通知，内容包括它发现了什么以及执行了什么。大多数日子它什么也没发现；有些日子它会在我手动注意到之前数周捕捉到衰退模式。

什么使它不同于 Cron 脚本

Cron 脚本每次执行相同的操作。代理 会对其看到的内容进行推理。

• 当 SERP 从列表文章（listicles）转变为视频轮播（video carousels）时，cron 脚本仍会继续优化列表文章。
• 代理会注意到这种格式的变化，并相应调整其推荐。

这种区别很重要：你定义目标和约束，而不是一步一步的指令。代理会自行找出步骤。

实际重要的防护措施

• Dry‑run 模式 – 每一次变更在执行前都会被记录。代理提出建议；由人类（或第二个代理）批准。
• 置信度阈值 – 置信度低于 0.7 的操作会被放入队列，供人工审查，而不是自动执行。
• 审计日志 – 每一次思考、工具调用和结果都会被记录。没有黑箱。
• 预算上限 – 每次运行的最大 API 调用次数、每个代理的最大 token 数、每个周期的最大变更次数。

目标不是“让人类脱离循环”，而是“让人类参与循环”。你设定边界，代理在这些边界内运行，你则审阅仪表盘。

入门

你不需要在第一天就构建一个复杂的多代理系统。先从一个只做 一件事 的单一代理开始：

1. 选取一个你每周都会重复的任务。
2. 编写一个 Python 脚本，实现 感知 步骤（获取数据）。
3. 为 推理 步骤添加一次 LLM 调用（分析数据）。
4. 为 行动 步骤添加一次 API 调用（基于分析执行操作）。
5. 用循环将其包装起来，并设置终止条件。

这就是一个代理。其余的都是优化。

进一步阅读

深入了解完整架构细节：
Agentic AI Workflows: Beyond Basic Content Generation