模块十二：Capstone项目-端到端企业级Agent系统设计与交付

模块十二：Capstone项目——端到端企业级Agent系统设计与交付前言：从“学会了”到“做到了”——跨越理论与实践的最后一公里在前十一个模块中，我们已经完成了一条完整的AgenticAI能力链——从认知重构到Prompt架构，从LangGraph工作流到LlamaIndex知识管理，从多Agent协作到记忆系统，从工具安全到评估可观测性，再到商业变现。你掌握了构建生产级Agent系统所需的全部技术能力。但有一个残酷的事实：掌握每个模块的知识，不等于能独立交付一个完整的Agent系统。正如KDG在2026年5月提出的Agentic交付模型所揭示的——传统SDLC和脆弱的“vibecoding”都无法满足企业级Agent系统的交付要求，需要一种全新的、AI辅助但保持严格人工监督的交付方法论。这就是Capstone项目的核心价值。它不是“再学一个模块”，而是将前十一个模块的知识在真实项目中融会贯通。你将从需求分析开始，经历架构设计、分阶段实现、安全加固、测试评估，最终交付一个完整的企业级Agent系统——以及一份能让客户买单的商业提案。2026年，企业正在从“评估AIAgent”转向“承诺多Agent、跨职能部署”。Gartner预测到2026年底40%的企业应用将集成任务型AIAgent。51%的企业已有Agent在生产环境中运行。这不是“未来趋势”——这是“正在发生的现实”。而你，通过完成这个Capstone项目，将成为能够抓住这轮机遇的少数实战派。本模块的核心使命：带你完成从“知识掌握者”到“系统交付者”的关键跃迁——综合运用全课程所学知识，独立完成一个企业级Agent系统的端到端设计与交付，产出可直接用于面试/接单/融资的核心作品。学习目标完成本模块学习后，你将能够：独立设计完整的企业级Agent系统架构，能根据业务需求选择最优的技术栈组合（LangGraph+LlamaIndex+AutoGen/CrewAI+Mem0+MCP/A2A）分阶段实现从MVP到生产级的完整系统，掌握“原型验证→核心功能→生产加固”的渐进式交付方法论综合应用前11个模块的全部知识——认知架构（模块一）指导需求分析、Prompt三角架构（模块二）设计Agent认知、LangGraph（模块三）编排工作流、LlamaIndex（模块四）构建知识检索、AutoGen（模块五）设计多Agent协作、Mem0（模块六）实现持久记忆、MCP（模块七）集成工具、A2A（模块八）实现跨框架互操作、安全护栏（模块九）加固系统、评估体系（模块十）持续监控、ROI计算（模块十一）支撑商业提案产出专业级交付物：系统架构文档、可运行代码、部署方案、安全审计报告、演示视频、商业提案掌握Agentic交付方法论：理解KDG等企业验证的“AI辅助但保持严格人工监督”的新型交付模式，能够在数周内交付原型、数月内交付生产系统第一章：Capstone项目全景——四个可选方向与选型策略1.1四个Capstone方向概览本课程提供四个Capstone方向，覆盖2026年AgenticAI商业化落地最成熟的四个领域。每个方向都有明确的业务场景、技术栈推荐和交付物标准。方向业务场景核心Agent技术栈重点商业化成熟度A：金融智能投研Agent自动抓取财报→提取关键指标→多维分析→生成投研报告数据采集Agent、财务分析Agent、报告生成Agent、审核AgentLlamaIndex（RAG）+LangGraph（工作流）+Mem0（记忆）+A2A（多Agent协作）⭐⭐⭐⭐⭐B：电商全链路运营Agent商品上架→智能定价→客服自动回复→营销内容生成→数据分析商品Agent、定价Agent、客服Agent、营销Agent、分析AgentLangGraph（编排）+AutoGen/CrewAI（多Agent）+LlamaIndex（知识库）+MCP（工具集成）⭐⭐⭐⭐⭐C：企业知识管理Agent文档自动解析→知识图谱构建→智能问答→自动报告生成文档解析Agent、分类Agent、检索Agent、问答Agent、报告AgentLlamaIndex+LlamaParse（文档处理）+向量数据库（检索）+LangGraph（工作流）+Mem0（记忆）⭐⭐⭐⭐D：内容创作Agent工厂热点监控→选题策划→多平台内容生成→发布→数据复盘监控Agent、策划Agent、写作Agent、优化Agent、分析AgentCrewAI（角色协作）+AutoGen（多Agent）+LangChain（工具链）+A2A（跨平台发布）⭐⭐⭐⭐1.2如何选择你的Capstone方向？选型决策需要考虑三个维度：你的行业背景、你的技术偏好、以及你的商业目标。决策指南：你的情况推荐方向理由有金融/数据分析背景A（金融智能投研Agent）行业壁垒高、客单价高、合规要求锻炼安全能力有电商/营销背景B（电商全链路运营Agent）场景标准化程度高、Agent角色清晰、多Agent协作模式典型有企业IT/知识管理背景C（企业知识管理Agent）LlamaIndex生态最成熟、交付周期短、容易看到效果有内容创作/媒体背景D（内容创作Agent工厂）创意场景最灵活、CrewAI角色协作最直观、商业验证快想展示最强综合能力A或B覆盖技术栈最全面、多Agent协作最复杂、商业价值最清晰1.3不管你选哪个方向，这七个核心组件你都会用到虽然四个方向各有侧重，但任何企业级Agent系统都绕不开七个核心组件：编排引擎（LangGraph）：所有方向都需要复杂工作流编排知识检索（LlamaIndex+LlamaParse）：所有方向都需要处理文档和检索知识多Agent协作（AutoGen或CrewAI）：方向A/B/D需要，方向C可选记忆系统（Mem0或Zep）：所有方向都需要跨会话记忆工具集成（MCP协议）：所有方向都需要标准化工具接入安全护栏（模块九）：所有方向都需要，金融方向尤其严格评估体系（模块十）：所有方向都需要持续监控和评估1.42026年Agent系统交付的新范式——KDGAgentic交付模型在开始项目之前，理解2026年的Agent系统交付新范式至关重要。KDG在2026年5月提出了Agentic交付模型（AgenticDeliveryModel），其核心理念是：AI辅助但保持严格人工监督——使用AI加速设计、编码、测试和文档生成，同时由资深工程师保持对整个过程的严格把控。这种模式的核心原则是：原型在数周内交付，生产系统在数月内交付AI负责加速，人类负责把关——AI生成代码和文档，工程师审查和验证从工作流开始，而非从Agent开始——在自动化之前先标准化和清晰定义流程多Agent而非单Agent——AgenticAI的典型形态是多Agent系统这个交付模型将贯穿我们Capstone项目的全过程。第二章：Phase1——需求分析与系统架构设计（第1-2周）2.1需求分析的四个核心问题在企业级Agent系统设计中，需求分析阶段的质量直接决定了整个项目的成败。你需要回答四个核心问题：问题一：这个Agent系统要解决什么业务问题？不是“我要构建一个Agent”，而是“我要解决什么业务痛点”。好的需求描述包含：当前流程（谁在什么时候做什么）、痛点量化（耗时多少、错误率多少、成本多少）、期望状态（自动化后期望的效率提升和成本节省）。问题二：任务的自动化可行性如何？使用模块一的“利润洼地识别公式”来评估：利润洼地得分=人力成本×发生频率×流程标准化程度。三个维度得分都高的任务，是最优先的Agent化目标。问题三：任务的复杂度适合什么样的Agent架构？任务越复杂，越需要LangGraph的图编排能力。有明确步骤和决策分支时选择LangGraph的StateGraph，而高度不确定、需要多视角碰撞时更适合AutoGen的对话驱动。问题四：系统的用户是谁？他们如何与Agent交互？是内部员工使用（如知识管理Agent），还是外部客户使用（如客服Agent）？交互方式是自然语言对话，还是API调用？是否需要多语言支持？2.2技术选型决策框架基于前11个模块的知识，2026年企业级Agent系统的标准技术栈已经形成：┌─────────────────────────────────────────────────────────────┐

│2026企业级Agent标准技术栈│

├─────────────────────────────────────────────────────────────┤

││

│编排层：LangGraph1.0(GA,October2025)│

│├──复杂工作流编排、状态持久化（PostgresSaver）、人机协作│

│└──生产验证：Uber等企业的大规模部署│

││

│知识层：LlamaIndexv0.10.68+LlamaParse│

│├──复杂文档解析（PDF/表格/图表）、混合索引、AgenticRAG│

│└──47KGitHubStars,5.2M月下载量│

││

│协作层：AutoGen0.4/CrewAI│

│├──对话驱动的多Agent协作（AutoGen）+角色驱动的任务编排（CrewAI）│

│└──生产混合策略：CrewAI管理工作流，AutoGen处理复杂决策│

││

│记忆层：Mem0/Zep│

│├──Mem0：用户个性化、三层记忆（user/session/agent）│

│└──Zep：时序知识图谱、事实变化追踪│

││

│工具层：MCP+A2A协议│

│├──MCP：Agent连接工具的标准化协议│

│└──A2Av1.0：Agent间通信的标准化协议（150+组织支持）│

││

│安全层：AEGIS预执行防火墙+AgentTrust运行时拦截│

│├──深度防御四层架构（输入→推理→执行→监控）│

│└──OWASPAgenticTop10全覆盖│

││

│评估层：agentevals+LangSmith/Langfuse│

│├──agentevals：跨模型、跨框架的持续评估│

│└──LangSmith/Langfuse：全链路追踪和可观测性│

││

└─────────────────────────────────────────────────────────────┘2.3架构设计文档模板完成需求分析和技术选型后，你需要产出一份架构设计文档。以下是标准模板：#[项目名称]系统架构设计文档

##1.业务背景与目标

-业务痛点：（量化描述）

-预期效果：（量化指标）

-目标用户：（用户画像）

##2.系统架构概览

-整体架构图：（四层架构——感知→规划→记忆→执行）

-Agent角色定义：（每个Agent的职责和能力边界）

-Agent间协作流程：（Mermaid序列图）

##3.技术栈选型

|层次|技术选型|选型理由|

|------|----------|----------|

|编排层|LangGraph1.0|...|

|知识层|LlamaIndex+LlamaParse|...|

|协作层|AutoGen0.4/CrewAI|...|

|记忆层|Mem0/Zep|...|

|工具层|MCP+A2A|...|

##4.核心工作流设计

-主工作流StateGraph设计

-关键节点的输入输出定义

-条件路由和异常处理

##5.安全架构

-威胁建模（OWASPAgenticTop10覆盖）

-深度防御四层设计

-人工审批触发条件

##6.部署架构

-基础设施需求

-容器化方案

-监控和告警策略

##7.里程碑与交付计划

|阶段|时间|交付物|验收标准|

|------|------|--------|----------|

|Phase1|第1-2周|架构设计文档|技术选型评审通过|

|Phase2|第3-6周|MVP原型|核心场景跑通|

|Phase3|第7-10周|生产级系统|安全+性能验收|

|Phase4|第11-12周|完整交付|全部交付物完成|第三章：Phase2——分阶段实现（第3-10周）3.1MVP阶段（第3-4周）：最小可行验证MVP阶段的目标不是“做一个完整的系统”，而是用最少的工作量验证核心假设——这个Agent真的能帮客户解决问题。MVP的范围定义：选择一个最小但完整的业务场景。如果是金融投研Agent（方向A），只做“单家公司Q2财报分析”这一个场景；如果是电商运营Agent（方向B），只做“客服自动回复”这一个场景。覆盖从输入到输出的完整链路。Agent能成功完成该场景的60-70%任务。MVP的技术实现：MVP技术栈（最小化）：

├──LangChaincreate_agent（快速搭建，暂不需要复杂的LangGraph工作流）

├──LlamaIndex基础索引（向量检索即可）

├──2-3个MCP工具（查询数据库+搜索+文件读取）

├──基础SystemPrompt（不使用复杂的多层Prompt架构）

└──MemorySaver（开发阶段，不做持久化）MVP的验收标准：核心场景能跑通（端到端成功执行）、输出质量基本可接受（不需要完美）、能演示给潜在客户看。如果MVP验证失败（Agent无法可靠地完成核心任务），应该在这个阶段停下来重新评估技术方案，而不是继续投入。3.2核心功能阶段（第5-8周）：从“能跑”到“能用”MVP验证通过后，进入核心功能开发阶段。这个阶段的目标是将MVP的“演示级”代码升级为“可用级”系统。核心功能开发Checklist：工作流升级：将简单的create_agent替换为LangGraphStateGraph，增加条件路由和人工审核节点，实现PostgresSaver状态持久化（替代MemorySaver）。知识检索升级：使用LlamaParse替代基础PDFReader，构建混合索引（向量+BM25+元数据过滤），实现AgenticRAG循环（多轮检索+自我评估+自动纠错）。工具系统升级：标准化所有工具为MCPServer，实现渐进式工具披露（ProgressiveToolDisclosure），增加工具调用的安全验证和权限分级。多Agent协作（如果适用）：使用AutoGenSelectorGroupChat实现智能调度，或使用CrewAI实现角色驱动的任务编排。关键设计原则：每个Agent的职责必须正交，Agent的description字段精确到能被LLM准确判断调度时机。记忆系统：集成Mem0（三层记忆：user/session/agent），实现跨会话用户画像和上下文保持。3.3生产加固阶段（第9-10周）：从“能用”到“敢用”生产加固阶段的目标是让系统具备企业级的安全性、可靠性和可运维性。生产加固Checklist：安全加固：实现四层深度防御架构（输入层→推理层→执行层→监控层）集成AEGIS预执行工具调用防火墙配置Docker沙箱执行环境（use_docker=True）实现人工审批节点（高风险操作强制人工确认）完成OWASPAgenticTop10威胁建模和缓解性能优化：实施模型分层策略（Tier1旗舰模型处理核心推理，Tier3开源模型处理批量任务）配置语义缓存（降低重复查询的Token消耗）设置对话轮次硬限制（防止无限循环消耗Token）实施工具调用结果缓存（减少重复工具调用）可观测性：集成LangSmith或Langfuse全链路追踪配置OpenTelemetrySpan追踪（每个LLM调用、工具调用、检索作为独立Span）部署实时监控仪表盘（延迟、Token消耗、错误率、成本）配置漂移检测告警（行为漂移+上下文漂移）部署就绪：容器化（DockerCompose或Kubernetes）CI/CD管道（自动化测试+评估门禁+灰度发布）灾备与恢复策略（Checkpointer断点续跑验证）运维手册（启动/停止/监控/故障恢复步骤）3.4阶段交付物清单阶段交付物验收标准MVP（第3-4周）可运行的MVP代码+演示脚本核心场景端到端跑通核心功能（第5-8周）完整Agent系统代码+单元测试+集成测试所有核心功能可用，测试覆盖率>80%生产加固（第9-10周）安全审计报告+性能测试报告+部署方案+运维手册安全审查通过，性能达标第四章：Phase3——测试策略与评估体系（持续贯穿）4.1四维质量体系基于模块十的五层并行评估体系，Capstone项目需要建立四维质量保障：维度一：单元测试——测试每个独立的Agent函数和工具函数。包括工具调用的参数验证、Prompt模板的变量替换、状态转换的逻辑正确性。维度二：集成测试——测试Agent之间的协作流程。包括LangGraph工作流的节点流转、多Agent间的消息传递、工具调用的端到端链路。维度三：场景测试——使用真实业务场景进行端到端测试。准备至少10个覆盖正常场景、边界场景和异常场景的测试用例。场景测试是发现“腐败成功”的最有效手段——Agent表面完成任务但过程违规的情况。维度四：压力测试——测试系统在高负载下的表现。包括并发请求下的响应延迟、大量工具调用时的Token消耗、Checkpointer的读写性能。4.2评估指标定义为Capstone项目定义明确的评估指标（基于模块十的十一维评估分类法）：指标类别具体指标Capstone目标值测量方法任务成功率端到端任务完成率>80%自动化场景测试工具使用准确率正确选择+正确参数>90%LangSmith/Langfuse追踪效率单任务Token消耗持续优化OTelSpan统计安全合规安全测试通过率100%安全审计响应延迟P95端到端延迟<30秒性能测试可靠性连续运行无故障时间>72小时稳定性测试4.3腐败成功检测基于模块十中PAE研究的教训——27-78%的基准成功可能是腐败成功——Capstone项目必须包含专门的腐败成功检测。检查Agent是否跳过了必要的验证步骤、检查Agent是否调用了预期之外的工具、检查工具调用结果与最终输出是否一致（如果工具返回了错误但Agent输出中未提及，这是忠诚度违反的信号）。第五章：Phase4——部署与交付（第11-12周）5.1容器化部署方案生产级Agent系统的标准部署架构：┌─────────────────────────────────────────────────────────────┐

│生产部署架构│

├─────────────────────────────────────────────────────────────┤

││

│接入层：Nginx/APIGateway（SSL终止、限流、认证）│

│↓│

│应用层：LangGraphServer（FastAPI+Uvicorn）│

│├──AgentRuntime（LangGraph工作流引擎）│

│├──MCPClient（工具服务器连接管理）│

│└──A2AInterface（跨Agent通信）│

│↓│

│数据层：│

│├──PostgreSQL（Checkpointer状态持久化）│

│├──Qdrant/Milvus（向量数据库——知识检索）│

│├──Redis（语义缓存+短期记忆）│

│└──Mem0/Zep（长期记忆管理）│

│↓│

│监控层：LangSmith/Langfuse+OpenTelemetry+Grafana│

││

└─────────────────────────────────────────────────────────────┘DockerCompose快速部署配置示例：#docker-compose.yml

version:'3.8'

services:

agent-server:

build:.

ports:

-"8000:8000"

environment:

-DATABASE_URL=postgresql://user:pass@postgres:5432/agent_db

-QDRANT_URL=http://qdrant:6333

-REDIS_URL=redis://redis:6379

-LANGSMITH_API_KEY=${LANGSMITH_API_KEY}

depends_on:

-postgres

-qdrant

-redis

postgres:

image:pgvector/pgvector:pg16

environment:

-POSTGRES_USER=user

-POSTGRES_PASSWORD=pass

-POSTGRES_DB=agent_db

volumes:

-postgres_data:/var/lib/postgresql/data

qdrant:

image:qdrant/qdrant:latest

volumes:

-qdrant_data:/qdrant/storage

redis:

image:redis:7-alpine

volumes:

postgres_data:

qdrant_data:5.2交付物完整清单交付物说明格式系统架构设计文档含架构图、Agent角色定义、工作流设计、技术选型理由PDF/Markdown完整可运行代码GitHub仓库，含README和安装说明GitRepositoryAPI文档Agent系统的API接口说明（OpenAPI规范）OpenAPI/Swagger单元测试+集成测试测试覆盖率>80%代码+报告安全审计报告OWASPAgenticTop10覆盖、威胁建模、渗透测试结果PDF性能测试报告响应时间、并发能力、Token消耗估算PDF部署方案DockerCompose/K8s配置+部署步骤YAML+Markdown运维手册启动/停止/监控/故障恢复/备份恢复步骤Markdown商业价值分析目标客户、ROI计算、定价策略、竞品对比PDF项目演示视频5分钟以内，展示核心场景的端到端执行MP4第六章：项目展示与商业提案——让你的作品“说话”6.1演示视频的五分钟黄金结构你的Capstone演示视频应该是5分钟以内、结构清晰、重点突出的。推荐结构如下：0:00-0:30：问题引入展示你要解决的业务痛点。最好用真实场景或数据来说明——“这是一个分析师团队，每人每天花4小时写报告”比“我们的Agent可以做金融分析”更有说服力。0:30-2:00：核心功能演示展示Agent系统完成一个完整的端到端任务。不要跳过任何步骤——让观众看到从用户输入到最终输出的完整过程。如果步骤太多，选择一个最有代表性的场景深入演示。2:00-3:30：技术亮点展示展示2-3个技术亮点，例如：Agent如何在多个工具之间自主选择和切换、Agent如何检测到检索结果不足并自动重新检索、安全护栏如何在Agent尝试危险操作时拦截。每个亮点控制在30秒左右。3:30-4:30：商业价值说明用数字说话。展示ROI计算结果——传统方式需要多少时间/人力/成本，Agent方式只需要多少。4:30-5:00：总结与联系方式一句话总结你的Agent的核心价值，留下联系方式或GitHub链接。6.2商业提案的关键数字基于模块十一的ROI计算框架，你的商业提案至少需要包含以下数字：指标计算方式示例值传统年人力成本人数×年薪×管理系数(1.3-1.5)10人×¥200K×1.4=¥2.8MAgent年运行成本API费+基础设施+运维人力¥200K自动化覆盖率Agent能自动完成的任务占比75%可靠性系数Agent实际成功率92%年化净收益传统成本×覆盖率×可靠性-Agent成本¥1.73MROI倍数净收益/Agent成本8.7x回本周期初始投资/月均净收益2-3个月6.3GitHub仓库的专业展示你的Capstone代码仓库是你的“技术名片”。以下是专业展示的关键要素：README.md：包含项目概述、核心功能、架构图、快速开始、技术栈、演示链接清晰的目录结构：src/（源代码）、tests/（测试）、docs/（文档）、deploy/（部署配置）、examples/（使用示例）完整的测试：包含单元测试和集成测试，CI管道自动运行详细的注释：关键代码有充分的注释说明设计决策License：选择合适的开源许可证第七章：项目评审与反馈——你的Capstone如何被评估7.1评审维度与权重维度权重评分标准架构设计25%Agent架构是否合理、模块边界是否清晰、是否考虑了扩展性、技术选型是否有充分理由代码质量20%代码规范、异常处理、测试覆盖、文档完整度、Git提交规范安全合规15%输入过滤、权限控制、审计日志、PII保护、OWASPAgenticTop10覆盖可观测性15%追踪、监控、告警、成本统计、漂移检测商业价值15%解决了什么真实问题、ROI如何、市场可行性创新性10%技术方案或商业模式是否有创新点7.2同行评审指南在提交之前，建议进行至少一次同行评审。评审要点包括：代码能否在另一台机器上成功运行（按README步骤操作）、核心工作流是否覆盖了正常和异常两条路径、安全措施是否有实际的拦截效果（不只是“配置了”）、商业提案中的ROI数字是否有合理的计算依据。7.3常见Capstone失败模式与避坑指南失败模式表现避坑方案范围过大设计了一个宏伟的系统但只完成了20%MVP阶段只做1个核心场景，完整实现后再扩展技术堆砌用了所有框架但都不深入选2-3个核心框架深耕，其他用最简单的方式实现Demo级代码能演示但不能部署从Day1就用生产级标准（环境变量、日志、错误处理、容器化）缺少安全设计能跑通但没有任何安全措施安全是设计阶段的问题，不是上线前再考虑的事商业故事空洞“这个Agent可以帮企业提效”但没有具体数字用ROI计算器量化商业价值，用具体的数字讲故事缺少测试改了代码不知道哪里坏了从MVP阶段就建立自动化测试，每次提交自动运行文档缺失代码写完了但没人能看懂README、架构文档、API文档、部署文档同等重要忽视演示系统很好但演示很糟糕录制演示视频前至少演练3次，确保端到端流程流畅第八章：从Capstone到职业跃迁——你的作品如何改变你的职业生涯8.1三种职业路径的Capstone策略路径一：求职面试Capstone是面试中最有力的“能力证明”。策略：将Capstone项目的架构图和核心代码放在GitHub首页，在面试中主动引导话题到Capstone项目，准备一个3分钟的“电梯演讲”版本展示核心价值。关键：面试官不会看完整代码，但会看README、架构图和演示视频。路径二：独立接单Capstone是你接单时的“样品间”。策略：将Capstone项目包装为行业解决方案模板，在与潜在客户沟通时展示Capstone的演示视频。关键：客户关心的不是技术细节，而是“这个Agent能帮我省多少钱”。路径三：创业融资Capstone是你的“MVP验证”。策略：在Capstone项目的基础上进行真实客户验证（找到3-5个付费或免费试用的种子客户），用验证数据支撑商业计划书中的市场假设。8.2Capstone作品的长期价值Capstone项目不是你学习生涯的终点，而是你AgenticAI职业生涯的起点。你的Capstone代码库将成为你未来项目的“起点模板”——下一个客户项目不需要从零开始。你的架构设计文档将在面试和客户沟通中反复使用——它可以被不断迭代优化。你的商业提案模板是通用的——换一个行业、换一组数字，就能适配新的场景。最好的Capstone是那些在课程结束后仍然在持续迭代的项目。代码实战Capstone脚手架代码——企业级Agent系统启动模板"""

Capstone项目脚手架——企业级Agent系统启动模板

本代码提供了Capstone项目的标准启动框架，集成了：

1.LangGraph工作流编排

2.LlamaIndex知识检索

3.MCP工具集成

4.Mem0记忆系统

5.安全护栏（AEGIS模式）

6.OpenTelemetry可观测性

使用方法：

1.复制此模板到你的Capstone项目

2.根据你的Capstone方向，实现TODO标记的方法

3.逐步替换模拟实现为真实的生产代码

作者：AgenticAI实战大师班

版本：v12.0

"""

importasyncio

importjson

importos

importtime

importuuid

importlogging

fromtypingimportDict,List,Any,Optional,TypedDict,Annotated,Callable

fromoperatorimportadd

fromdatetimeimportdatetime

fromdataclassesimportdataclass,field

fromcontextlibimportcontextmanager

fromlanggraph.graphimportStateGraph,END

fromlanggraph.checkpoint.postgresimportPostgresSaver

fromlanggraph.typesimportinterrupt

logging.basicConfig(level=logging.INFO)

logger=logging.getLogger(__name__)

#============================================================

#Part1:项目配置管理

#============================================================

@dataclass

classCapstoneConfig:

"""Capstone项目全局配置"""

#项目信息

project_name:str="enterprise-agent-system"

project_direction:str="A"#A/B/C/D

#模型配置（模块一：模型选型决策框架）

primary_model:str="claude-opus-4-7"

fallback_model:str="gpt-5.4"

economy_model:str="deepseek-v4"

#数据库配置

postgres_url:str=field(default_factory=lambda:os.getenv(

"DATABASE_URL","postgresql://user:pass@localhost:5432/agent_db"

))

qdrant_url:str=field(default_factory=lambda:os.getenv(

"QDRANT_URL","http://localhost:6333"

))

redis_url:str=field(default_factory=lambda:os.getenv(

"REDIS_URL","redis://localhost:6379"

))

#记忆系统配置（模块六）

memory_provider:str="mem0"#mem0/zep

#安全配置（模块九）

security_level:str="standard"#standard/strict/paranoid

docker_sandbox:bool=True

#可观测性配置（模块十）

tracing_enabled:bool=True

tracing_provider:str="langsmith"

#============================================================

#Part2:Agent状态定义（模块三：LangGraphState）

#============================================================

classCapstoneAgentState(TypedDict):

"""Capstone项目的标准Agent状态"""

#消息历史

messages:Annotated[List[dict],add]

#任务追踪

task_id:str

task_description:str

task_type:str

current_step:str

step_results:dict

#检索状态（模块四：LlamaIndex）

retrieved_context:List[str]

retrieval_quality_score:float

#工具调用状态（模块七：MCP工具）

tool_calls_made:int

tool_results:dict

#安全状态（模块九）

risk_score:float

human_approval_required:bool

human_decision:Optional[str]

security_audit_log:List[dict]

#记忆状态（模块六）

memory_context:Optional[str]

user_profile:Optional[dict]

#执行统计

errors_encountered:List[str]

execution_start_time:str

#============================================================

#Part3:核心工作流节点

#============================================================

deftask_understanding_node(state:CapstoneAgentState)->dict:

"""

任务理解节点——模块一（认知架构）的感知层+模块二（PromptEngineering）

TODO:根据你的Capstone方向实现具体的任务理解逻辑

"""

task=state["task_description"]

return{

"current_step":"understanding",

"step_results":{

**state.get("step_results",{}),

"understanding":{

"intent":f"分析任务:{task[:100]}",

"complexity":"medium",

"domain":"待定",

}

},

"messages":[{

"role":"agent",

"content":f"任务理解完成:{task[:100]}"

}],

"execution_start_time":datetime.now().isoformat(),

}

defknowledge_retrieval_node(state:CapstoneAgentState)->dict:

"""

知识检索节点——模块四（LlamaIndexAgenticRAG）

TODO:集成LlamaIndex+LlamaParse进行文档检索

"""

task=state["task_description"]

#模拟检索结果

retrieved=[

"相关文档片段1:...",

"相关文档片段2:...",

]

return{

"current_step":"retrieval",

"retrieved_context":retrieved,

"retrieval_quality_score":0.85,

"step_results":{

**state.get("step_results",{}),

"retrieval":{

"documents_found":len(retrieved),

"quality_score":0.85,

}

},

"messages":[{

"role":"agent",

"content":f"知识检索完成:找到{len(retrieved)}个相关文档"

}],

}

deftool_execution_node(state:CapstoneAgentState)->dict:

"""

工具执行节点——模块七（MCP工具集成）

TODO:集成MCPServer进行工具调用

"""

return{

"current_step":"execution",

"tool_calls_made":state.get("tool_calls_made",0)+1,

"step_results":{

**state.get("step_results",{}),

"execution":{"status":"success","tools_used":["search","analyze"]}

},

"messages":[{

"role":"agent",

"content":"工具执行完成"

}],

}

defsecurity_check_node(state:CapstoneAgentState)->dict:

"""

安全审查节点——模块九（AEGIS预执行防火墙）

TODO:集成安全护栏进行风险检测

"""

risk_score=state.get("risk_score",0.0)

#模拟安全检查

human_required=risk_score>0.6

ifhuman_required:

decision=interrupt({

"type":"security_review",

"task_id":state["task_id"],

"risk_score":risk_score,

"message":"高风险操作需要人工审批",

})

return{"human_decision":decision}

return{

"human_approval_required":False,

"security_audit_log":[{

"timestamp":datetime.now().isoformat(),

"verdict":"ALLOW",

"risk_score":risk_score,

}],

}

defresult_synthesis_node(state:CapstoneAgentState)->dict:

"""

结果合成节点——综合所有前置步骤的结果

TODO:根据你的Capstone方向实现具体的结果合成逻辑

"""

all_results=state.get("step_results",{})

return{

"current_step":"completed",

"step_results":{**all_results,"synthesis":{"status":"completed"}},

"messages":[{

"role":"agent",

"content":"任务执行完成，结果已合成"

}],

}

defmemory_update_node(state:CapstoneAgentState)->dict:

"""

记忆更新节点——模块六（Mem0/Zep长期记忆）

TODO:集成Mem0存储跨会话记忆

"""

return{

"memory_context":f"记忆已更新:{state['task_id']}",

"messages":[{

"role":"system",

"content":"记忆系统已更新"

}],

}

#============================================================

#Part4:条件路由函数

#============================================================

defroute_after_understanding(state:CapstoneAgentState)->str:

"""任务理解后的路由"""

complexity=state.get("step_results",{}).get("understanding",{}).get("complexity","medium")

ifcomplexity=="high":

return"knowledge_retrieval"

return"tool_execution"

defroute_after_security(state:CapstoneAgentState)->str:

"""安全审查后的路由"""

ifstate.get("human_decision")=="rejected":

returnEND

return"result_synthesis"

#============================================================

#Part5:构建Capstone工作流

#============================================================

defbuild_capstone_workflow(direction:str="A")->StateGraph:

"""

构建Capstone工作流

根据Capstone方向定制工作流结构：

-方向A（金融投研）：理解→检索→分析→生成→审核

-方向B（电商运营）：路由→专业Agent→结果聚合

-方向C（知识管理）：解析→索引→检索→问答

-方向D（内容创作）：监控→策划→写作→优化→发布

"""

workflow=StateGraph(CapstoneAgentState)

#通用节点（所有方向都需要）

workflow.add_node("task_understanding",task_understanding_node)

workflow.add_node("knowledge_retrieval",knowledge_retrieval_node)

workflow.add_node("tool_execution",tool_execution_node)

workflow.add_node("security_check",security_check_node)

workflow.add_node("result_synthesis",result_synthesis_node)

workflow.add_node("memory_update",memory_update_node)

#设置入口

workflow.set_entry_point("task_understanding")

#通用工作流

workflow.add_conditional_edges(

"task_understanding",

route_after_understanding,

{

"knowledge_retrieval":"knowledge_retrieval",

"tool_execution":"tool_execution",

}

)

workflow.add_edge("knowledge_retrieval","tool_execution")

workflow.add_edge("tool_execution","security_check")

workflow.add_conditional_edges(

"security_check",

route_after_security,

{"result_synthesis":"result_synthesis",END:END}

)

workflow.add_edge("result_synthesis","memory_update")

workflow.add_edge("memory_update",END)

returnworkflow

#============================================================

#Part6:Capstone项目管理器

#============================================================

classCapstoneProject:

"""

Capstone项目管理器——管理项目的完整生命周期

核心功能：

-工作流编译和执行

-配置管理

-状态持久化

-交付物生成

"""

def__init__(self,config:CapstoneConfig):

self.config=config

self.workflow=build_capstone_workflow(ject_direction)

self.checkpointer=None

self.app=None

self.execution_history:List[Dict]=[]

asyncdefinitialize(self):

"""初始化Capstone项目"""

#配置Checkpointer（生产环境使用PostgresSaver）

try:

self.checkpointer=PostgresSaver.from_conn_string(self.config.postgres_url)

self.checkpointer.setup()

("✅PostgresSaver初始化成功")

exceptExceptionase:

fromlanggraph.checkpoint.memoryimportMemorySaver

logger.warning(f"⚠️PostgresSaver不可用，使用MemorySaver（仅开发环境）:{e}")

self.checkpointer=MemorySaver()

#编译工作流

self.app=pile(checkpointer=self.checkpointer)

(f"✅Capstone工作流编译完成[{ject_direction}]")

asyncdefrun_task(self,task_description:str,session_id:Optional[str]=None)->Dict[str,Any]:

"""执行一个任务"""

ifnotself.app:

awaitself.initialize()

task_id=str(uuid.uuid4())[:8]

session_id=session_idorf"session-{task_id}"

initial_state={

"task_id":task_id,

"task_description":task_description,

"task_type":"analysis",

"messages":[],

"current_step":"initialized",

"step_results":{},

"retrieved_context":[],

"retrieval_quality_score":0.0,

"tool_calls_made":0,

"tool_results":{},

"risk_score":0.0,

"human_approval_required":False,

"security_audit_log":[],

"memory_context":None,

"user_profile":None,

"errors_encountered":[],

}

config={"configurable":{"thread_id":session_id}}

start_time=time.time()

final_state=awaitself.app.ainvoke(initial_state,config)

execution_time=time.time()-start_time

result={

"task_id":task_id,

"session_id":session_id,

"status":"completed",

"execution_time_ms":round(execution_time*1000,2),

"steps_completed":list(final_state.get("step_results",{}).keys()),

"final_output":final_state.get("messages",[{}])[-1].get("content","")iffinal_state.get("messages")else"",

}

self.execution_history.append(result)

returnresult

defget_stats(self)->Dict[str,Any]:

"""获取项目统计"""

return{

"project_name":ject_name,

"direction":ject_direction,

"total_executions":len(self.execution_history),

"recent_executions":self.execution_history[-5:],

}

#============================================================

#Part7:使用示例

#============================================================

asyncdefdemo_capstone():

"""演示Capstone脚手架"""

print("="*60)

print("🏗️Capstone项目脚手架演示")

print("="*60)

#创建配置

config=CapstoneConfig(

project_name="financial-research-agent",

project_direction="A",

)

#初始化项目

project=CapstoneProject(config)

awaitproject.initialize()

#运行示例任务

result=awaitproject.run_task(

task_description="分析2026年Q2大中华区消费电子行业的营收趋势"

)

print(f"\n📊任务结果:")

print(f"任务ID:{result['task_id']}")

print(f"耗时:{result['execution_time_ms']}ms")

print(f"完成步骤:{result['steps_completed']}")

print(f"\n📈项目统计:")

print(f"{json.dump