前言

在构建 CodeFlow AI 的过程中，我意识到：AI 的最大风险不在于“不会写”，而在于“乱写”。一个能够直接修改磁盘文件的 Agent 就像一个没带保险栓的武器。

本文将详细介绍我为 CodeFlow AI 设计的全局智能体公共安全机制——Change Gate，并展示如何基于此机制实现高效、可靠的 Doc & API Doc Agent。

---

一、 核心架构：全局 Change Gate 机制

1. 为什么需要全局 Gate？

如果每个 Agent 都自己实现一套“写入确认”逻辑，代码会变得臃肿且难以维护。通过将 Change Gate 抽离为公共组件，我们实现了：

• 统一安全策略：所有写操作强制进入“预览态”。

• 统一 UI 交互：侧边栏只需一个组件即可展示所有 Agent 提交的变更。

• 追溯性：系统自动记录谁在什么时候、为了什么目的提出了修改。

2. 技术实现：双阶段提交模型

我们设计了一个 ChangeManager 类，作为整个系统变更的“中转站”。

1. 为什么要设计 Change Gate？

AI 智能体在处理代码时是“概率性”的。如果让 Agent 直接 open(file, 'w').write()，一旦它产生幻觉（例如删掉了关键逻辑或写错了语法），你的项目会立刻崩溃。

Change Gate 的核心理念：

• 预览权：Agent 只能提出修改建议（生成 Diff），不能直接写磁盘。

• 确认权：修改是否生效，必须由人类点击“Apply”。

• 审计性：每一笔修改都有 ID 可查，有日志可追溯。

2. 实现原理：双阶段提交（Two-Phase Commit）

1. Phase 1 (Proposal)：Agent 调用 modify_code_preview 工具。系统计算当前代码与新代码的差异（Diff），生成一个唯一的 change_id，并存入缓存（Session State 或 Redis）。

2. Phase 2 (Confirmation)：前端 UI 展示 Diff。用户点击“应用”时，后台根据 change_id 找到对应的修改内容，执行真正的磁盘写入。

3. 核心代码实现逻辑

在你的 codeflow/langchain_agent.py 中，我们可以抽象出一个变更管理器：

后端实现（Python）

import difflib
import uuid
import os
from typing import Dict

class ChangeManager:
    """全局变更管理器：负责拦截、存储、展示和应用 Agent 的修改建议"""
    
    def __init__(self):
        # 存储结构: {change_id: {path, original_content, modified_content, agent_name, timestamp}}
        self.pending_changes: Dict[str, dict] = {}

    def propose_change(self, file_path: str, new_content: str, agent_name: str) -> str:
        """Agent 调用此方法提交修改建议"""
        original_content = ""
        if os.path.exists(file_path):
            with open(file_path, "r", encoding="utf-8") as f:
                original_content = f.read()

        change_id = f"CHG-{uuid.uuid4().hex[:6].upper()}"
        
        # 计算 Diff
        diff = difflib.unified_diff(
            original_content.splitlines(keepends=True),
            new_content.splitlines(keepends=True),
            fromfile=f"a/{file_path}",
            tofile=f"b/{file_path}"
        )

        self.pending_changes[change_id] = {
            "path": file_path,
            "original": original_content,
            "modified": new_content,
            "diff": "".join(diff),
            "agent": agent_name
        }
        return change_id

    def apply_change(self, change_id: str):
        """用户点击确认后，执行物理写入"""
        if change_id not in self.pending_changes:
            return False, "Change ID 不存在"
        
        change = self.pending_changes[change_id]
        try:
            # 确保目录存在
            os.makedirs(os.path.dirname(change["path"]), exist_ok=True)
            with open(change["path"], "w", encoding="utf-8") as f:
                f.write(change["modified"])
            
            # 记录审计日志并移除待办
            print(f"[Audit] {change['agent']} applied changes to {change['path']}")
            del self.pending_changes[change_id]
            return True, "成功应用变更"
        except Exception as e:
            return False, str(e)

---

二、 进阶智能体：Doc & API Doc Agent

有了 Change Gate，我们就可以放心地让 Doc Agent 去扫描整个项目并生成文件。

1. 实现难点：上下文爆炸

代码项目可能包含成千上万行代码。如果直接把所有代码丢给 LLM 生成文档，不仅耗时耗钱，还会导致 LLM 遗忘关键细节。

2. 解决方案：AST（抽象语法树）骨架抽取

在将代码交给 Doc Agent 之前，我们先通过 Python 自带的 ast 模块进行“脱水”处理：

• 只保留：类名、方法名、函数参数签名、装饰器、Docstring。

• 剔除：具体的循环逻辑、计算逻辑等。

这样，1000 行的代码会缩减为 50 行的“精华骨架”，Agent 处理起来极快且精准。

3. Agent 工具链实现

我们将 Doc Agent 拆分为两个核心 Tool。

Tool 1: scan_api_structure (API 结构扫描)

@tool
def scan_api_structure(file_path: str):
    """利用 AST 提取代码中的 API 结构，不包含具体逻辑实现"""
    import ast
    with open(file_path, "r") as f:
        tree = ast.parse(f.read())
    
    structure = []
    for node in ast.walk(tree):
        if isinstance(node, ast.FunctionDef):
            # 获取装饰器（如 @app.get）
            decorators = [ast.dump(d) for d in node.decorator_list]
            structure.append({
                "name": node.name,
                "args": [a.arg for a in node.args.args],
                "decorators": decorators
            })
    return structure

Tool 2: generate_doc_and_propose (生成并提交预览)

@tool
def generate_doc_and_propose(file_path: str, analysis_result: str):
    """根据分析结果生成 Markdown，并通过 Change Gate 提交"""
    
    prompt = f"根据以下代码结构分析，编写一份专业的 API 文档：{analysis_result}"
    doc_content = llm.invoke(prompt)
    
    # 定义文档存放路径
    target_doc_path = f"docs/api_{os.path.basename(file_path).replace('.py', '.md')}"
    
    # 重点：此处不直接写文件，而是调用 Change Gate
    change_id = global_change_manager.propose_change(
        file_path=target_doc_path,
        new_content=doc_content,
        agent_name="DocAgent"
    )
    
    return f"文档已生成。变更ID: {change_id}。请在侧边栏预览并确认。"

---

三、 全局联动：用户如何交互？

在 Streamlit 前端，我们只需要在侧边栏维护一个 st.expander，即可显示所有待确认的变更。

# streamlit_app.py
with st.sidebar:
    st.subheader("🛡️ 安全变更网关")
    pending = change_manager.pending_changes
    
    for cid, info in pending.items():
        with st.expander(f"[{info['agent']}] {info['path']}"):
            st.code(info['diff'], language="diff") # 展示差异
            col1, col2 = st.columns(2)
            if col1.button("✅ 确认应用", key=f"app_{cid}"):
                success, msg = change_manager.apply_change(cid)
                st.success(msg)
                st.rerun()
            if col2.button("❌ 拒绝", key=f"rej_{cid}"):
                del change_manager.pending_changes[cid]
                st.rerun()

---

四、 总结与心得

1. 这种架构的优势

• 解耦：Agent 只管“提建议”，Change Manager 只管“审建议”，UI 只管“展建议”。

• 容错：由于有了预览步骤，Doc Agent 生成的 Markdown 如果格式有误，用户可以立即发现并拒绝。

• 工程化：这套机制同样适用于我之前的 Code Review 和 Test Generation 智能体，实现了真正的“全家桶”安全管理。

2. 学习建议

如果你也想实现类似的系统，建议先打好基础工具类。不要急着写 Prompt，先写好 ChangeManager 和 ASTExtractor。当你的工具足够稳健时，上层的智能体会变得非常容易调试。

---

下一篇预告：我们将深入探讨 Bug Diagnosis Agent。看 AI 如何通过分析堆栈日志，逆向检索 RAG 知识库，最终锁定代码中那行该死的 Bug！

---

给读者的练习建议：

1. 尝试：把你的 Code Review Agent 改造成“建议修复版”，并接入 Change Gate。

2. 思考：如果是大规模重构，一个 Change ID 包含多个文件修改，该如何扩展 ChangeManager？

---

学习要点总结：

• Change Gate：本质是双阶段提交与人为干预（HITL）。

• AST 抽取：解决长文本上下文限制的有效工程手段。

• Diff 算法：利用 difflib 生成标准补丁格式，极大提升可读性。

• 全局单例：在应用中，ChangeManager 通常作为单例运行，保证状态同步。