30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度最近在跟进AI Agent技术落地时发现很多开发者对“Agentic AI”这个概念既兴奋又困惑。兴奋的是它代表了AI从被动问答走向主动规划和执行的新范式困惑的是如何从零开始构建一个真正能投入生产的智能体。恰好亚马逊云科技2026中国峰会的“Agentic AI 代码秀”提供了一个绝佳的实战观察窗口。本文将以一个技术实践者的视角深度拆解这场代码秀背后的技术逻辑、架构选型与工程实践并手把手带你复现一个具备“Agentic”能力的生产级智能体核心模块。无论你是想了解前沿趋势的架构师还是急需项目落地的全栈开发者都能从中获得可直接复用的代码与架构思路。1. 理解 Agentic AI从 RAG 到自主工作流在开始动手之前我们必须厘清一个核心概念什么是 Agentic AI它和普通的 RAG检索增强生成有什么区别普通 RAG的工作模式本质上是“增强的搜索引擎”。用户提问系统检索相关文档片段然后将问题和片段一起交给大模型生成答案。整个过程是单次、被动的。模型不决定下一步做什么也不管理任务状态。它的核心是“检索”与“生成”的静态结合。Agentic AI则引入了“智能体”的思维范式。一个 Agentic 系统通常包含以下核心组件规划器将复杂目标拆解为可执行的子任务序列。执行器调用各种工具Tool来完成具体任务如调用 API、查询数据库、运行代码。记忆体维护对话历史、任务上下文和工具执行结果用于后续决策。评估器对当前结果进行批判性评估决定是继续、重试还是调整策略。正如网络热词中提到的“rag-critic: leveraging automated critic-guided agentic workflow for retrieva”这揭示了一个关键模式引入一个自动化的“批评家”来引导工作流。在 Agentic 架构中这个“批评家”就是评估器它能够检查工具执行的结果是否满足要求并指导规划器进行下一步动作从而形成一个“感知-规划-执行-评估”的闭环。简单来说RAG 告诉你“答案是什么”而 Agentic AI 能帮你“把事做成”。例如一个普通RAG客服机器人只能回答“退货政策是什么”而一个Agentic客服机器人可以主动帮你完成“提交退货申请、通知仓库、更新订单状态”等一系列操作。2. 环境准备与核心工具栈为了复现代码秀中演示的智能体构建流程我们需要搭建一个贴近生产环境的开发栈。本次实战将基于 Python 生态并优先选用在峰会中被多次提及的、成熟稳定的开源框架。2.1 基础环境与 Python 版本操作系统推荐 Ubuntu 20.04/22.04 LTS 或 macOSWindows 用户可使用 WSL2 以获得最佳体验。Python 版本3.10 或 3.11。这是目前主流AI框架兼容性最好的版本。包管理工具使用pip和venv创建独立的虚拟环境避免依赖冲突。# 创建并激活虚拟环境 python -m venv agentic_env source agentic_env/bin/activate # Linux/macOS # agentic_env\Scripts\activate # Windows # 升级pip pip install --upgrade pip2.2 核心框架与库我们将使用LangChain作为智能体编排框架它提供了丰富的工具集成和智能体模板。同时为了构建“批评家”引导的工作流我们会引入LangGraph它擅长描述有状态、多步骤的工作流。# 安装核心框架 pip install langchain langchain-community langgraph # 安装大模型接口。这里以 OpenAI 兼容的 API 为例如 Azure OpenAI, DeepSeek等 pip install openai # 安装用于工具调用的必要库 pip install requests python-dotenv2.3 辅助工具与可观测性在峰会“分论坛 3数据团队”中“Agent 黑盒拆解术基于 Langfuse 的 Trace、Token、Tool Call 可观测”被重点强调。对于生产级智能体可观测性至关重要。我们将使用 Langfuse 来追踪和调试智能体的内部决策过程。# 安装 Langfuse 用于链路追踪和监控 pip install langfuse2.4 项目结构初始化创建一个清晰的项目结构有助于管理复杂的智能体工作流。agentic_project/ ├── .env # 存储API密钥等敏感配置 ├── requirements.txt # 项目依赖 ├── main.py # 主入口文件 ├── agents/ # 智能体模块 │ ├── __init__.py │ ├── planner_agent.py # 规划器 │ ├── critic_agent.py # 批评家/评估器 │ └── orchestration.py # LangGraph 工作流编排 ├── tools/ # 自定义工具集 │ ├── __init__.py │ ├── web_search.py # 网络搜索工具 │ └── calculator.py # 计算工具 └── memory/ # 记忆管理 ├── __init__.py └── conversation_memory.py在.env文件中配置你的密钥# .env OPENAI_API_KEYyour_openai_api_key_here LANGFUSE_PUBLIC_KEYyour_langfuse_public_key LANGFUSE_SECRET_KEYyour_langfuse_secret_key LANGFUSE_HOSThttps://cloud.langfuse.com3. 构建核心组件工具、智能体与记忆接下来我们按照 Agentic 架构自底向上构建核心组件。3.1 创建可复用的工具工具是智能体与世界交互的“手”和“脚”。我们创建两个基础工具一个用于计算一个用于模拟网络搜索。# tools/calculator.py from langchain.tools import tool import math tool def calculator(expression: str) - str: 执行数学计算。支持加减乘除 - * /和乘方**。 例如calculator(\3 5 * 2\) 返回 13.0。 try: # 警告使用eval存在安全风险仅用于演示。生产环境应使用安全表达式解析器如 ast.literal_eval 限制操作。 result eval(expression, {__builtins__: None}, {math: math}) return f计算结果: {result} except Exception as e: return f计算错误: {e} # tools/web_search.py from langchain.tools import tool import requests import json tool def web_search(query: str) - str: 模拟网络搜索返回与查询相关的信息摘要。 在实际应用中应替换为真实的搜索引擎API如Serper、Google Custom Search。 # 这里用一个模拟的响应来演示工具调用流程 mock_responses { 亚马逊云科技峰会: 亚马逊云科技中国峰会于2026年6月举办设有Agentic AI代码秀、多场分论坛涵盖Agent构建、数据团队、模型实践等主题。, Agentic AI 趋势: Agentic AI 强调智能体的自主规划、工具调用和迭代执行能力是当前AI应用从问答走向自动化的重要方向。, Python 最新版本: Python 3.11 在性能上有显著提升是构建AI应用的推荐版本。 } for key, value in mock_responses.items(): if key.lower() in query.lower(): return f搜索到相关信息: {value} return f未找到与 {query} 直接相关的模拟信息。建议尝试其他关键词。3.2 构建规划器与批评家智能体我们使用 LangChain 的create_react_agent模式来构建一个能使用工具的规划器。同时单独构建一个批评家智能体来评估结果。# agents/planner_agent.py from langchain.agents import create_react_agent, AgentExecutor from langchain.prompts import PromptTemplate from langchain_openai import ChatOpenAI import os from dotenv import load_dotenv load_dotenv() def create_planner_agent(tools): 创建一个ReAct模式的规划器智能体。 ReAct (Reason Act) 模式让智能体能够思考后选择工具。 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, openai_api_keyos.getenv(OPENAI_API_KEY)) # ReAct 提示词模板 prompt PromptTemplate.from_template( 你是一个任务规划专家。请根据用户的问题逐步思考并选择合适的工具来解决问题。 你可以使用的工具如下 {tools} 请严格按照以下格式回应 思考首先分析问题决定需要做什么。 行动选择要使用的工具输入必须是有效的JSON字符串格式为{{action: 工具名, action_input: 工具输入}} 观察工具返回的结果。 ... (这个思考/行动/观察循环可以重复多次) 最终答案当问题被解决或无法进一步处理时汇总所有观察结果给出清晰、完整的最终答案。 现在开始 问题{input} {agent_scratchpad} ) agent create_react_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) return agent_executor # agents/critic_agent.py from langchain.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI import os def create_critic_agent(): 创建一个批评家智能体用于评估任务执行结果的质量和完整性。 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, openai_api_keyos.getenv(OPENAI_API_KEY)) system_prompt 你是一个严格的质量评估专家批评家。你的任务是对智能体给出的答案进行评估。 请从以下维度进行评估 1. **准确性**答案中的事实信息是否准确是否与工具返回的观察结果一致 2. **完整性**是否完全回答了原始问题有没有遗漏关键点 3. **清晰度**答案是否组织良好、易于理解 你的输出必须是严格的JSON格式 {{ \assessment\: \通过\ 或 \不通过\, \score\: 0-100的整数, \reason\: \详细的评估理由如果‘不通过’必须明确指出缺失或错误的部分。\, \suggestion\: \如果‘不通过’给出具体的改进建议或下一步行动指示。\ }} critic_prompt ChatPromptTemplate.from_messages([ (system, system_prompt), (human, 原始问题{question}\n智能体给出的答案{agent_answer}\n请进行评估。) ]) critic_chain critic_prompt | llm return critic_chain3.3 实现工作流记忆智能体需要记住对话历史和工具执行结果。我们使用 LangChain 的ConversationBufferMemory。# memory/conversation_memory.py from langchain.memory import ConversationBufferMemory def get_conversation_memory(): 获取一个对话记忆实例。 memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue) return memory4. 使用 LangGraph 编排 Agentic 工作流这是实现“批评家引导的工作流”的关键。我们将使用 LangGraph 来定义智能体的状态和决策流程。# agents/orchestration.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, AIMessage import json from .planner_agent import create_planner_agent from .critic_agent import create_critic_agent from ..tools.calculator import calculator from ..tools.web_search import web_search # 1. 定义工作流状态 class AgentState(TypedDict): 定义图工作流的状态结构。 question: str # 用户原始问题 agent_answer: str # 规划器智能体产生的答案 critique_result: dict # 批评家的评估结果 messages: Annotated[List, operator.add] # 对话消息历史 iteration_count: int # 循环迭代次数防止无限循环 # 2. 初始化工具和智能体 tools [calculator, web_search] planner_agent_executor create_planner_agent(tools) critic_agent create_critic_agent() # 3. 定义各个节点函数 def call_planner(state: AgentState): 节点调用规划器智能体回答问题。 print(f\n--- 第 {state[iteration_count]1} 轮规划 ---) question state[question] # 调用规划器执行任务 result planner_agent_executor.invoke({input: question}) agent_answer result.get(output, 智能体未产生输出。) # 更新状态 new_state { agent_answer: agent_answer, iteration_count: state[iteration_count] 1, messages: state[messages] [AIMessage(contentagent_answer)] } return new_state def call_critic(state: AgentState): 节点调用批评家评估答案。 print(f\n--- 第 {state[iteration_count]} 轮批评 ---) question state[question] agent_answer state[agent_answer] # 调用批评家链 critique_response critic_agent.invoke({question: question, agent_answer: agent_answer}) # 解析JSON响应 try: critique_result json.loads(critique_response.content) except json.JSONDecodeError: critique_result {assessment: 不通过, score: 0, reason: 批评家返回了无效的JSON格式。, suggestion: 检查批评家提示词和模型输出。} print(f批评家评估: {critique_result}) return {critique_result: critique_result} def decide_next_step(state: AgentState): 条件边根据批评家的评估决定下一步。 如果评估‘通过’或迭代超过3次则结束。否则返回‘revise’让规划器重试。 critique state[critique_result] max_iterations 3 if critique.get(assessment) 通过 or state[iteration_count] max_iterations: print(f\n--- 工作流结束 ---) print(f最终答案: {state[agent_answer]}) return end else: print(f\n--- 评估不通过开始第 {state[iteration_count]1} 轮修订 ---) # 这里可以将批评家的建议整合成一个新的、更明确的问题再次交给规划器。 # 简化处理我们将原始问题与批评家建议拼接作为新的输入。 suggestion critique.get(suggestion, ) revised_question f{state[question]} 之前的答案不完整/不准确。请根据以下反馈改进{suggestion} return {question: revised_question, agent_answer: , critique_result: {}} # 4. 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(planner, call_planner) workflow.add_node(critic, call_critic) # 设置入口点 workflow.set_entry_point(planner) # 添加边规划器 - 批评家 workflow.add_edge(planner, critic) # 添加条件边批评家 - 决策 workflow.add_conditional_edges( critic, decide_next_step, { end: END, # 结束工作流 revise: planner # 返回规划器节点进行修订 } ) # 编译图 agentic_app workflow.compile() # 5. 运行工作流的函数 def run_agentic_workflow(question: str): 运行完整的Agentic工作流。 initial_state { question: question, agent_answer: , critique_result: {}, messages: [HumanMessage(contentquestion)], iteration_count: 0 } print(f开始处理问题: {question}) final_state None # 以流式方式运行可以看到每一步的输出 for event in agentic_app.stream(initial_state): for node_name, node_output in event.items(): print(f节点 [{node_name}] 执行完毕) final_state node_output if node_output else final_state return final_state5. 集成可观测性使用 Langfuse 追踪一切为了让这个智能体黑盒变得透明我们集成 Langfuse。它能记录每一次LLM调用、工具调用和工作流状态是调试和优化不可或缺的工具。# main.py from langfuse import Langfuse from langfuse.callback import CallbackHandler import os from dotenv import load_dotenv from agents.orchestration import run_agentic_workflow load_dotenv() # 初始化 Langfuse langfuse Langfuse( public_keyos.getenv(LANGFUSE_PUBLIC_KEY), secret_keyos.getenv(LANGFUSE_SECRET_KEY), hostos.getenv(LANGFUSE_HOST) ) # 创建 Langfuse 回调处理器用于自动追踪 langfuse_handler CallbackHandler() def main_with_tracing(): 主函数运行智能体并启用追踪。 questions [ 计算一下 (15 7) * 3 等于多少, 告诉我亚马逊云科技2026峰会上关于Agentic AI代码秀的重点内容。, # 可以尝试更复杂的问题如“先搜索‘Python 3.11的新特性’然后告诉我它对于AI应用开发最主要的两个优势是什么” ] for question in questions: print(f\n{*60}) print(f处理问题: {question}) print(*60) # 使用 Langfuse 创建一个独立的追踪会话 trace langfuse.trace(nameAgentic Workflow Execution, inputquestion) try: # 运行工作流并将回调处理器传入这里需要稍微调整工作流以支持回调为简化示例我们直接运行 # 在实际中需要将 handler 传递给 LangChain 或 LangGraph 的执行器 final_state run_agentic_workflow(question) # 记录最终输出到追踪 trace.output final_state.get(agent_answer, No answer generated.) trace.update() # 更新追踪记录 except Exception as e: trace.output fError: {str(e)} trace.update() print(f工作流执行出错: {e}) print(f\n追踪已生成可在 Langfuse 控制台查看详情。Trace ID: {trace.id}) if __name__ __main__: main_with_tracing()运行python main.py你不仅会在控制台看到智能体“思考-行动-观察-被批评-再思考”的完整过程还能在 Langfuse 的Web控制台中看到一个可视化的执行轨迹图清晰地展示每个节点的耗时、Token 消耗和输入输出。6. 生产级考量与最佳实践构建一个玩具智能体和工作流相对简单但要将其投入生产必须考虑以下在2026峰会各分论坛中被反复强调的工程问题。6.1 应对 Token 爆炸与成本控制在“分论坛 3数据团队”中“取之有度用之有节-破解 Agentic 应用 Token 爆炸难题”是核心议题。Agentic 工作流因多轮LLM调用极易导致Token消耗剧增。最佳实践记忆压缩不要将全部对话历史都塞进上下文。使用ConversationSummaryMemory或ConversationBufferWindowMemory来限制历史长度。工具设计精简工具的描述应简洁准确避免冗长。工具返回的结果也应进行摘要或过滤只保留关键信息。流式中断为工作流设置最大迭代次数如我们代码中的max_iterations3和超时时间。模型分级让“批评家”这类对创造力要求不高的环节使用更便宜、更快的模型如 GPT-3.5-Turbo而让“规划器”使用能力更强的模型。6.2 确保可信查询与数据安全“构建 Agentic 的可信查询链路”是另一个关键话题。智能体自动调用工具可能带来数据泄露、未授权访问等风险。最佳实践工具权限沙箱为每个工具定义清晰的权限边界。例如数据库查询工具只能访问特定的只读视图。输入验证与清理对所有从用户输入传递到工具的参数进行严格的验证和清理防止注入攻击。审计日志记录智能体做出的每一个决策、调用的每一个工具及其参数和结果便于事后审计和复盘。人工审核环对于高风险操作如发送邮件、修改数据库、支付在工作流中设计“人工批准”节点。6.3 架构可扩展性与团队协作从“分论坛 2Agent 构建者”的案例如安克创新、CoCo零代码平台可以看出企业级应用需要平台化支持。最佳实践微服务化工具将工具封装成独立的API服务如gRPC或RESTful便于管理、升级和扩展。统一智能体网关构建一个统一的网关来路由请求、管理认证、限流和监控类似于韶音科技分享的“统一模型网关”理念。配置化与低代码将智能体的流程Planning、Critic逻辑、工具集通过配置文件或可视化界面进行管理降低开发门槛让业务专家也能参与构建。6.4 监控、调试与持续改进“Agent 黑盒拆解术”强调了可观测性的重要性。没有监控的智能体上线等同于盲人摸象。最佳实践全链路追踪必须集成像 Langfuse、Arize、Weights Biases 这样的工具追踪每一次LLM调用、工具调用、工作流状态。关键指标监控定义并监控业务指标任务成功率、用户满意度和技术指标延迟、Token消耗、工具调用错误率。A/B测试与冠军/挑战者模式对于智能体的策略、提示词Prompt采用A/B测试来评估哪种效果更好实现数据驱动的迭代优化。7. 从演示到部署云原生实践峰会中多次提到利用Amazon EKS和Kata Container构建通用 AI Agents 平台。这指向了云原生部署的最佳路径。部署建议容器化将你的智能体应用、工具服务分别打包成 Docker 镜像。Kubernetes 编排使用 Amazon EKS 来管理容器编排实现自动扩缩容、滚动更新和高可用。安全隔离考虑使用 Kata Containers 或 gVisor 等安全容器运行时为每个智能体执行环境提供更强的隔离这在多租户场景下尤为重要。服务网格使用 Istio 或 AWS App Mesh 来管理服务间通信、流量治理和可观测性。CI/CD 流水线建立自动化的代码构建、测试、镜像打包和部署流水线确保快速、安全地迭代你的 Agentic 应用。通过以上七个部分的拆解我们从概念理解、环境搭建、核心编码、工作流编排、可观测性集成一直讨论到生产级的最佳实践和部署方案完整复现了“代码秀”中构建生产级Agent的核心技术路径。这套架构不仅是一个演示更是一个坚实的、可扩展的工程起点。接下来你可以在此基础上接入真实的数据库、API设计更复杂的业务工作流让AI智能体真正成为你业务中的“超级搭档”。 30款热门AI模型一站整合DeepSeek/GLM/Qwen 随心用限时 5 折。 点击领海量免费额度