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

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

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

openai>=1.0.0
python-dotenv>=1.0.0

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

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

# 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,
        })

# 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,
}

# 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"],
            },
        },
    },
]

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

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

用户目标：
{{GOAL}}

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

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

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

计划：
{{PLAN_JSON}}

用户目标：
{{GOAL}}

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

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

用户目标：
{{GOAL}}

执行草稿：
{{DRAFT}}

工具 Trace：
{{TRACE}}

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

# 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,
    }

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.

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

{
  "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": "..."
  }
}

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

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

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

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

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

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

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

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

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

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