《2026 Agentic AI 实战大师班：从提示到自主智能体系统》

《2026AgenticAI实战大师班：从提示到自主智能体系统》课程总览课程定位：面向中高级职场人士的高端实战课程，系统化掌握AgenticAI的架构设计、开发落地与商业变现全链路能力。2026年被公认为AIAgent商业化元年，能够自主拆解任务、调用工具并交付结果的智能体正在重塑财富分配规则。本课程助你从“会用AI的从业者”进阶为“能构建AI系统的架构师”，抓住本轮技术红利核心。核心承诺：12个模块学完，能独立设计并交付企业级多智能体系统掌握LangChain1.0+LangGraph、AutoGen0.4+、LlamaIndex、CrewAI四大框架的实战用法拥有金融、电商、内容创作、营销、制造等8+行业的真实案例和可直接复用的代码模板理解按效果付费（RaaS）、订阅制、项目制等多种商业变现路径目标学员画像：技术管理者（CTO/技术总监/架构师）：需要评估和落地Agent技术AI工程师/数据科学家：需要系统提升Agent开发能力产品经理/业务负责人：需要找到AI驱动业务增长的切入点咨询顾问/独立开发者：希望通过AgenticAI实现商业变现前置要求：Python基础+至少一个LLMAPI使用经验+对RAG/PromptEngineering有基本了解学完你将收获：认知升级：从“AI是工具”到“AI是系统”的思维跃迁技术硬实力：能独立搭建生产级Agent系统，包括规划、记忆、工具调用、多Agent协作实战作品集：完成4个可直接部署的企业级项目（金融分析/电商运营/内容工厂/智能运维）商业变现力：掌握至少3种AgenticAI变现路径，具备接单/创业/升职的核心竞争力模块一：AgenticAI认知重构——从“提示词”到“自主系统”的思维跃迁学习目标理解AgenticAI与对话式AI的本质区别掌握Agent的四层架构模型（感知→规划→记忆→执行）2026主流模型选型能力：ClaudeOpus4.7vsGPT-5.4vsGemini3.1Pro的场景化选择建立“以终为始”的商业思维：先找到利润池，再设计技术方案核心理论讲解1.什么是AgenticAI？传统AI交互是“输入→输出”的线性模式，而AgenticAI是一个持续运行的自主系统，具备四大核心能力：环境感知（Perception）、任务规划（Planning）、记忆管理（Memory）和工具执行（Action）。2.四层架构深度拆解：感知层：实时接入多模态数据，不再依赖静态训练数据规划层：面对复杂指令自动拆解子任务，支持ReAct/Plan-and-Solve/Tree-of-Thoughts等多种策略记忆层：瞬时记忆（会话级）+短期记忆（Redis/会话缓存）+长期记忆（向量数据库），实现千人千面执行层：通过API调用支付、物流、办公软件等，完成端到端闭环3.2026模型选型决策框架：截至2026年5月，主流旗舰模型格局：ClaudeOpus4.7以SWE-bench72.7%确立代码工程优势，Gemini3.1Pro以MCPAtlas78.2%在多步骤工作流领先，GPT-5.4在通用任务平衡性最佳。选型核心原则：匹配核心业务场景，而非追求单一榜单分数。场景推荐模型核心理由AgenticCoding/长周期自动任务ClaudeOpus4.7工具调用错误率降低50-75%，长时间保持上下文一致性多步骤工作流/企业系统编排Gemini3.1ProMCPAtlas78.2%，OSWorld76.2%通用任务/内容创作/混合推理GPT-5.4最均衡，ScaleMultiChallenge69.6%开源/成本敏感/私有化部署DeepSeekV4/Qwen3.6DeepAgents的HarnessProfiles支持开放模型达到生产级性能，成本降低20倍以上4.商业思维先于技术思维：核心利润公式：利润=(传统人力成本−Agent运行成本)×自动化规模。寻找高价值低效率的堵点——哪里的人力成本最贵、流程最繁琐，哪里就是Agent最赚钱的地方。企业级案例案例1：金融分析——从“人工写报告”到“Agent自主分析”某金融机构的量化分析场景：传统模式下分析师需要4小时完成一份行业研究报告；引入Agent后，系统自动完成数据采集→清洗→建模→可视化→报告生成，耗时降至8分钟，且洞察稳定性显著提升。案例2：电商运营——明略科技AgenticAI全链路改造秒针系统历时60天完成全链路AgenticAI改造，全链路AI自动完成率达90%，深度复盘报告产出实现20倍人效提升，营销智能业务交付效率提升最高达4倍。关键是“全链路”——从数据抓取、分析框架搭建到洞察提炼、报告生成，每个环节都被纳入Agent调度框架。案例3：AgenticCommerce——Visa智能支付AgentVisa已完成数百笔安全的Agent自主交易验证，近半数美国消费者（47%）已使用AI进行购物任务。2026年，AIAgent将不再只是辅助购物——它们将直接完成购买。Visa与100+合作伙伴构建的AgenticCommerce生态正在改变零售支付范式。Prompt模板##通用Agent任务拆解模板

你是一个{domain}领域的AIAgent系统架构师。请针对以下业务需求，设计Agent的完整执行方案。

【业务需求】

{具体业务描述}

请从以下四个层次给出方案：

1.**感知层**：需要接入哪些数据源？数据格式是什么？更新频率如何？

2.**规划层**：任务如何拆解为子任务？画出依赖关系图（DAG）

3.**记忆层**：哪些信息需要持久化？短期/长期记忆分别存储什么？

4.**执行层**：需要调用哪些外部工具/API？失败重试策略是什么？

输出格式：结构化Markdown，包含具体的工具名称和API描述。代码示例：模型选型工厂模式"""

模型选型工厂——根据任务类型自动选择最优模型

适用于2026年5月主流模型生态

"""

fromenumimportEnum

fromdataclassesimportdataclass

fromtypingimportOptional

importos

classTaskType(Enum):

CODING_AGENT="coding_agent"#代码生成/工程化任务

LONG_WORKFLOW="long_workflow"#长时间多步骤工作流

CONTENT_CREATION="content"#内容创作/通用推理

MULTI_AGENT="multi_agent"#多Agent协调

COST_SENSITIVE="cost_sensitive"#成本敏感/批量任务

@dataclass

classModelConfig:

provider:str

model_name:str

input_price_per_1m:float#每百万token价格

output_price_per_1m:float

max_context:int

#2026年5月主流模型定价与能力矩阵

MODEL_REGISTRY={

TaskType.CODING_AGENT:ModelConfig(

provider="anthropic",model_name="claude-opus-4-7",

input_price_per_1m=5.0,output_price_per_1m=25.0,max_context=200_000

),

TaskType.LONG_WORKFLOW:ModelConfig(

provider="google",model_name="gemini-3.1-pro",

input_price_per_1m=3.5,output_price_per_1m=10.5,max_context=1_000_000

),

TaskType.CONTENT_CREATION:ModelConfig(

provider="openai",model_name="gpt-5.4",

input_price_per_1m=5.0,output_price_per_1m=20.0,max_context=256_000

),

TaskType.COST_SENSITIVE:ModelConfig(

provider="deepseek",model_name="deepseek-v4",

input_price_per_1m=0.5,output_price_per_1m=2.0,max_context=128_000

),

}

classModelRouter:

"""智能模型路由器——根据任务类型和预算自动选模"""

def__init__(self,budget_limit_usd:Optional[float]=None):

self.budget_limit=budget_limit_usd

self.usage_tracker={}#实际用量追踪

defroute(self,task_type:TaskType)->ModelConfig:

"""根据任务类型返回最优模型配置"""

config=MODEL_REGISTRY.get(task_type)

ifnotconfig:

config=MODEL_REGISTRY[TaskType.CONTENT_CREATION]

#如果设置了预算上限且当前模型太贵，自动降级

ifself.budget_limitandconfig.output_price_per_1m>self.budget_limit:

config=MODEL_REGISTRY[TaskType.COST_SENSITIVE]

returnconfig

#使用示例

router=ModelRouter(budget_limit_usd=3.0)

coding_model=router.route(TaskType.CODING_AGENT)

print(f"代码任务使用:{coding_vider}/{coding_model.model_name}")动手练习练习：设计你的第一个Agent架构方案选择一个你熟悉的业务场景（如：客户工单自动分类与处理、市场竞品日报生成）使用四层架构模型，画出Agent的完整设计方案标注每个层次涉及的具体技术和工具评估标准：✓四个层次都有具体的技术选型（框架/工具/模型名称）✓至少识别出一个“人工→自动”的效率提升点✓考虑了异常处理（工具调用失败、数据缺失等边界情况）常见坑与避坑指南坑表现避坑方案模型崇拜症所有任务都用最贵的旗舰模型，月账单惊人建立模型路由层，简单任务用开源模型（DeepSeek/Qwen），复杂任务才调用旗舰模型，成本可降低10-20倍过度工程化一个5分钟手动完成的任务也要建Agent系统先计算ROI，只有高频×高价值×多步骤的任务才值得Agent化忽视模型特性差异用GPT-5.4做长周期代码工程，效果不佳ClaudeOpus4.7的AgenticCoding工具调用错误减少50-75%，选模型必须匹配场景幻觉风险金融/医疗等高风险场景直接输出结果给用户必须加入Human-in-the-Loop审核机制商业变现建议咨询/架构设计服务：很多传统企业有Agent需求但缺乏架构能力，可提供“Agent方案设计+技术选型”咨询服务，单项目收费5-30万行业Agent模板：将某个行业的Agent方案做成可复用的模板（如“金融研报Agent模板”），SaaS化按年收费成本优化顾问：帮助企业审核Agent系统架构，减少模型调用成本，按节省金额的15-20%收费模块二：PromptEngineering2.0——面向Agent系统的提示工程新范式学习目标理解Agent时代的PromptEngineering已从“单轮优化”进化为“多轮指令链设计”掌握SystemPrompt+TaskPrompt+ToolDescription的三角架构学会设计和优化Agent的工具描述（ToolDescription），直接影响工具调用准确率掌握Prompt版本管理和A/B测试方法核心理论讲解1.PromptEngineering的范式跃迁传统PromptEngineering关注的是“如何让模型输出更好的文本”，而Agent时代的PromptEngineering关注的是“如何让模型成为可靠的决策者和执行者”。这要求Prompt不仅要描述“做什么”，还要定义“如何思考”“何时调用工具”“如何处理失败”。2.AgentPrompt三角架构：┌──────────────────────────────────────────┐

│SystemPrompt（系统提示）│

│定义Agent角色、行为边界、安全约束│

│示例：“你是一位严谨的金融分析师，任何│

│投资建议前必须引用至少两个权威数据源”│

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

│ToolDescriptions（工具描述）│

│精确描述每个工具的功能、参数、适用场景和│

│返回值格式。这是影响工具调用准确率的│

│最关键因素，GPT-5在τ2-bench得分96.7%│

└──────────────────────────────────────────┘

┌──────────────────────────────────────────┐

│TaskPrompt（任务提示）│

│具体任务描述，包含上下文、约束和期望输出格式│

│支持动态变量注入（从记忆层/工具调用结果获取）│

└──────────────────────────────────────────┘3.工具描述（ToolDescription）的撰写艺术工具描述是Agent正确调用工具的关键。一个优秀的工具描述应包含：功能描述：一句话说明工具做什么适用场景：什么情况下应该调用（正面示例）和不应该调用（负面示例）参数定义：每个参数的类型、含义、是否必填、默认值、示例值返回格式：返回值的数据结构，便于Agent解析和下一步决策异常处理：调用失败时返回什么，Agent应如何反应企业级案例案例1：金融风控Agent的工具链Prompt设计某银行风控部门将反欺诈规则验证从72小时缩短至8小时，关键是让Agent能够精确理解每个风控工具的作用边界，减少误调用。案例2：电商智能客服的多工具协同某电商平台客服Agent需要同时调用用户画像服务、库存管理系统和物流接口，工具描述的精确度直接决定调用成功率。案例3：内容创作Agent的迭代优化某内容团队使用AgenticAI进行多平台内容分发，通过A/B测试不同Prompt版本，将内容互动率提升37%。Prompt模板##Agent工具描述标准模板

【工具名称】：{tool_name}

【一句话描述】：{一句话说明工具的核心功能}

【适用场景】：

-应该调用时：{列出正向触发条件}

-不应调用时：{列出负向排除条件}

【参数规格】：

|参数名|类型|必填|默认值|描述|示例值|

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

|{param1}|{type}|{Y/N}|{default}|{desc}|{example}|

【返回值】：

```json

{

"status":"success|error",

"data":{具体数据结构},

"error_message":"仅在status为error时出现"

}【异常处理指引】：超时（>10s）：重试最多3次，指数退避返回空数据：向用户确认查询条件是否正确权限不足：告知用户需要什么权限，不要尝试绕过

###代码示例：LangChainAgent的工具描述框架

```python

"""

LangChain1.0Agent工具定义与Prompt管理

使用LangChain1.0的create_agent声明式API

"""

fromlangchain.agentsimportcreate_agent

fromlangchain.toolsimporttool

fromlangchain.memoryimportVectorStoreMemory

frompydanticimportBaseModel,Field

fromtypingimportList,Optional

importjson

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

#Step1:使用Pydantic精确描述工具参数（这是提高工具调用准确率的关键）

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

classStockQueryInput(BaseModel):

"""股票查询工具的输入参数——描述越精确，Agent调用越准确"""

symbol:str=Field(

description="股票代码，如AAPL、600519.SH。注意：美股直接用代码，A股需加后缀",

examples=["AAPL","600519.SH","0700.HK"]

)

date_range:Optional[str]=Field(

default="1M",

description="查询时间范围，格式：1D(1天)/1M(1月)/3M(3月)/1Y(1年)",

examples=["1M","3M"]

)

indicators:List[str]=Field(

default=["close","volume"],

description="需要查询的技术指标：open/close/high/low/volume/rsi/macd",

)

classRiskAssessmentInput(BaseModel):

"""风险评估工具输入"""

portfolio_id:str=Field(description="投资组合ID")

risk_metrics:List[str]=Field(

default=["var","sharpe","max_drawdown"],

description="风险评估指标：var(风险价值)/sharpe(夏普比率)/max_drawdown(最大回撤)/beta"

)

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

#Step2:定义工具，附带详细的docstring作为工具描述

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

@tool(args_schema=StockQueryInput)

defquery_stock_data(symbol:str,date_range:str="1M",indicators:List[str]=None):

"""

查询股票历史行情数据和技术指标。

##适用场景

-应该调用时：用户询问某只股票的价格、走势、技术指标

-不应调用时：用户问的是宏观经济数据、行业分析（应使用query_macro_data）

##返回格式

返回JSON，包含日期、各指标值和查询状态

##注意事项

-A股代码必须带后缀（.SH/.SZ），如"600519.SH"而非"600519"

-实时数据延迟约15分钟（免费接口限制）

"""

ifindicatorsisNone:

indicators=["close","volume"]

#模拟数据（生产环境替换为真实API）

returnjson.dumps({

"status":"success",

"symbol":symbol,

"data":{

"latest_close":185.50,

"30d_high":198.20,

"30d_low":172.80,

"volume_avg":52_000_000,

"indicators":{ind:f"value_of_{ind}"forindinindicators}

}

})

@tool(args_schema=RiskAssessmentInput)

defcalculate_risk_metrics(portfolio_id:str,risk_metrics:List[str]=None):

"""

计算投资组合的风险指标，用于风险评估和合规检查。

##适用场景

-应该调用时：用户要求评估投资组合风险、计算VaR/夏普比率等

-不应调用时：简单的股票查询（用query_stock_data）

##⚠️安全约束

-计算结果仅供参考，不构成投资建议

-风险评分>0.7时必须触发人工审核

"""

ifrisk_metricsisNone:

risk_metrics=["var","sharpe","max_drawdown"]

returnjson.dumps({

"status":"success",

"portfolio_id":portfolio_id,

"metrics":{m:round(hash(m)%100/100,3)forminrisk_metrics},

"risk_warning":"计算结果仅供参考，不构成投资建议"

})

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

#Step3:创建Agent——LangChain1.0声明式API

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

financial_agent=create_agent(

model="claude-opus-4-7",#金融场景选择Claude（工具调用更可靠）

tools=[query_stock_data,calculate_risk_metrics],

memory=VectorStoreMemory(),#长期记忆：向量数据库

system_prompt="""你是一位严谨的金融数据分析Agent。

##核心原则

1.**任何分析结论必须引用数据来源**（来自哪个工具的哪次调用）

2.**风险提示必须前置**：在给出任何分析前，先说明分析方法论的局限性

3.**绝对不给出“买入/卖出”建议**，只提供数据分析和风险评估

4.**当风险评分>0.7时，必须建议用户咨询专业投资顾问**

##工作流程

1.理解用户意图→判断需要哪些数据

2.调用工具获取数据→交叉验证

3.综合分析→输出结构化报告

4.如遇到工具调用失败，先重试，再降级（使用替代数据源）

""",

#关键参数调优

max_iterations=8,#金融分析需要多步推理

temperature=0.1,#金融场景需要确定性输出

early_stopping_method="generate",#达到目标后提前停止

)

#使用示例

response=financial_agent.invoke({

"input":"分析苹果公司(AAPL)过去一个月的股价表现和投资组合风险"

})

print(response["output"])动手练习练习1：为你的业务场景设计Agent工具描述选择一个你常用的外部工具/API（如：飞书文档、JIRA、数据库查询等）使用标准模板为该工具撰写完整的工具描述在工具描述中至少包含2个“应该调用”和2个“不应调用”的场景练习2：AgentPromptA/B测试为同一个任务设计两个不同风格的SystemPrompt（如：一个偏“严谨保守”，一个偏“灵活进取”）在同样任务下对比输出质量、工具调用准确率和token消耗记录对比结果，形成选型建议评估标准：✓工具描述包含至少5个要素（功能/场景/参数/返回值/异常处理）✓SystemPrompt明确定义了Agent的行为边界和安全约束✓A/B测试有量化对比数据常见坑与避坑指南坑表现避坑方案工具描述太模糊Agent频繁调用错误的工具或漏掉应有的调用工具描述中包含正反例，用Pydantic强类型约束参数SystemPrompt过于冗长Token消耗巨大但效果提升有限SystemPrompt控制在500字以内，核心约束优先，细节放到工具描述中温度参数一刀切创意任务过于死板，分析任务过于发散金融/法律/医疗场景temperature=0.1；创意内容temperature=0.7-0.9忽视Prompt版本管理改了Prompt不知道哪个版本效果好建立Prompt版本库，每次修改记录变更原因和效果对比商业变现建议Prompt工程咨询服务：为企业优化AgentPrompt，提升工具调用准确率和任务完成率行业Prompt模板库：制作“金融分析AgentPrompt包”“电商客服AgentPrompt包”等，按月订阅培训服务：为企业IT团队提供“AgentPromptEngineering”内训，1天课程收费1-3万模块三：LangChain1.0+LangGraph——构建生产级Agent的工业级框架学习目标掌握LangChain1.0的声明式Agent创建范式（create_agent）理解LangGraph的有向图驱动执行引擎学会使用DeepAgents构建具有内置规划、子Agent生成、长期记忆和上下文管理能力的长时运行Agent掌握DeepAgentsv0.6的新能力：代码解释器、HarnessProfiles、DeltaChannels、流式输出核心理论讲解1.LangChain1.0的架构革新LangChain1.0与LangGraph的发布标志着Agent框架正式进入生产就绪阶段。核心变化：声明式API：create_agent()替代传统手动编码，开发者只需定义工具链、记忆机制和规划策略LangGraph执行引擎：采用有向无环图（DAG）驱动，每个节点代表原子操作，边定义操作间依赖关系持久化状态管理：自动状态快照、多级存储策略、原子性操作，某金融机构采用后系统可用性提升至99.995%标准内容块：统一的多模型内容规范，使模型切换成本降低65%2.DeepAgents：面向长时运行任务的Agent框架DeepAgents是LangChain的开源Agentharness，提供内置的任务规划、子Agent生成、长期记忆和上下文管理，支持运行数分钟到数小时、跨越数十个步骤的Agent。DeepAgentsv0.6四大新能力：代码解释器：Agent内部可编程工作空间，中间结果保留在运行时状态，减少模型上下文消耗HarnessProfiles：为开源模型（Kimi/Qwen/DeepSeek）提供调优配置，达到生产级性能，成本降低20倍以上DeltaChannels：将长时间运行的检查点存储减少高达100倍流式输出：类型化的事件投影，支持消息、工具调用、子Agent的实时订阅企业级案例案例1：电商订单处理工作流使用LangGraph的StateChart构建订单状态流转引擎，实现pending→paid→shipped→completed的自动流转，并在异常时触发人工介入。案例2：金融量化交易系统某金融机构使用LangGraph的持久化状态管理和复杂流程编排，实现动态分支、循环控制和人工审批节点的量化交易决策系统。案例3：内容生产Pipeline使用DeepAgents的代码解释器实现多主题并行研究、数据处理和报告生成，将原本需要数小时的内容生产压缩至数分钟。代码示例：LangGraph1.0企业级工作流"""

LangGraph1.0企业级Agent工作流——含持久化状态、人工介入、错误恢复

"""

fromlanggraph.graphimportStateGraph,END

fromlanggraph.checkpoint.memoryimportMemorySaver

fromtypingimportTypedDict,Literal,Annotated

fromoperatorimportadd

importjson

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

#Step1:定义工作流状态

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

classOrderState(TypedDict):

"""订单处理状态——LangGraph会自动为每一步做状态快照"""

order_id:str

status:str#pending→paid→shipped→completed

risk_score:float#风控评分

messages:Annotated[list,add]#操作日志（Annotated[list,add]实现追加而非覆盖）

human_approval_required:bool

human_decision:str#approved/rejected/revise

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

#Step2:定义节点函数（每个节点是一个原子操作）

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

definit_order(state:OrderState)->dict:

"""初始化订单"""

return{

"status":"pending",

"messages":[f"[INIT]订单{state['order_id']}创建成功"]

}

defrisk_assessment(state:OrderState)->dict:

"""风险评估节点"""

#模拟风控逻辑

risk=0.2#正常订单风险低

#模拟高风险场景

ifstate["order_id"]=="ORD-HIGH-001":

risk=0.85

return{

"risk_score":risk,

"messages":[f"[RISK]风险评估完成，得分:{risk:.2f}"],

"human_approval_required":risk>0.7

}

defroute_after_risk(state:OrderState)->Literal["auto_approve","human_review"]:

"""条件路由：根据风险评分决定下一步"""

ifstate["human_approval_required"]:

return"human_review"

return"auto_approve"

defauto_approve(state:OrderState)->dict:

"""自动审批"""

return{

"status":"paid",

"messages":[f"[APPROVE]风险评分{state['risk_score']:.2f}，自动通过审批"]

}

defhuman_review(state:OrderState)->dict:

"""人工审核节点——暂停等待人工决策"""

#实际部署时这里会触发飞书/钉钉/Slack通知

decision=state.get("human_decision","pending")

ifdecision=="approved":

return{

"status":"paid",

"messages":["[HUMAN]人工审核通过"]

}

elifdecision=="rejected":

return{

"status":"rejected",

"messages":["[HUMAN]人工审核拒绝"]

}

return{"status":"pending_review"}

defship_order(state:OrderState)->dict:

"""发货节点"""

return{

"status":"shipped",

"messages":[f"[SHIP]订单{state['order_id']}已发货"]

}

defcomplete_order(state:OrderState)->dict:

"""完成节点"""

return{

"status":"completed",

"messages":[f"[DONE]订单{state['order_id']}处理完成"]

}

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

#Step3:构建工作流图

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

workflow=StateGraph(OrderState)

#添加节点

workflow.add_node("init",init_order)

workflow.add_node("risk",risk_assessment)

workflow.add_node("auto_approve",auto_approve)

workflow.add_node("human_review",human_review)

workflow.add_node("ship",ship_order)

workflow.add_node("complete",complete_order)

#设置入口

workflow.set_entry_point("init")

#添加边和条件边

workflow.add_edge("init","risk")

workflow.add_conditional_edges(

"risk",

route_after_risk,

{

"auto_approve":"auto_approve",

"human_review":"human_review"

}

)

workflow.add_edge("auto_approve","ship")

workflow.add_edge("human_review","ship")#审核通过后继续

workflow.add_edge("ship","complete")

workflow.add_edge("complete",END)

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

#Step4:编译并运行（含检查点持久化）

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

checkpointer=MemorySaver()#生产环境替换为Redis/PostgresSaver

app=pile(checkpointer=checkpointer)

#运行示例

config={"configurable":{"thread_id":"order-001"}}

#第一次运行——正常订单

foreventinapp.stream(

{"order_id":"ORD-NORMAL-001"},

config

):

node_name=list(event.keys())[0]

print(f"节点:{node_name}→{event[node_name].get('status','N/A')}")

#第二次运行——高风险订单（触发人工审核）

print("\n---高风险订单（触发人工审核）---")

foreventinapp.stream(

{"order_id":"ORD-HIGH-001"},

config

):

node_name=list(event.keys())[0]

if"messages"inevent[node_name]:

print(f"[{node_name}]{event[node_name]['messages']}")

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

#Step5:人工决策恢复执行（LangGraph支持断点续跑）

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

print("\n---人工决策后恢复执行---")

state=app.get_state(config)

print(f"当前状态:{state.values.get('status')}")

#注入人工决策并恢复

app.update_state(config,{"human_decision":"approved"})

foreventinapp.stream(None,config):

node_name=list(event.keys())[0]

print(f"恢复:{node_name}")动手练习练习：用LangGraph构建一个业务流程选择一个工作中的多步骤业务流程（如：请假审批、采购申请、合同审核等）使用LangGraph构建完整的工作流图，至少包含：1个条件分支（根据条件走向不同节点）1个人工审核节点状态持久化（使用MemorySaver）模拟正常流程和异常流程的执行评估标准：✓工作流图至少包含5个节点和1个条件边✓实现了人工审核节点（支持断点续跑）✓有完整的错误处理逻辑✓能展示状态恢复能力常见坑与避坑指南坑表现避坑方案过度使用条件边工作流图过于复杂，难以调试和维护简单场景用顺序边，复杂决策才用条件边。优先用子图（subgraph）封装复杂逻辑状态膨胀长时间运行后Agent上下文过长、token消耗巨大使用DeepAgents的DeltaChannels减少存储开销（可达100倍压缩）；及时清理过期状态忽视检查点恢复Agent中断后需要从头重新执行生产环境必须启用checkpointer，使用PostgresSaver替代MemorySaver硬编码模型直接写死模型名称，切换模型需改代码使用DeepAgents的HarnessProfiles实现模型无关配置商业变现建议企业级工作流自动化服务：用LangGraph为企业构建定制化业务自动化流程，按项目收费Agent运维服务：提供Agent系统的监控、状态恢复和性能优化运维服务，按月收费开源贡献+品牌建设：在LangChain生态中贡献工具/模板，提升个人品牌，吸引高端咨询客户模块四：LlamaIndex——AgenticRAG与企业级知识管理学习目标理解LlamaIndex从RAG框架到Agentic文档处理平台的进化掌握AgenticRAG的核心架构：检索→推理→行动→反思的循环学会使用LlamaIndex构建多Agent文档处理工作流掌握LlamaParse进行复杂文档（PDF/表格/图表）的结构化提取核心理论讲解1.LlamaIndex2026：不止是RAG框架LlamaIndex已从RAG框架转变为综合性Agentic文档处理平台，服务300k+用户、支持50+格式，服务企业包括OneCarlyle、CEMEX和KPMG。其核心哲学是“数据优先、检索为王”——所有组件围绕文档加载→处理→索引→检索→问答的全链路构建。2.AgenticRAGvs传统RAG维度传统RAGAgenticRAG检索方式单次检索→生成多轮检索→评估→再检索→生成工具使用仅向量检索检索+API调用+计算+推理记忆无状态跨会话状态保持输出质量依赖检索质量可自我纠错和补充检索3.LlamaIndex2026关键新能力LlamaAgentsBuilder：自然语言创建AgentDBOS持久工作流：Agent崩溃后自动恢复，无需手动编写检查点代码LlamaParse升级：混合文本提取+视觉模型，专攻复杂PDF解析（字形级提取+表格重建）企业级案例案例1：KPMG文档智能处理KPMG使用LlamaIndex构建多Agent文档处理工作流，实现审计文档的自动分类、关键信息提取和风险标注。案例2：制造业技术文档知识库某制造企业将技术文档转化为可查询的知识库，向量检索使文档查询响应时间从15分钟缩短至0.8秒。案例3：金融研报自动分析某券商使用AgenticRAG系统，自动抓取公司财报→提取关键指标→对比行业数据→生成分析摘要，研报生产效率提升10倍。代码示例：AgenticRAG系统"""

LlamaIndexAgenticRAG——多轮检索增强生成系统

支持文档索引、多步推理、自我纠错

"""

fromllama_index.coreimport(

VectorStoreIndex,SimpleDirectoryReader,Settings,

StorageContext,load_index_from_storage

)

fromllama_index.core.toolsimportQueryEngineTool,ToolMetadata

fromllama_index.core.agentimportReActAgent

fromllama_index.llms.openaiimportOpenAI

fromllama_index.embeddings.openaiimportOpenAIEmbedding

fromllama_index.core.node_parserimportSentenceSplitter

fromllama_index.core.memoryimportChatMemoryBuffer

importos

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

#Step1:配置全局设置

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

Settings.llm=OpenAI(model="gpt-5.4",temperature=0.1)

Settings.embed_model=OpenAIEmbedding(model="text-embedding-3-large")

Settings.node_parser=SentenceSplitter(

chunk_size=1024,

chunk_overlap=200,

paragraph_separator="\n\n"

)

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

#Step2:构建多知识库索引

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

defbuild_indices():

"""构建多源知识库的向量索引"""

#加载文档（支持PDF/Word/Markdown/CSV等50+格式）

financial_docs=SimpleDirectoryReader(

input_dir="./data/financial_reports",

required_exts=[".pdf",".docx",".md"]

).load_data()

legal_docs=SimpleDirectoryReader(

input_dir="./data/legal_docs"

).load_data()

#构建索引

financial_index=VectorStoreIndex.from_documents(financial_docs)

legal_index=VectorStoreIndex.from_documents(legal_docs)

#持久化

financial_index.storage_context.persist("./storage/financial")

legal_index.storage_context.persist("./storage/legal")

returnfinancial_index,legal_index

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

#Step3:创建查询工具（Agent可通过工具调用进行检索）

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

defcreate_query_tools(financial_index,legal_index):

"""将索引封装为Agent可调用的工具"""

financial_engine=financial_index.as_query_engine(

similarity_top_k=5,

response_mode="tree_summarize",#对多个文档块进行汇总

verbose=True

)

legal_engine=legal_index.as_query_engine(

similarity_top_k=3,

response_mode="compact"

)

tools=[

QueryEngineTool(

query_engine=financial_engine,

metadata=ToolMetadata(

name="financial_search",

description=(

"搜索财务报告和金融数据。"

"适用场景：查询公司财报、收入数据、资产负债表、现金流量等金融信息。"

"输入：与财务数据相关的自然语言查询。"

)

)

),

QueryEngineTool(

query_engine=legal_engine,

metadata=ToolMetadata(

name="legal_search",

description=(

"搜索法律文件和合规文档。"

"适用场景：查询法规条款、合同条款、合规要求等法律信息。"

"输入：与法律/合规相关的自然语言查询。"

)

)

),

]

returntools

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

#Step4:创建AgenticRAGAgent（ReAct模式）

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

defcreate_agentic_rag_agent(tools):

"""创建具有多步推理和自我纠错能力的RAGAgent"""

#长期记忆——保留最近20轮对话

memory=ChatMemoryBuffer.from_defaults(token_limit=8000)

#ReActAgent：Reasoning+Acting循环

agent=ReActAgent.from_tools(

tools=tools,

llm=Settings.llm,

memory=memory,

verbose=True,

max_iterations=10,#最多10轮推理-行动循环

system_prompt="""你是一个专业的文档分析Agent。

##工作流程

1.理解用户问题→判断需要查询哪些知识库

2.调用合适的检索工具→获取相关信息

3.分析检索结果→如果信息不足，重新检索或切换工具

4.综合多源信息→给出有据可查的回答

##核心规则

-**必须引用来源**：每个关键信息都要标注来自哪个工具

-**信息不足时主动说明**：不要说"可能"或"大概"

-**遇到矛盾信息时**：标注冲突并说明各来源的立场

-**数字要精确**：财务数据必须标注货币单位和时间点

"""

)

returnagent

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

#Step5:运行AgenticRAG

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

defrun_analysis(agent,query:str):

"""执行文档分析和问答"""

response=agent.chat(query)

print(f"📋问题:{query}")

print(f"🤖回答:{response.response}")

print(f"📊推理步骤:{len(response.sources)}步")

fori,stepinenumerate(response.sources):

print(f"步骤{i+1}:{step.content[:200]}...")

returnresponse

#主流程

if__name__=="__main__":

#构建索引（首次运行）或加载已有索引

financial_idx,legal_idx=build_indices()

#创建工具和Agent

tools=create_query_tools(financial_idx,legal_idx)

agent=create_agentic_rag_agent(tools)

#执行分析

queries=[

"根据最新财报，该公司营收增长率是多少？与行业平均水平相比如何？",

"该公司是否存在未披露的法律风险？请从合规文档中查找相关信息。",

"综合财务和法律角度，给出该公司的风险评估（低/中/高）并说明理由。",

]

forqinqueries:

run_analysis(agent,q)

print("\n"+"="*60+"\n")动手练习练习：构建你的私有知识库Agent收集10-20份与你工作相关的文档（PDF、Word、网页等）使用LlamaIndex构建索引，创建至少2个不同的查询引擎创建AgenticRAGAgent，测试多轮检索和跨文档综合分析对比“单次RAG”和“AgenticRAG”在复杂问题上的回答质量评估标准：✓成功构建至少2个知识库索引✓Agent能自主判断使用哪个工具进行检索✓复杂问题（需要跨文档综合）的回答准确率>80%常见坑与避坑指南坑表现避坑方案Chunk分割不当文档被切碎，检索时丢失上下文根据文档类型选择chunk策略：技术文档1024tokens，法律文档512tokens，并保持200tokens的overlapEmbedding模型与任务不匹配英文Embedding模型用于中文文档，检索精度差中文场景使用text-embedding-3-large或bge-large-zh检索结果不评估质量检索到的文档块与问题无关，但Agent仍强行生成回答在RAG循环中加入相关性评分，低于阈值的检索结果触发重新查询忽略PDF解析难度PDF中的表格、图表无法正确提取使用LlamaParse的混合文本+视觉模型方案处理复杂PDF商业变现建议企业知识库搭建服务：为企业搭建AgenticRAG知识库系统，项目制收费5-50万（取决于数据量和复杂度）文档智能处理SaaS：针对特定行业（法律/金融/医疗）的文档自动分类+提取+分析工具，按月订阅RAG性能优化顾问：帮助企业优化现有RAG系统的检索精度和响应速度，按优化效果收费模块五：AutoGen——多智能体协作系统的设计与落地学习目标理解多Agent协作的核心模式：对话驱动、角色分配、共识达成掌握AutoGen0.4的异步消息架构和模块化设计学会设计金融风控、创意协作、需求探索等多Agent场景掌握Agent群组的动态编排和任务路由核心理论讲解1.为什么需要多Agent协作？单一Agent在处理复杂任务时存在局限：缺乏多视角审视、无法并行处理子任务、错误难以自我发现。AutoGen的核心设计理念是：用“多个专业化Agent之间的结构化对话”替代“单一大模型的黑盒思考”。2.AutoGen0.4架构突破v0.4版本将任务调度吞吐量提升3倍，推理资源消耗降低20%。核心创新：异步消息机制：事件驱动架构，支持百万级QPS消息吞吐，系统吞吐量提升40%模块化设计：基础能力层→领域适配层→编排控制层，三层架构支持灵活扩展企业级增强：重试机制+结果缓存+熔断设计使系统可用性提升300%，Token消耗降低42%3.多Agent协作模式分类模式描述适用场景Leader-Follower一个主Agent分配任务，多个专业Agent执行项目管理、任务分解Peer-to-Peer平等Agent通过对话协商创意协作、需求讨论SequentialAgent按顺序执行，前一Agent的输出是后一Agent的输入数据处理PipelineCompetitive多个Agent独立完成同一任务，投票/评分选出最优代码审查、内容评审企业级案例案例1：金融风控多Agent系统某银行反欺诈系统：需求分析师Agent将需求转化为规范文档→代码生成器Agent实现Python模块→测试验证器Agent自动生成测试套件。反欺诈规则验证周期从72小时缩短至8小时，人工复核减少65%。案例2：电商运营多Agent协作秒针系统使用多Agent协作完成营销数据分析：数据抓取Agent→分析框架Agent→洞察提炼Agent→报告生成Agent，全链路自动化率90%。案例3：创意内容生产团队产品经理Agent→架构师Agent→安全专家Agent通过多轮对话达成共识，某零售企业从单一大模型切换到多Agent方案后，报告数据口径统一、关键指标完整性大幅提升。代码示例：AutoGen多Agent协作系统"""

AutoGen0.4多Agent协作系统——金融风控场景

演示：异步消息、角色分配、共识达成、人机协作

"""

importasyncio

fromautogen_agentchat.agentsimportAssistantAgent,UserProxyAgent

fromautogen_agentchat.teamsimportRoundRobinGroupChat

fromautogen_agentchat.conditionsimportTextMentionTermination

fromautogen_ext.models.openaiimportOpenAIChatCompletionClient

fromautogen_core.toolsimportFunctionTool

importjson

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

#Step1:定义各专业Agent的工具函数

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

asyncdeffetch_transaction_data(user_id:str,date_range:str)->str:

"""获取用户交易数据"""

#模拟从数据库获取

returnjson.dumps({

"user_id":user_id,

"transactions":[

{"id":"T001","amount":150.00,"merchant":"超市A","time":"2026-05-2510:30"},

{"id":"T002","amount":25000.00,"merchant":"境外商户X","time":"2026-05-2514:15"},

{"id":"T003","amount":28000.00,"merchant":"境外商户Y","time":"2026-05-2514:45"},

],

"total_count":3,

"total_amount":53150.00

})

asyncdefcheck_risk_rules(transaction:dict)->str:

"""检查单笔交易的风险规则"""

risk_flags=[]

#大额交易规则

iftransaction["amount"]>10000:

risk_flags.append(f"大额交易(>{10000})")

#境外交易规则

if"境外"intransaction["merchant"]:

risk_flags.append("境外商户")

#短时间多笔大额

risk_score=min(len(risk_flags)*0.35,1.0)

returnjson.dumps({

"transaction_id":t