8.3. 构建自我进化的编码智能体

构建自我进化的编码智能体

时间线:在上一章里,我们用 LangGraph 实现了带记忆断点恢复的工作流。现在我们把时间拨快两周——你正在为一个中型 Python 微服务项目开发新功能,却发现智能体不断重复同一种 lint 错误,也完全忘记你上周刚修过的数据库连接池配置。你渴望一位能够像人类工程师一样“记住教训”的 AI 队友。
本章,我们就把记忆能力再推进一步:让智能体在编写代码时,自动借鉴该项目的过往经验,并从每一次交互中持续进化。我们将围绕 LangGraph 的状态图作为骨架,使用 Letta 作为长期记忆中间层,构建一个真正“越写越聪明”的编码智能体。


你需要什么

资源 说明 预计获取时间
Python 3.10+ 环境 推荐使用虚拟环境 已具备
LangGraph (>=0.2.0) 包含检查点与内存功能 1 分钟
Letta Python SDK (letta) 提供持久记忆块与语义搜索 1 分钟
OpenAI API Key 或兼容的 LLM 服务 用于智能体推理 已具备
一个示例 Git 仓库 我们将模拟一个 Python 项目 5 分钟

总预计时间:约 30 分钟。


最终成果

完成本章后,你将获得一个自进化的编码智能体,它具备以下三种长期记忆能力:

  1. 项目上下文记忆块:自动从代码库和 CI 日志中提取摘要,并持久化到 Letta 记忆块中,供每次编程任务调用。
  2. 历史修复记录的建议:当用户遇到与过去类似的报错时,智能体会检索历史修复记录,优先尝试之前成功的方案。
  3. 定期学习循环:每周运行一个 review agent,复盘近期的交互与代码变更,更新长期编程策略和风格偏好。

你不仅会自己实现这套机制,也会理解为什么“有状态的记忆”是工程化 AI Agent 的核心差异化能力。


步骤一:项目上下文记忆块的构建

目标:从代码库结构与 CI 日志中提取关键信息,生成一份“项目画像摘要”,并持久化到 Letta 的记忆块中。

1.1 初始化 Letta 客户端并创建记忆块

Letta(原名 MemGPT)是一个有状态代理平台,提供 Python SDK(即 letta 包)以便开发者集成。我们首先要创建一个专属于该项目的记忆块(block),用于存放长期上下文。

# 安装: pip install letta
from letta import LettaClient

# 连接到 Letta 服务(本地或远程)
client = LettaClient(base_url="http://localhost:8283")  # 如果是本地部署

# 创建一个新的项目记忆块
project_id = "my-python-service"
block = client.create_block(
    label=f"project_context_{project_id}",
    description="持久化存储项目的代码结构与 CI 摘要",
)
print(f"记忆块 ID: {block.id}")

⚠️ 踩坑经验

如果 Letta 启动时报错 port already in use,请检查是否已有残留的 MemGPT 或 Letta 进程。在迁移到 Letta 后,旧版 MemGPT 的端口可能与新版冲突,建议彻底关闭旧容器或在 Docker Compose 中指定新端口。

1.2 提取代码库摘要

我们无需将整个代码仓库塞进上下文窗口(那会迅速超标),而是只提取高层结构:目录树、关键文件列表、主要依赖。

import os
import json
from pathlib import Path

def extract_repo_summary(root_path: str) -> str:
    """生成代码库的结构摘要"""
    summary_lines = []
    # 只遍历最多两层目录,避免信息过载
    for root, dirs, files in os.walk(root_path):
        depth = root.replace(root_path, "").count(os.sep)
        if depth > 2:
            continue
        # 忽略隐藏目录与虚拟环境
        dirs[:] = [d for d in dirs if not d.startswith('.') and d != 'venv']
        rel_root = os.path.relpath(root, root_path)
        if rel_root == ".":
            rel_root = ""
        summary_lines.append(f"{'  ' * depth}📁 {rel_root or '/'}: {len(files)} files")
    # 附加依赖文件内容摘要
    deps_path = os.path.join(root_path, "requirements.txt")
    if os.path.exists(deps_path):
        with open(deps_path) as f:
            deps = f.read().strip()[:500]    # 截断长文件
        summary_lines.append(f"\n📦 依赖:\n{deps}")
    return "\n".join(summary_lines)

repo_summary = extract_repo_summary("./")   # 假设当前目录为项目根

1.3 合并 CI 日志并写入 Letta 块

假设我们可以从 GitHub Actions 或其他 CI 获取最新的构建日志(这里用模拟文本代替)。

ci_log_snippet = """
[CI] flake8: W503 line break before binary operator (utils.py:42)
[CI] pytest: 5 passed, 1 error in test_db.py
[CI] docker build: SUCCESS
"""

full_context = f"项目代码结构:\n{repo_summary}\n\n最近一次 CI 日志:\n{ci_log_snippet}"

# 将摘要写入记忆块
client.update_block(
    block_id=block.id,
    value=full_context,
)
print("项目上下文已持久化至 Letta 记忆块。")

预期结果:在 Letta 界面或通过 API 查看该记忆块时,可以看到结构清晰的代码库摘要与 CI 日志片段,后续任何编程智能体都会先读取此块作为基础上下文。


步骤二:基于历史修复记录的建议

目标:当用户报告类似错误时,智能体能够检索过去解决过的问题,并优先采用已经验证的修复策略。

2.1 存储历史修复记录

每当我们修复一个 bug 或完成一次代码迭代,就把“问题描述-解决方案”对存储到 Letta 的另一个记忆块中。这里我们设计一个 save_fix 工具函数。

def save_fix(problem_desc: str, solution: str, embedding_label: str):
    """
    将一次修复记录存入 Letta 的记忆块,并附一个可语义检索的标签。
    """
    fix_block_id = client.get_block_id_by_label("project_fixes")  # 预先创建的全局修复块
    if not fix_block_id:
        raise RuntimeError("请先创建名为 'project_fixes' 的块")

    # 将条目追加为记忆块的值(实际生产环境可切成多个块,此处简化为文本追加)
    current_val = client.get_block(fix_block_id).value or ""
    new_entry = f"[问题]: {problem_desc}\n[方案]: {solution}\n---\n"
    client.update_block(fix_block_id, value=current_val + new_entry)

    # 也可以为该条目创建一条独立的嵌入记忆(利用 Letta 的语义搜索)
    client.create_passage(
        block_id=fix_block_id,
        text=f"修复: {problem_desc} -> {solution[:200]}",
        embedding_label=embedding_label,
    )
    print(f"修复记录已保存: {embedding_label}")

💡 为什么用 embedding_label
Letta 支持基于向量的语义搜索,当智能体遇到新问题时,我们可以用错误信息的描述向量去查找最相似的历史修复条目,而不需要依赖关键词匹配。

2.2 在智能体中实现“记忆式排错”

我们将在 LangGraph 的智能体节点中,插入一个检索历史修复记录的逻辑。如果找到相似问题,直接建议过去成功的方案;否则才进入常规的排错流程。

from langgraph.graph import StateGraph, END

class CodingState(TypedDict):
    user_input: str
    error_log: str
    suggested_fix: str
    # ... 其他状态

def retrieve_past_fix(state: CodingState):
    """检索历史修复记录"""
    query = state["error_log"] or state["user_input"]  # 用错误日志作为检索条件
    # 使用 Letta 的语义搜索方法(假设 client 已扩展)
    results = client.search_passages(
        block_label="project_fixes",
        query=query,
        limit=3,
    )
    if results and results[0].score > 0.8:   # 阈值可调
        best_fix = results[0].text
        return {"suggested_fix": f"根据历史记录,建议尝试: {best_fix}"}
    else:
        return {"suggested_fix": None}

def fix_error(state: CodingState):
    """常规修复逻辑,或结合历史建议"""
    if state["suggested_fix"]:
        # 优先采纳记忆中的方案
        final_fix = apply_fix_strategy(state["suggested_fix"], state["error_log"])
    else:
        final_fix = run_new_troubleshooting(state["error_log"])
    return {"fix_output": final_fix}

# 构建图(省略完整流程,仅展示关键节点)
builder = StateGraph(CodingState)
builder.add_node("retrieve", retrieve_past_fix)
builder.add_node("fix", fix_error)
builder.set_entry_point("retrieve")
builder.add_edge("retrieve", "fix")
# ... 

预期结果:当你第二次遇到 flake8 W503 警告时,智能体会直接输出:“根据历史记录,建议在 utils.py 第 42 行调整二进制运算符换行位置。” 而不是重新开始分析。


步骤三:学习循环:定期复盘项目交互

目标:让智能体每周自动运行一次 review agent,总结这一周的编程交互历史,提炼出新的长期策略,并更新自己的“编程规范”记忆。

3.1 设计复盘 Prompt

我们编写一个 review agent,它读取近期所有对话摘要(由每次任务的 checkpoint 生成,或直接存储在 Letta 的对话记忆块中),然后生成三个输出:

  • 新的编码约定:例如“在所有的异步函数中统一使用 async with session
  • 常见错误模式:如“频繁出现 KeyError: 'DB_HOST'
  • 性能优化建议:如“数据库连接池大小调整为 20 以避免等待”
review_prompt_template = """
你是一个资深后端工程师,请审查以下本周的编程交互历史,并给出:

1. 需要固化的新编码约定(若有)
2. 最常见的一类错误及预防措施
3. 任何可执行的最佳实践优化建议

交互历史:
{interaction_log}

输出格式: JSON
{
  "new_conventions": "...",
  "common_error": "...",
  "optimizations": "..."
}
"""

3.2 执行审核并更新记忆

每周运行一次(可通过 cron job 或 CI 定时任务),执行如下代码:

import json

def weekly_review():
    # 从 Letta 获取本周的对话摘要
    log_block = client.get_block_by_label("weekly_interaction_log")
    interaction_text = client.get_block(log_block).value or ""

    # 调用 LLM 生成复盘内容
    review_response = call_llm(review_prompt_template.format(interaction_log=interaction_text))
    review = json.loads(review_response)

    # 将提炼后的准则更新到“编程策略”记忆块
    strategy_block_id = client.get_block_id_by_label("coding_strategy")
    current_strategy = client.get_block(strategy_block_id).value or ""
    updated_strategy = (
        f"{current_strategy}\n\n[Week Review Update]\n"
        f"约定: {review['new_conventions']}\n"
        f"错误模式防范: {review['common_error']}\n"
        f"优化: {review['optimizations']}\n"
    )
    client.update_block(strategy_block_id, value=updated_strategy)
    print("每周复盘已完成,长期策略已更新。")

    # 最后,清空本周日志,为下周做准备
    client.update_block(log_block.id, value="")

预期结果:几周后,查看 coding_strategy 记忆块,你会看到一份不断增长的“团队编码宪法”,智能体在每次生成代码前都会参考这些准则,从而使输出风格与项目历史保持高度一致。


回顾

  • 做了什么:我们搭建了一个自我进化的编码智能体,它利用 Letta 持久化记忆块,能够记住项目上下文、从历史修复中学习,并定期复盘迭代。
  • 核心组件:Letta 记忆块(项目上下文、修复记录、策略)、LangGraph 状态图(检索节点、修复节点)、定时 review agent。
  • 花了多久:按照本教程操作,大约 30 分钟可完成基础版本搭建并亲眼看到记忆的“生长”。

行动清单

  1. 在本地启动 Letta 服务,创建三个记忆块:project_context_xxxproject_fixescoding_strategy
  2. 运行代码库摘要提取脚本,将内容写入第一个记忆块。
  3. 编写一个简单的 LangGraph 图,包含检索历史修复记录的节点,并接入你的 IDE 或命令行编程助手。
  4. 手动触发几次修复并调用 save_fix 存储记录,立即测试检索效果。
  5. 实现 weekly_review 函数,并配置定时任务,观察几周后的策略记忆变化。

下一章《多智能体协作中的上下文共享与隔离实战》,我们将进一步面对多个智能体共同工作时的记忆治理难题:如何让研究员、分析员、工程师共享关键发现,同时确保各自的机密上下文不被泄露。你刚才为单个编码智能体设计的记忆块与语义检索机制,恰好就是多智能体协调的中间层雏形——在那里,我们将用相同的思想管理更大的群体。

本文章首发在 LearnKu.com 网站上。

上一篇 下一篇
讨论数量: 0
发起讨论 只看当前版本


暂无话题~