企业网站和信息化建设,在上海总部互联网公司排名,秦皇岛市海港区邮编,网站是生成静态好还是动态好本文针对长周期Agent落地中的上下文丢失和任务完成误判问题#xff0c;提出三大解决方案#xff1a;双Agent架构实现任务拆解与增量开发#xff1b;基于Milvus向量数据库的语义检索机制保障跨会话状态恢复#xff1b;测试驱动的端到端验证确保功能完整可用。通过LangGraph和…本文针对长周期Agent落地中的上下文丢失和任务完成误判问题提出三大解决方案双Agent架构实现任务拆解与增量开发基于Milvus向量数据库的语义检索机制保障跨会话状态恢复测试驱动的端到端验证确保功能完整可用。通过LangGraph和Milvus协同工作构建短期记忆长期记忆系统使Agent能可靠完成复杂长周期任务。文尹珉做agent简单但是做能落地的agent难做能落地的长周期agent更是难上加难这是不是你搞agent开发时的常态长周期 Agent落地失效通常来说会分两类典型模式第一种发生在任务初期收到“搭建类 claude.ai 的 Web 应用”这类高阶指令后采用贪心策略一次性推进全量开发执行中逐步丢失上下文最终在功能开发半途耗尽上下文窗口后续新实例面对无进度文档的半成品代码库需要耗费大量精力梳理复盘才能恢复应用基础功能。第二种发生在项目中后期已有部分功能落地后新 Agent 因缺乏全局目标认知仅看到局部功能可用便过早判定任务完成忽略大量未实现的功能模块。表面上看这两种失效模式发生在项目的不同阶段、表现形式也截然不同但深挖本质就会发现其核心原因完全一致Agent无法准确确认任务切分的粒度进而导致多个会话间的上下文传递出现断裂或偏差。既然找到了问题的根源那么该如何针对性解决这一核心痛点本文将从架构设计、状态恢复、功能验证三个核心维度详细介绍如何通过双Agent架构、基于文件与向量数据库的状态恢复机制以及测试驱动的功能验证系统性破解长周期Agent的落地难题。01为什么需要两个 Agent要解决任务切分粒度不准和上下文传递失效的问题首先需要从架构层面搭建清晰的分工机制。不久前Anthropic 发布了一套双 Agent 架构方案这套方案由两个agent组成Initializer Agent Coding Agent。其中Initializer Agent的核心职责是“当领导”负责统筹全局具体需要完成三件核心工作第一是任务拆解即把用户的高级需求精准展开为具体的、可验证的功能点。以claude.ai克隆项目为例Initializer Agent不会停留在构建一个对话应用这种模糊表述上而是会将其细化为200多项具体功能每一项都清晰描述了用户可执行的具体操作和对应的预期结果让后续开发有明确的执行标准第二是进度跟踪通过创建进度追踪文件实时记录每个功能的完成状态让开发进度可视化避免出现局部完成即判定整体结束的误判第三是环境搭建编写初始化脚本init.sh提前配置好开发所需的基础环境确保后续Coding Agent能快速启动开发无需在环境配置上耗费额外精力。与Initializer Agent的统筹角色相对应Coding Agent的职责是充当“专家”在Initializer Agent的指挥下聚焦具体功能的落地执行。其核心工作逻辑是增量迭代每轮会话只选择一个功能进行实现完成后先进行充分测试再用描述性的提交信息将代码提交到Git同步更新进度文件最后结束会话从根本上避免了半成品代码堆积和上下文丢失的问题。需要特别注意的是进度追踪列表的格式选择对Agent的执行效率影响极大。这里更推荐采用JSON格式尽量不要用Markdown格式因为模型在编辑JSON时更倾向于精准修改特定字段但处理Markdown时容易产生意外的格式变化或整体重写。这种差异在处理包含数百个功能项的大型列表时会尤为明显可能直接导致进度追踪失效。一个标准的功能项JSON结构如下{ category: functional, description: New chat button creates a fresh conversation, steps: [ Navigate to main interface, Click the New Chat button, Verify a new conversation is created, Check that chat area shows welcome state, Verify conversation appears in sidebar ], passes: false }这种结构化格式让 Agent 能够清晰地理解每个功能的范围和验证标准同时通过 passes 字段追踪完成状态为跨会话的进度衔接提供可靠依据。02从文本匹配到语义检索Agent 记忆的升级在解决了任务切分和增量开发的问题后另一个影响长周期Agent落地的关键痛点在于跨会话的状态恢复与经验复用。起初基于进度文件与Git的状态恢复机制在项目初期能正常运行但随着项目规模扩大三类问题会逐渐凸显一是Git历史提交次数超过数百次后线性扫描的效率会大幅下降无法快速定位关键信息二是基于文本的查找只能匹配关键词无法理解“JWT刷新令牌”与“用户认证”这类语义关联导致历史经验无法有效复用三是各项目的经验被孤立存储无法跨项目迁移复用增加了重复开发成本。向量数据库通过语义检索解决了这些问题。其核心逻辑是将进度记录、Git提交信息等历史数据转换为向量嵌入当Agent需要查询历史信息时只需输入查询语句向量数据库就能在毫秒级响应时间内找到语义相关的内容。在这一场景中Milvus数据库的优势尤为突出一方面Milvus Lite支持轻量化部署可像SQLite一样直接嵌入应用无需复杂的集群配置另一方面我们也可以采用Milvus的docker、k8s版本全都原生支持主流的嵌入模型API且通过TextEmbedding功能Agent无需手动处理向量生成直接用自然语言查询语句就能检索历史记录大幅降低了使用门槛。03代码跑通≠功能完成测试驱动的功能验证闭环即便解决了任务切分和记忆复用问题长周期Agent仍可能因功能验证不充分导致落地失效。因为很多时候Agent在实现功能后仅通过简单的代码测试或curl命令测试就标记功能完成但这类测试往往无法覆盖真实使用场景很容易遗漏实际应用中的问题——这一痛点在Web应用开发中尤为明显。要解决这一问题可以让Agent使用浏览器自动化工具进行端到端测试。对于Web应用而言就是通过Puppeteer等工具模拟真实用户的操作流程打开页面、点击按钮、填写表单、检查页面元素等只有当功能通过这种全流程的真实场景测试后Agent才能将其标记为完成。这种测试方式让 Agent 能够发现仅从代码层面无法察觉的问题。比如一个对话功能代码逻辑可能完全正确API 返回也符合预期但在浏览器中可能因为 CSS 问题导致用户看不到回复内容。通过浏览器自动化测试Agent 可以截图验证页面显示是否符合预期从而发现这类问题。但这种方法也有局限性。某些浏览器原生功能特别是系统级弹窗如确认对话框、文件选择器无法被 Puppeteer 这样的自动化工具访问。依赖这些原生弹窗的功能往往更容易出现测试盲区。在设计应用时需要考虑这一点尽量使用可测试的自定义 UI 组件替代原生弹窗。04动手实操LangGraph 管状态Milvus 管记忆为了让上述方案总结更容易理解下面通过一个实操示例展示如何通过LangGraph和Milvus的协同实现长周期任务重双Agent的状态管理与记忆复用。该示例的核心逻辑是“短期记忆长期记忆”的协同工作LangGraph的检查点机制负责维护单个会话内的状态一致性确保会话内的开发流程不中断当需要跨会话查询历史信息时由Milvus提供语义检索支持帮助Agent快速恢复上下文彻底避免从零开始的冷启动问题。from sentence_transformers import SentenceTransformer from pymilvus import MilvusClient from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph, START, END from typing import TypedDict, Annotated import operator import subprocess import json # 初始化 embedding_model SentenceTransformer(all-MiniLM-L6-v2) milvus_client MilvusClient(./milvus_agent_memory.db) # 创建集合 if not milvus_client.has_collection(agent_history): milvus_client.create_collection( collection_nameagent_history, dimension384, auto_idTrue ) # Milvus操作函数 def retrieve_context(query: str, top_k: int 3): 从Milvus检索相关历史核心要素语义检索 query_vec embedding_model.encode(query).tolist() results milvus_client.search( collection_nameagent_history, data[query_vec], limittop_k, output_fields[content] ) if results and results[0]: return [hit[entity][content] for hit in results[0]] return [] def save_progress(content: str): 保存进度到Milvus长期记忆 embedding embedding_model.encode(content).tolist() milvus_client.insert( collection_nameagent_history, data[{vector: embedding, content: content}] ) # 核心要素1Git提交 def git_commit(message: str): Git提交原文核心要素 try: # 实际项目中会执行真实的Git命令 # subprocess.run([git, add, .], checkTrue) # subprocess.run([git, commit, -m, message], checkTrue) print(f[Git提交] {message}) return True except Exception as e: print(f[Git提交失败] {e}) return False # 核心要素2测试验证 def run_tests(feature: str): 执行测试原文强调的端到端测试 try: # 实际项目中会调用Puppeteer等测试工具 # 这里简化为模拟测试 print(f[测试验证] 正在测试功能: {feature}) # 模拟测试结果 test_passed True # 实际会根据测试结果返回 if test_passed: print(f[测试通过] {feature}) return test_passed except Exception as e: print(f[测试失败] {e}) return False # 状态定义 class AgentState(TypedDict): messages: Annotated[list, operator.add] features: list# 所有功能列表 completed_features: list # 已完成功能 current_feature: str # 当前正在处理的功能 session_count: int # 会话计数器 # 双Agent节点 def initialize_node(state: AgentState): 初始化Agent生成功能列表和工作环境 print(\n 初始化Agent启动 ) # 生成功能列表实际会根据需求生成详细的功能清单 features [ 实现用户注册功能, 实现用户登录功能, 实现密码重置功能, 实现用户资料编辑, 实现会话管理功能 ] # 保存初始化信息到Milvus init_summary f项目初始化完成生成{len(features)}个功能点 save_progress(init_summary) print(f[初始化完成] 功能列表: {features}) return { **state, features: features, completed_features: [], current_feature: features[0] if features else , session_count: 0, messages: [init_summary] } def code_node(state: AgentState): 编码Agent实现、测试、提交核心循环节点 print(f\n 编码Agent会话 #{state[session_count] 1} ) current_feature state[current_feature] print(f[当前任务] {current_feature}) # 核心要素3从Milvus检索历史经验跨会话记忆 print(f[检索历史] 查询与{current_feature} 相关的历史经验...) context retrieve_context(current_feature) if context: print(f[检索结果] 找到 {len(context)} 条相关记录:) for i, ctx in enumerate(context, 1): print(f {i}. {ctx[:60]}...) else: print([检索结果] 无相关历史记录首次实现此类功能) # 步骤1实现功能 print(f[开始实现] {current_feature}) # 实际会调用LLM生成代码 implementation_result f已实现功能: {current_feature} # 步骤2测试验证核心要素 test_passed run_tests(current_feature) if not test_passed: print(f[会话结束] 测试未通过需要修复) return state# 测试失败则不推进 # 步骤3Git提交核心要素 commit_message ffeat: {current_feature} git_commit(commit_message) # 步骤4更新进度文件 print(f[更新进度] 标记功能为已完成) # 步骤5保存到Milvus长期记忆 progress_record f完成功能: {current_feature} | 提交信息: {commit_message} | 测试状态: 通过 save_progress(progress_record) # 步骤6更新状态并准备下一个功能 new_completed state[completed_features] [current_feature] remaining_features [f for f in state[features] if f not in new_completed] print(f[进度统计] 已完成: {len(new_completed)}/{len(state[features])}) # 核心要素4会话结束明确的会话边界 print(f[会话结束] 代码库处于干净状态可安全中断\n) return { **state, completed_features: new_completed, current_feature: remaining_features[0] if remaining_features else , session_count: state[session_count] 1, messages: [implementation_result] } # 核心要素3循环控制 def should_continue(state: AgentState): 判断是否继续下一个功能实现增量循环开发 if state[current_feature] and state[current_feature] ! : return code# 继续执行下一个功能 else: print(\n 所有功能已完成 ) return END # 构建工作流 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(initialize, initialize_node) workflow.add_node(code, code_node) # 添加边 workflow.add_edge(START, initialize) workflow.add_edge(initialize, code) # 添加条件边实现循环 workflow.add_conditional_edges( code, should_continue, { code: code,# 继续循环 END: END# 结束 } ) # 编译工作流使用MemorySaver作为检查点器 app workflow.compile(checkpointerMemorySaver()) # 使用示例演示跨会话恢复 if __name__ __main__: print( * 60) print(演示场景长周期Agent的多会话开发) print( * 60) # 会话1初始化 完成前2个功能 print(\n【场景1】第一次启动完成前2个功能) config {configurable: {thread_id: project_001}} result app.invoke({ messages: [], features: [], completed_features: [], current_feature: , session_count: 0 }, config) print(\n * 60) print(【模拟场景】开发者手动中断CtrlC或上下文窗口耗尽) print( * 60) # 会话2从checkpoint恢复状态 print(\n【场景2】新会话启动从上次中断处继续) print(通过相同的thread_idLangGraph自动恢复checkpoint...) # 使用相同的thread_idLangGraph会自动从checkpoint恢复状态 result app.invoke({ messages: [], features: [], completed_features: [], current_feature: , session_count: 0 }, config) print(\n * 60) print(演示完成) print( * 60) print(\n关键要点:) print(1. ✅ 双Agent架构initialize code) print(2. ✅ 增量循环开发条件边控制循环) print(3. ✅ Git提交每个功能完成后提交) print(4. ✅ 测试验证端到端测试) print(5. ✅ 会话管理明确的会话边界) print(6. ✅ 跨会话恢复thread_id checkpoint) print(7. ✅ 语义检索Milvus长期记忆)尾声此前市面上的agent方案中多采用单一Coding Agent处理所有编码相关任务但在真实的软件开发流程中不同阶段的需求差异极大。因此需要多 Agent 专业分工测试 Agent 关注边界条件质量保证 Agent 负责架构检查清理 Agent 处理重构和文档。每个 Agent 针对专门领域优化提示和工具集通过结构化接口协作就像真实的开发团队。值得注意的是这套方案的核心原则结构化初始环境、增量式进展、状态追踪、验证测试并非仅适用于软件开发场景还可迁移到其他长周期任务中例如科学研究中的数周连续实验、金融建模中的多步骤复杂计算、法律文档的多轮迭代审查等。随着向量数据库技术的成熟和多Agent框架的完善让AI可靠完成复杂长周期任务已经从理论探索逐步进入工程实践阶段未来将释放更多生产力价值。如何学习AI大模型 “最先掌握AI的人将会比较晚掌握AI的人有竞争优势”。这句话放在计算机、互联网、移动互联网的开局时期都是一样的道理。我在一线互联网企业工作十余年里指导过不少同行后辈。帮助很多人得到了学习和成长。我意识到有很多经验和知识值得分享给大家故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。【保证100%免费】CSDN粉丝独家福利这份完整版的 AI 大模型学习资料已经上传CSDN朋友们如果需要可以扫描下方二维码点击下方CSDN官方认证链接免费领取【保证100%免费】读者福利CSDN大礼包《最新AI大模型学习资源包》免费分享 安全链接放心点击对于0基础小白入门如果你是零基础小白想快速入门大模型是可以考虑的。一方面是学习时间相对较短学习内容更全面更集中。二方面是可以根据这些资料规划好学习计划和方向。1.大模型入门学习思维导图要学习一门新的技术作为新手一定要先学习成长路线图方向不对努力白费。对于从来没有接触过AI大模型的同学我们帮你准备了详细的学习成长路线图学习规划。可以说是最科学最系统的学习路线大家跟着这个大的方向学习准没问题。全套教程文末领取哈2.AGI大模型配套视频很多朋友都不喜欢晦涩的文字我也为大家准备了视频教程每个章节都是当前板块的精华浓缩。3.大模型实际应用报告合集这套包含640份报告的合集涵盖了AI大模型的理论研究、技术实现、行业应用等多个方面。无论您是科研人员、工程师还是对AI大模型感兴趣的爱好者这套报告合集都将为您提供宝贵的信息和启示。全套教程文末领取哈4.大模型实战项目项目源码光学理论是没用的要学会跟着一起做要动手实操才能将自己的所学运用到实际当中去这时候可以搞点实战项目来学习。全套教程文末领取哈5.大模型经典学习电子书随着人工智能技术的飞速发展AI大模型已经成为了当今科技领域的一大热点。这些大型预训练模型如GPT-3、BERT、XLNet等以其强大的语言理解和生成能力正在改变我们对人工智能的认识。 那以下这些PDF籍就是非常不错的学习资源。全套教程文末领取哈6.大模型面试题答案截至目前大模型已经超过200个在大模型纵横的时代不仅大模型技术越来越卷就连大模型相关的岗位和面试也开始越来越卷了。为了让大家更容易上车大模型算法赛道我总结了大模型常考的面试题。全套教程文末领取哈为什么分享这些资料?只要你是真心想学AI大模型我这份资料就可以无偿分享给你学习我国在这方面的相关人才比较紧缺大模型行业确实也需要更多的有志之士加入进来我也真心希望帮助大家学好这门技术如果日后有什么学习上的问题欢迎找我交流有技术上面的问题我是很愿意去帮助大家的这些资料真的有用吗?这份资料由我和鲁为民博士共同整理鲁为民博士先后获得了北京清华大学学士和美国加州理工学院博士学位在包括IEEE Transactions等学术期刊和诸多国际会议上发表了超过50篇学术论文、取得了多项美国和中国发明专利同时还斩获了吴文俊人工智能科学技术奖。目前我正在和鲁博士共同进行人工智能的研究。资料内容涵盖了从入门到进阶的各类视频教程和实战项目无论你是小白还是有些技术基础的这份资料都绝对能帮助你提升薪资待遇转行大模型岗位。CSDN粉丝独家福利这份完整版的 AI 大模型学习资料已经上传CSDN朋友们如果需要可以扫描下方二维码点击下方CSDN官方认证链接免费领取【保证100%免费】读者福利CSDN大礼包《最新AI大模型学习资源包》免费分享 安全链接放心点击