# 从零构建一个可审查的 Agentic AI 系统：规划、工具调用、记忆与自我评估

- 原文：How to Build an Advanced Agentic AI System with Planning, Tool Calling, Memory, and Self-Critique Using OpenAI API
- 原文 URL：https://www.marktechpost.com/2026/05/18/how-to-build-an-advanced-agentic-ai-system-with-planning-tool-calling-memory-and-self-critique-using-openai-api/
- 作者：Sana Hassan
- 发布时间：2026-05-19
- 本文依据：已保存的 `/gsummary` 摘要与直接 HTML 正文抽取结果
- 边界说明：原文是教学型 MVP。下面的教程保留其核心结构，并补充了更适合本地验证、团队复用和低风险落地的工程化步骤；补充内容会明确标为“实践扩展”。

## 一句话结论

一个可落地的 Agentic AI 系统，不应只是“给大模型一个长 Prompt”。更稳妥的做法是把任务拆成四层：

1. **Planner：先规划**，把模糊目标转成结构化步骤。
2. **Executor：再执行**，按计划调用受控工具。
3. **Trace / Memory：全过程留痕**，记录目标、短期记忆、工具调用和结果。
4. **Critic：最后审查**，基于目标和执行轨迹修正输出。

这套结构适合做会议纪要、报告生成、运维排查建议、知识库问答、轻量自动化助手等“需要多步推理但不能完全放任模型”的场景。

## 适合谁读

适合：

- 想把普通 ChatGPT/OpenAI API 调用升级为“会规划、会调工具、能留痕”的开发者。
- 想给内部脚本加上 LLM 规划层，但又担心幻觉和不可审计的团队。
- 想做一个最小可行 Agent 原型，并逐步加评估、权限、安全边界的人。

不适合：

- 需要完全无人值守生产自动化的场景。本文只是 MVP 起点，不包含完整权限隔离、沙箱、回滚、审计平台和评估集。
- 需要浏览器自动操作、数据库写入、Kubernetes 操作等高风险动作的生产系统。那些必须先做只读试跑和人工确认。

## 1. 核心架构

原文的核心架构可以抽象成下面这条流水线：

```text
用户目标
  ↓
Planner：生成结构化计划 JSON
  ↓
Executor：按计划调用工具，生成草稿
  ↓
Trace / Memory：记录中间过程
  ↓
Critic：审查草稿，生成最终结果
  ↓
可交付物：文本、JSON、文件、报告
```

关键思想不是“让模型更自由”，而是**给模型更清楚的职责边界**：

- Planner 只负责“下一步应该怎么做”。
- Executor 只负责“按计划执行，并在必要时调用工具”。
- Critic 只负责“基于目标和执行记录检查质量”。
- 工具层只暴露白名单函数，不让模型直接碰系统命令、数据库或文件系统。

## 2. 最小可运行项目结构

实践扩展：建议先做一个本地只读/低风险版本，目录如下：

```text
agentic-ai-demo/
├── agent.py              # 主流程：plan -> execute -> critique
├── tools.py              # 工具白名单
├── prompts.py            # Planner / Executor / Critic 提示词
├── models.py             # AgentState 等数据结构
├── examples/
│   ├── meeting.txt       # 示例输入 1：会议记录
│   └── ops_incident.txt  # 示例输入 2：运维故障描述
├── outputs/              # 生成结果
├── .env.example          # 环境变量模板，不提交真实 key
└── requirements.txt
```

`requirements.txt`：

```text
openai>=1.0.0
python-dotenv>=1.0.0
```

安装命令：

```bash
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
cp .env.example .env
```

`.env.example`：

```bash
OPENAI_API_KEY=your-api-key-here
OPENAI_MODEL=gpt-5.2
```

> 安全提示：真实 API key 只能写入 `.env` 或运行时环境变量，不要写进代码、Prompt、日志或示例输出。

## 3. 定义状态：目标、记忆与执行轨迹

原文使用 `AgentState` 保存目标、记忆和 Trace。一个简单版本如下：

```python
# models.py
from dataclasses import dataclass, field
from typing import Any


@dataclass
class AgentState:
    goal: str
    memory: list[str] = field(default_factory=list)
    trace: list[dict[str, Any]] = field(default_factory=list)

    def add_trace(self, tool: str, args: dict[str, Any], result: dict[str, Any]) -> None:
        self.trace.append({
            "tool": tool,
            "args": args,
            "result": result,
        })
```

这里的 `trace` 很重要。它不是普通日志，而是给 Critic 审查用的事实依据。没有 Trace，Critic 很容易只按最终文本“感觉不错”来评价；有 Trace，它可以看到工具是否失败、参数是否错误、结果是否被误读。

## 4. 定义工具白名单

原文示例包含四类工具：

- `calc`：安全计算。
- `kb_search`：搜索内部知识库。
- `extract_json`：从模型输出中提取 JSON。
- `write_file`：保存交付物。

下面是一个简化但可复制的版本：

```python
# tools.py
import ast
import hashlib
import json
import os
import re
from typing import Any


KB = [
    {
        "title": "会议纪要规范",
        "text": "会议纪要必须包含：结论、行动项、负责人、截止时间、风险。",
    },
    {
        "title": "输出质量规范",
        "text": "最终答案必须包含步骤、检查项和可交付物。邮件必须包含主题和下一步。",
    },
]


def safe_calc(expression: str) -> dict[str, Any]:
    allowed = set("0123456789+-*/().% ")
    if any(ch not in allowed for ch in expression):
        return {"ok": False, "error": "invalid characters"}

    try:
        tree = ast.parse(expression, mode="eval")
        for node in ast.walk(tree):
            if not isinstance(
                node,
                (
                    ast.Expression,
                    ast.BinOp,
                    ast.UnaryOp,
                    ast.Constant,
                    ast.Add,
                    ast.Sub,
                    ast.Mult,
                    ast.Div,
                    ast.Mod,
                    ast.Pow,
                    ast.USub,
                    ast.UAdd,
                    ast.Load,
                ),
            ):
                return {"ok": False, "error": f"unsupported node: {type(node).__name__}"}
        value = eval(compile(tree, "<calc>", "eval"), {"__builtins__": {}}, {})
        return {"ok": True, "expression": expression, "value": value}
    except Exception as exc:
        return {"ok": False, "error": str(exc)}


def kb_search(query: str, k: int = 3) -> dict[str, Any]:
    tokens = set(re.findall(r"\w+", query.lower()))
    scored = []
    for item in KB:
        haystack = f"{item['title']} {item['text']}".lower()
        score = sum(1 for token in tokens if token in haystack)
        scored.append((score, item))
    scored.sort(key=lambda pair: pair[0], reverse=True)
    return {"ok": True, "results": [item for score, item in scored[:k] if score > 0]}


def extract_json(text: str) -> dict[str, Any]:
    match = re.search(r"\{.*\}", text, flags=re.DOTALL)
    if not match:
        return {"ok": False, "error": "no json object found"}
    try:
        return {"ok": True, "json": json.loads(match.group(0))}
    except json.JSONDecodeError as exc:
        return {"ok": False, "error": str(exc), "raw": match.group(0)[:1000]}


def write_file(path: str, content: str) -> dict[str, Any]:
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    with open(path, "w", encoding="utf-8") as file:
        file.write(content)
    sha16 = hashlib.sha256(content.encode("utf-8")).hexdigest()[:16]
    return {"ok": True, "path": path, "sha16": sha16, "bytes": len(content.encode("utf-8"))}


TOOLS = {
    "calc": safe_calc,
    "kb_search": kb_search,
    "extract_json": extract_json,
    "write_file": write_file,
}
```

### 工具设计原则

- 工具返回值必须是结构化对象，至少包含 `ok` 字段。
- 文件写入必须返回路径、字节数、hash，便于后续校验。
- 不要把 `shell`、`eval 任意代码`、数据库写入、生产 API 写操作直接暴露给模型。
- 高风险工具必须先做只读版本，再做人工确认版本，最后才考虑自动执行。

## 5. 定义 OpenAI 工具 Schema

工具 Schema 是模型和 Python 函数之间的契约。示例：

```python
# tools.py 追加
TOOL_SCHEMAS = [
    {
        "type": "function",
        "function": {
            "name": "calc",
            "description": "Safely compute a numeric expression.",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string"},
                },
                "required": ["expression"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "kb_search",
            "description": "Search internal knowledge base.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "k": {"type": "integer", "default": 3},
                },
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "write_file",
            "description": "Write content to a file path.",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string"},
                    "content": {"type": "string"},
                },
                "required": ["path", "content"],
            },
        },
    },
]
```

## 6. 三段 Prompt 模板

### 6.1 Planner Prompt

```text
你是 Planner。你的任务是把用户目标拆成可执行计划。

要求：
1. 只输出 JSON，不要输出解释。
2. JSON 必须包含：goal、steps、needed_tools、success_criteria。
3. steps 每一步必须短、可验证。
4. 如果目标涉及文件写入或外部副作用，必须在 steps 中加入“先生成草稿并等待确认”。

用户目标：
{{GOAL}}
```

期望输出形状：

```json
{
  "goal": "生成会议纪要、行动项 JSON 和跟进邮件",
  "steps": [
    "读取会议记录并识别决策",
    "提取行动项、负责人和截止时间",
    "生成跟进邮件草稿",
    "保存结果到指定文件"
  ],
  "needed_tools": ["kb_search", "write_file"],
  "success_criteria": [
    "行动项包含 owner/action/due_date 字段",
    "邮件包含 subject 和 body",
    "文件写入返回 ok=true"
  ]
}
```

### 6.2 Executor Prompt

```text
你是 Executor。你必须按 Planner 的计划执行。

约束：
1. 只在必要时调用工具。
2. 不要编造工具结果。
3. 如果工具失败，说明失败并尝试一次低风险修正。
4. 最终输出必须包含：结果草稿、已调用工具、未完成事项。

计划：
{{PLAN_JSON}}

用户目标：
{{GOAL}}
```

### 6.3 Critic Prompt

```text
你是 Critic。请基于用户目标、执行草稿和工具 Trace 审查输出质量。

审查维度：
1. 是否满足用户目标。
2. 是否遗漏必要字段。
3. 是否有未经工具支持的断言。
4. 工具失败是否被正确处理。
5. 最终结果是否可以交付。

用户目标：
{{GOAL}}

执行草稿：
{{DRAFT}}

工具 Trace：
{{TRACE}}

请输出：
- 问题列表
- 修正后的最终版本
- 是否通过：pass/fail
```

## 7. 主流程伪代码

下面是整个 Agent 的最小流程：

```python
# agent.py
import json
import os
from openai import OpenAI
from dotenv import load_dotenv

from models import AgentState
from tools import TOOLS, TOOL_SCHEMAS

load_dotenv()
client = OpenAI()
MODEL = os.getenv("OPENAI_MODEL", "gpt-5.2")


def chat(messages: list[dict], tools: list[dict] | None = None):
    kwargs = {
        "model": MODEL,
        "messages": messages,
        "temperature": 0.2,
    }
    if tools:
        kwargs["tools"] = tools
        kwargs["tool_choice"] = "auto"
    return client.chat.completions.create(**kwargs)


def run_tool(state: AgentState, name: str, args: dict):
    tool = TOOLS.get(name)
    if not tool:
        result = {"ok": False, "error": f"unknown tool: {name}"}
    else:
        try:
            result = tool(**args)
        except Exception as exc:
            result = {"ok": False, "error": str(exc)}
    state.add_trace(name, args, result)
    return result


def run_agent(goal: str) -> dict:
    state = AgentState(goal=goal)
    state.memory.append("必要时先检索内部规范；涉及文件写入时保留路径和 hash。")

    # 1. Planner
    plan_response = chat([
        {"role": "system", "content": "你是 Planner，只输出 JSON。"},
        {"role": "user", "content": f"请为目标生成执行计划：{goal}"},
    ])
    plan_text = plan_response.choices[0].message.content or "{}"
    plan = json.loads(plan_text)

    # 2. Executor（真实实现中应处理 tool_calls，这里省略 SDK 细节）
    draft_response = chat([
        {"role": "system", "content": "你是 Executor，按计划执行并生成草稿。"},
        {"role": "user", "content": json.dumps(plan, ensure_ascii=False)},
    ], tools=TOOL_SCHEMAS)
    draft = draft_response.choices[0].message.content or ""

    # 3. Critic
    critique_response = chat([
        {"role": "system", "content": "你是 Critic，基于目标、草稿和 Trace 审查。"},
        {"role": "user", "content": json.dumps({
            "goal": goal,
            "draft": draft,
            "trace": state.trace[-20:],
        }, ensure_ascii=False)},
    ])
    final = critique_response.choices[0].message.content or draft

    return {
        "plan": plan,
        "draft": draft,
        "final": final,
        "trace": state.trace,
    }
```

> 注意：上面的代码保留了结构骨架。真实 OpenAI tool calling 需要处理 `message.tool_calls`，再把工具结果作为 `tool` role 消息传回模型。不要把这段伪代码当作完整 SDK 适配层。

## 8. 案例一：会议纪要 Agent

### 样本输入

保存为 `examples/meeting.txt`：

```text
Decision: Ship v2 dashboard on March 15.
Risk: Data latency might spike; Priya will run load tests.
Amir will update the KPI definitions doc and share with finance.
Next check-in: Tuesday. Owner: Nikhil.
```

### 目标 Prompt

```text
请基于会议记录生成：
A) 简洁会议摘要
B) JSON 行动项数组，字段为 owner、action、due_date
C) 跟进邮件，包含 subject 和 body
D) 保存到 outputs/meeting_followup.md
```

### 期望输出形状

```json
{
  "summary": "团队决定在 March 15 发布 v2 dashboard，但需要关注数据延迟风险。",
  "action_items": [
    {
      "owner": "Priya",
      "action": "Run load tests for data latency risk",
      "due_date": null
    },
    {
      "owner": "Amir",
      "action": "Update KPI definitions doc and share with finance",
      "due_date": null
    },
    {
      "owner": "Nikhil",
      "action": "Lead next check-in on Tuesday",
      "due_date": "Tuesday"
    }
  ],
  "email": {
    "subject": "Follow-up: v2 dashboard launch and next actions",
    "body": "..."
  }
}
```

### 验收标准

- JSON 可以被 `json.loads()` 解析。
- 每个行动项都有 `owner`、`action`、`due_date`。
- 生成文件存在，且 `write_file` 返回 `ok=true`、`bytes>0`、`sha16` 非空。
- Critic 没有发现“缺字段”或“未保存文件”。

### 失败处理

- 如果行动项缺 `due_date` 字段，即使没有日期也要填 `null`。
- 如果文件写入失败，不要声称保存成功；最终结果必须展示失败原因。
- 如果 Planner 输出不是 JSON，先调用 `extract_json` 尝试修复一次；仍失败则中止。

## 9. 案例二：运维故障分析 Agent（只读版）

实践扩展：不要一开始就让 Agent 执行 `kubectl delete`、重启服务或修改配置。先做只读分析版。

### 样本输入

保存为 `examples/ops_incident.txt`：

```text
服务 httpapi 最近 30 分钟 P95 延迟从 120ms 上升到 2.5s。
错误率从 0.2% 上升到 3%。
最近一次变更：新增数据库查询统计接口。
数据库 CPU 80%，慢查询数量增加。
```

### 目标 Prompt

```text
请分析这个故障描述，输出：
1. 最可能原因排序
2. 需要补充的只读检查命令
3. 不能直接执行的高风险动作
4. 给值班同学的短消息草稿
```

### 期望输出示例

```text
最可能原因：
1. 新增统计接口触发慢查询，导致数据库 CPU 升高。
2. 数据库连接池等待增加，放大接口 P95 延迟。
3. 下游超时导致错误率上升。

只读检查命令：
- 查看慢查询日志
- 查看接口维度延迟
- 查看数据库连接池等待

禁止直接执行：
- 不要直接 kill 数据库连接
- 不要直接重启生产服务
- 不要直接回滚，除非确认变更关联并得到审批
```

### 验收标准

- 输出必须区分“已知事实”和“推测”。
- 所有建议命令必须是只读检查。
- 高风险动作必须列入禁止或需确认清单。
- 不得编造具体机器 IP、库名、Pod 名。

### 失败处理

- 如果输入信息不足，Agent 应输出“需要补充的证据”，而不是强行给确定结论。
- 如果模型建议破坏性动作，Critic 必须判定 fail，并要求改成只读检查。

## 10. 案例三：知识库问答 Agent

实践扩展：把原文的 `kb_search` 替换成更贴近团队场景的本地文档检索，先从关键词检索开始，不急着上向量库。

### 样本知识库

```python
KB = [
    {
        "title": "SQL 变更规范",
        "text": "高风险 DDL 必须先审查；执行前需要备份；禁止无 WHERE 的 DELETE。",
    },
    {
        "title": "告警通知规范",
        "text": "企业微信通知超过 4000 字符需要分页；故障通知必须包含影响范围和下一步。",
    },
]
```

### 目标 Prompt

```text
用户问：我准备执行一个 DELETE FROM users;，需要注意什么？
请先检索知识库，再给出风险判断和下一步建议。
```

### 期望输出

```text
风险判断：高风险，不能直接执行。
依据：知识库“SQL 变更规范”明确禁止无 WHERE 的 DELETE。
建议：
1. 停止执行当前 SQL。
2. 补充 WHERE 条件和影响行数预估。
3. 走审查、备份、审批流程。
```

### 验收标准

- 必须引用命中的知识库标题。
- 必须明确“不能直接执行”。
- 不得把危险 SQL 改写成另一个仍危险的 SQL。

### 失败处理

- 如果 `kb_search` 没命中，输出“知识库未命中”，并只给通用安全建议。
- 如果用户要求绕过审批，Agent 必须拒绝并解释风险。

## 11. 质量门禁：上线前至少检查这些

### 11.1 Planner 检查

- 是否总能输出可解析 JSON？
- 是否包含成功标准？
- 是否把高风险动作改成“先生成草稿/等待确认”？

### 11.2 Tool 检查

- 工具是否有白名单？
- 工具返回是否结构化？
- 工具失败是否进入 Trace？
- 文件写入是否返回 hash 和 bytes？

### 11.3 Critic 检查

- 是否检查字段完整性？
- 是否检查工具失败？
- 是否能识别模型编造工具结果？
- 是否能阻止高风险动作？

### 11.4 输出检查

- 最终结果是否满足原始目标？
- 是否区分事实、推测和建议？
- 是否保留可审计路径，例如文件路径、hash、Trace 摘要？

## 12. 常见坑

### 坑 1：把 Agent 做成一个超长 Prompt

问题：所有职责都塞进一个 Prompt，模型既要计划、执行、审查，又要记住工具结果，容易混乱。

更好的做法：拆成 Planner、Executor、Critic 三个角色，每个角色只做一件事。

### 坑 2：工具没有边界

问题：直接暴露 shell、数据库写入或生产 API，模型一旦误判就可能造成真实损害。

更好的做法：从只读工具开始；写操作先生成草稿；高风险动作必须人工确认。

### 坑 3：没有 Trace

问题：最终结果看起来正确，但无法解释中间过程，也无法定位工具失败。

更好的做法：每次工具调用都记录工具名、参数、返回值和错误。

### 坑 4：Critic 只做“润色”

问题：Critic 如果只改文字，不检查目标、字段、工具失败和安全边界，就没有真正质检价值。

更好的做法：给 Critic 明确的检查清单，并允许它判定 fail。

### 坑 5：过早上复杂基础设施

问题：还没验证任务价值，就引入向量库、任务队列、多 Agent 编排、长期记忆，复杂度迅速失控。

更好的做法：先用关键词 KB、本地文件、短期 Trace 验证流程；跑通后再逐层升级。

## 13. 一周落地路线

### 第 1 天：做最小 Agent 骨架

目标：跑通 `plan -> execute -> critique`。

交付物：

- `AgentState`
- Planner Prompt
- Executor Prompt
- Critic Prompt
- 一个固定输入样例

验收：能输出结构化计划和最终结果。

### 第 2 天：接入 2 个低风险工具

建议先接：

- `kb_search`
- `write_file`

验收：Trace 中能看到工具调用记录，文件写入返回 hash。

### 第 3 天：加入失败处理

覆盖：

- Planner 输出非 JSON。
- 工具不存在。
- 工具返回 `ok=false`。
- 文件写入失败。

验收：Agent 不再把失败说成成功。

### 第 4 天：加入 Critic 检查清单

重点检查：

- 目标是否满足。
- 字段是否完整。
- 工具结果是否被正确引用。
- 是否有未经证据支持的断言。

验收：故意给一个缺字段草稿，Critic 能判定 fail。

### 第 5 天：增加两个真实业务样例

建议：

- 会议纪要。
- 运维故障只读分析。
- 知识库安全问答。

验收：每个样例都有输入、期望输出、失败处理。

### 第 6 天：补审计与安全边界

必须补：

- 工具白名单。
- 高风险动作拒绝或人工确认。
- Trace 截断策略，避免上下文过长。
- API key 不进日志。

### 第 7 天：写 README 和验收清单

README 至少包含：

- 架构图。
- 如何运行。
- 三个示例。
- 工具列表。
- 风险边界。
- 已知限制。

## 14. 什么时候升级为生产系统

满足下面条件前，不建议进入生产写操作：

- 已有至少 20 条真实或仿真样例。
- 每类任务都有期望输出和失败样例。
- Critic 能拦截明显缺字段、工具失败、高风险动作。
- 所有工具有权限边界。
- 写操作支持人工确认或回滚。
- 日志不包含 API key、用户隐私、生产凭据。
- 有人工可读的 Trace 摘要。

## 15. 最小可复制检查清单

```text
[ ] Planner 输出 JSON，可解析
[ ] Plan 包含 steps、needed_tools、success_criteria
[ ] Executor 只调用白名单工具
[ ] 每次工具调用进入 Trace
[ ] 工具失败不会被说成成功
[ ] 文件写入返回 path、sha16、bytes
[ ] Critic 检查目标、字段、工具失败和安全边界
[ ] 高风险动作需要人工确认
[ ] 最终输出区分事实、推测、建议
[ ] README 写明边界和已知限制
```

## 16. 推荐的 KISS 版本

如果只想先做一个够用版本，不要一开始就做多 Agent 平台。先做这个：

```text
一个 Python 脚本
+ 三个 Prompt
+ 两个工具：kb_search / write_file
+ 一个 AgentState
+ 三个样例
+ 一个 Critic 检查清单
```

这已经能覆盖多数“半自动工作流助手”的第一阶段需求。等它在真实任务里稳定，再考虑引入向量检索、任务队列、并发工具、长期记忆和自动评估。

## 附：文章来源与限制

本文教程的核心来自 MarkTechPost 文章中的教学型 Agent 实现：Planner、Executor、Critic、`AgentState`、工具 Schema、Trace、会议纪要 Demo。实践扩展部分加入了更保守的安全边界、只读运维案例、知识库问答案例、一周落地路线和上线检查清单。

原文没有完整覆盖生产级权限隔离、自动回滚、长期记忆压缩、上下文治理、评估集建设和工具沙箱；这些不应被视为原文已解决的问题。