1. 为什么我们需要“自我反思”型 Agent？

在开发 CodeFlow AI 的过程中，我发现了一个 RAG 项目普遍面临的痛点：大模型生成的代码往往“看起来很对，一跑就废”。

即便是最强大的 Qwen2.5-Coder 或 GPT-4o，在生成单元测试（Unit Test）时，也经常会出现：

• 导入错误：引用了不存在的模块或路径。

• 断言失败：逻辑理解偏差导致测试用例无法通过。

• 语法幻觉：使用了错误的库函数。

如果只是简单地将代码扔给用户，这不叫“人工智能”，这叫“给用户添乱”。为了解决这个问题，我为 CodeFlow AI 引入了第二个核心智能体：Test Generation Agent。它的核心思想是：不只写代码，还要自己跑代码、看报错、改 Bug。

---

2. 核心架构：Self-Reflection（自我反思）模式

不同于传统的线性逻辑（Chain），该 Agent 采用了 LangGraph 构建了一个循环状态机：

1. Designer（设计者）：分析源码，生成初步的 pytest 代码。

2. Executor（执行者）：在真实 Python 环境中运行测试，捕获 stdout 和 stderr。

3. Reflector（反思者）：如果测试失败，将 Traceback 报错信息反馈给设计者。

4. Loop（循环）：设计者根据报错重新修正代码，直到测试通过或达到尝试上限。

---

3. 技术实现深度拆解

3.1 状态定义 (State)

使用 TypedDict 定义 Agent 的记忆，记录源码、生成的测试、报错信息及迭代次数。

from typing import TypedDict

class TestAgentState(TypedDict):
    source_code: str          # 源码
    test_code: str            # 生成的测试代码
    error_message: str        # 报错堆栈
    iteration_count: int      # 当前迭代次数
    is_passed: bool           # 是否通过
    max_iterations: int       # 最大尝试次数

3.2 环境反馈：物理执行层

这是 Agent 触达“物理世界”的桥梁。我们使用 subprocess 真实触发 pytest 命令。

import subprocess

def execute_pytest(test_code: str):
    # 将代码写入临时文件并运行
    with open("temp_test.py", "w") as f:
        f.write(test_code)
    
    result = subprocess.run(
        ["pytest", "temp_test.py"], 
        capture_output=True, text=True
    )
    
    if result.returncode == 0:
        return "SUCCESS"
    return f"STDOUT: {result.stdout}\nSTDERR: {result.stderr}"

3.3 逻辑流转：LangGraph 编排

利用 LangGraph 的 add_conditional_edges 实现“写-跑-改”的逻辑闭环。

from langgraph.graph import StateGraph, END

workflow = StateGraph(TestAgentState)

# 定义节点：生成与验证
workflow.add_node("designer", designer_node)
workflow.add_node("executor", executor_node)

# 定义连线
workflow.set_entry_point("designer")
workflow.add_edge("designer", "executor")

# 核心逻辑：失败了就回去重写，成功了就结束
workflow.add_conditional_edges(
    "executor", 
    lambda x: END if x["is_passed"] or x["iteration_count"] >= x["max_iterations"] else "designer"
)

test_gen_agent = workflow.compile()

---

4. 实战案例演示

输入（源码）：
一个简单的除法函数，但模型第一次生成测试时忘记处理 ZeroDivisionError。

第一轮循环：

• Agent 输出：生成了基础测试。

• 执行结果：FAILED (division by zero)。

• 反思反馈：反馈 Traceback 给模型。

第二轮循环：

• Agent 反思：“我发现之前的测试没有覆盖异常处理，我现在加上 pytest.raises。”

• 执行结果：SUCCESS。

最终结果：
用户拿到的不是一份“可能有用”的代码，而是一份已经在后台验证通过、逻辑完备的生产级代码。

详细演示版本是

为了让你深度理解这个 Agent 是如何“思考”和“修正”代码的，我们来模拟一个真实的研发场景。

这个案例将展示：Agent 如何从“写错代码”到“看到报错”，最后“自愈成功”的全过程。

---

场景设定：为“折扣计算函数”编写单测

1. 待测试的源码 (utils.py)

这段代码逻辑很简单，但有一个潜在风险：它没有处理折扣大于 1 的情况。

# utils.py
def calculate_price(price: float, discount: float) -> float:
    """计算折扣后的价格"""
    if price < 0:
        raise ValueError("价格不能为负数")
    return price * (1 - discount)

---

第一轮迭代：Agent 的“盲目”尝试

Step 1: Designer 节点生成代码

Agent 此时的 Prompt 输入： “请为 calculate_price 函数生成 pytest 测试用例。”

Agent 输出的测试代码 (temp_test.py)：

注：此时 Agent 犯了一个低级错误：它忘记导入 calculate_price 函数了。

import pytest
# 错误点：忘记从 utils 导入 calculate_price

def test_calculate_price_success():
    # 逻辑错误：它以为 discount 传的是百分比(20)，但函数定义要求的是小数(0.2)
    assert calculate_price(100, 20) == 80

Step 2: Executor 节点运行并报错

系统在后台运行 pytest temp_test.py。

终端返回的报错信息（反馈给 Agent）：

________________ ERROR collecting temp_test.py ________________
temp_test.py:6: in <module>
    def test_calculate_price_success():
E   NameError: name 'calculate_price' is not defined

---

第二轮迭代：Agent 的“自我反思”与修正

Step 3: Reflector 逻辑启动

此时系统给 Agent 的 Prompt：

“你生成的代码运行失败了。
报错信息： NameError: name 'calculate_price' is not defined
源码参考： (给出 utils.py 的内容)
请分析原因并修正。”

Agent 内部的“反思”过程（在面试中可以重点讲）：

1. 识别问题：看到 NameError，意识到是 import 语句缺失。

2. 逻辑对齐：通过阅读源码发现，1 - discount 意味着 discount 应该是 0.2 而不是 20。

3. 完善覆盖：顺便发现源码里有 ValueError 的逻辑，应该增加一个异常测试。

Step 4: Designer 节点输出修正后的代码

import pytest
from utils import calculate_price  # 修复了导入问题

def test_calculate_price_success():
    # 修复了数值逻辑错误：20% 折扣应传 0.2
    assert calculate_price(100, 0.2) == 80.0

def test_calculate_price_negative_value():
    # 额外增加了对异常路径的覆盖
    with pytest.raises(ValueError, match="价格不能为负数"):
        calculate_price(-10, 0.1)

---

第三轮迭代：验证通过

Step 5: Executor 节点再次运行

系统再次运行 pytest temp_test.py。

终端返回信息：

======================= test session starts =======================
collected 2 items

temp_test.py ..                                              [100%]

======================== 2 passed in 0.01s ========================

最终结果： Agent 检测到 is_passed = True，停止循环，将这份已经跑通的代码交付给用户。

---

面试中你可以如何利用这个案例？

如果你被面试官问到：“你的 Agent 怎么保证生成的代码能用？”

你可以这样回答（结合上面的案例）：

“我想举一个具体的例子。当我输入一个折扣计算函数时，Agent 第一次生成的代码往往会忽略一些细节，比如忘记写 import 语句，或者对参数含义理解有误（比如把折扣比例当成了百分比整数）。

传统的 LLM 到这一步就结束了，但我的 Agent 会把代码放到物理环境中去跑。当 pytest 返回一个 NameError 或者 AssertionError 时，我的 Self-Reflection 循环会捕获这个真实的 Traceback。

就像一个真实的程序员一样，Agent 会看到：‘哦，原来是我没导包’，或者‘原来断言的值对不上’。它会基于这个反馈重新改写代码。

在我的实验中，这种‘执行-报错-重写’的机制，让代码的一次性通过率（Pass@1）大幅度提升，真正实现了代码生成的自愈闭环。”

---

5. 项目总结与思考

在 CodeFlow AI 的 V2 版本中，通过引入 Self-Reflection 架构，我们将测试生成的可用率从原本的 60% 提升到了 90% 以上。

面试技术点总结：

• 从 Chain 到 Graph：解决了复杂业务中的循环逻辑问题。

• 环境驱动的 Agent：LLM 不再是闭门造车，而是通过外部工具反馈进行进化。

• 工程化安全：在实际生产中，这类执行环境建议放在 Docker 容器中实现沙箱隔离。

---

6. 后续计划

目前 CodeFlow AI 已经完成了 Code Review Agent 和 Test Gen Agent。下一步，我将引入 Ruflo 作为全局编排层，实现从“代码变更触发”到“自动审计+测试”的全自动化流水线。

如果你对本项目感兴趣，欢迎在 GitHub 上关注我的项目：CodeFlow AI。