进阶示例:构建全能出行助手(集成模型/记忆/沙箱/网关/高德mcp)
场景概述
在基础示例中,创建了简单的对话智能体,并部署托管到云上的AgentArts环境中。但是在真实的生产环境中,企业级的智能体需要具备持久化记忆、安全沙箱以及无缝对接外部系统的能力。
本示例中将构建一个“全能出行助手”智能体,并重点展示以下能力:
- 记忆库(Memory):接入云端长短期记忆,实现跨会话的上下文持久化。
- 网关(Gateway):通过标准的MCP协议转换能力,接入外部工具。
- 沙箱工具(Sandbox):在平台提供的完全隔离的容器中安全执行Python代码。
- 身份认证(Identity):结合MCP网关,实现工具的标准化鉴权。
环境准备
- 操作系统:Linux ARM64及X86,服务器可访问公网。
- 安装Python:请确保Python 3.10及以上版本已安装。
大多数Linux发行版(如Ubuntu)都预装了Python,您可以先通过python3 --version检查。如未安装,可以使用如下命令安装:
sudo apt update sudo apt install python3
- 安装Docker:请确保Docker 18.06及以上版本已安装。如未安装,可以使用如下命令安装:
# 查询 Docker 版本 docker --version # 安装Docker sudo apt update sudo apt install docker.io
华为云SWR基础版不支持OCI镜像格式,如果您使用的是Docker 27及以上版本,并且需要处理OCI镜像,可以通过设置环境变量来关闭OCI支持。
export DOCKER_BUILDKIT=0 # 或者 export BUILDKIT_USE_OCI_MEDIA_TYPES=0
- 执行以下命令安装SDK(建议在Python虚拟环境中安装,以避免与系统包产生冲突)。
# 创建并激活虚拟环境 (linux) python3 -m venv venv source venv/bin/activate # 安装sdk pip install agentarts-sdk
- 执行以下命令配置华为云凭证,获取华为云凭证请参考认证鉴权。
export HUAWEICLOUD_SDK_AK="your-access-key" export HUAWEICLOUD_SDK_SK="your-secret-key"
核心代码实现
- 在AgentArts平台创建记忆库。
- 登录AgentArts智能体平台。
- 在左侧导航栏选择“开发中心 > 组件库 ”,进入“记忆库”界面。
- 单击“创建记忆库”。填写记忆库名称、描述;勾选“私网访问”、“公网访问”;勾选全部的“长期记忆提取策略”,其余参数使用默认配置。 图1 创建记忆库
- 记忆库创建完成后,单击记忆库名称,进入“基本信息”界面,获取记忆库ID。 图2 获取记忆库ID
- 在“API Key”处单击“更新”。 图3 更新API Key
- 弹框中单击“确定”后,复制或下载API Key。 图4 复制或下载API Key
- 在AgentArt平台创建网关,配置高德MCP服务。
- 高德MCP如下,使用前,请参考高德开放平台文档,创建获取API Key。
{ "mcpServers": { "amap-maps-streamableHTTP": { "url": "https://mcp.amap.com/mcp?key=您在高德官网上申请的key" } } } - 在AgentArts平台左侧导航栏中选择“开发中心 > 组件库”,并进入“网关”页面,单击“创建网关”。
- 填写网关基础信息并配置权限与身份认证。参考下表进行配置。
表1 网关基础信息与权限身份认证配置 参数
说明
名称
可自定义。
描述(可选)
网关的描述信息,可自定义。
委托
使用平台的默认值。
入站身份认证
选择API Key认证。
API Key名称
可自定义。
日志记录
选择开启。
图5 填写网关基础信息与权限身份认证
- 单击“创建Target”,配置高德MCP服务。
表2 高德Target配置 参数
说明
名称
可自定义。
描述(可选)
可自定义。
类型
选择“MCP”,使用现成的高德MCP服务,不需要使用直接对接外部REST API的方式。
传输方式
选择“Streamable HTTP”,与高德MCP的调用方式保持一致。
MCP地址
填写:https://mcp.amap.com/mcp
出站身份认证
选择“API Key”,单击“创建出站身份”,参数配置如下:
- 身份名称:可自定义。
- 认证类型:选择“API Key”。
- API Key的值:填写高德开发平台中创建的API Key。
出站身份创建完成后,返回“创建Target”页面,选择已创建的出站身份。
- 位置:选择“查询参数”。
- 参数名称:填写为key。
- 前缀:不填。
图6 创建出站身份
图7 配置出站身份认证
- 配置完成后,单击“确定”创建Target。Target创建完成后,回到“创建网关”页面,高级配置选择“公网访问”后,单击“创建网关”。
- 网关创建完成后,回到网关列表页面。单击网关名称,记录网关URL及网关的API Key(注意是网关的API Key,不是高德mcp的key)。 图8 网关列表
- 在网关详情页面获取网关URL。 图9 获取网关URL
- 在网关详情页面,单击URN链接。进入Agent Identity页面,获取网关的API Key。 图10 单击URN链接
图11 获取网关API Key
- 高德MCP如下,使用前,请参考高德开放平台文档,创建获取API Key。
- 执行如下命令安装langchain、langgraph、langchain-openai。
pip install -U langchain langgraph langchain-openai
- 在本地创建agent.py文件,编辑代码如下。
示例代码中创建了出行助手智能体,并对接记忆库、网关、沙箱、身份认证、华为云MaaS服务提供的模型。
""" ================================================================================ AgentArts LangGraph 智能体核心实现 ================================================================================ """ import os import uuid import json import requests import urllib3 import re from typing import List, Optional, Dict, Any, TypedDict, Annotated from pathlib import Path from dotenv import load_dotenv # 禁用内网自签名证书的 SSL 警告 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # ================================================================================ # 基础依赖:LangGraph + LangChain # ================================================================================ from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolNode from langgraph.graph import MessagesState from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage, AIMessage, ToolMessage from langchain_core.tools import tool from langchain_openai import ChatOpenAI # 引入 AgentArts 平台级能力 from agentarts.sdk.integration.langgraph import AgentArtsMemorySessionSaver try: from agentarts.sdk.memory import MemoryClient except ImportError: MemoryClient = None print("[Warning] 未检测到 agentarts.sdk,将跳过云端记忆同步。") # 从当前目录的 .env 文件加载环境变量 load_dotenv() def clean_messages_for_llm(messages: List[BaseMessage]) -> List[BaseMessage]: """ 清洗消息队列,防止因历史记录截断或脏数据导致 ModelArts 报 81001 校验错误。 原理: 当记忆库截断(只读取最近10条)时,可能把上一步的工具调用请求(AIMessage)丢弃, 只留下了工具结果(ToolMessage),导致第一条消息不合规。 本函数会自动检测并过滤掉所有无对应大模型请求的孤立 ToolMessage。 """ cleaned = [] for msg in messages: if isinstance(msg, ToolMessage): # 只有前一个消息是 AIMessage 并且含有 tool_calls 时,这个 ToolMessage 才是合法的 if cleaned and isinstance(cleaned[-1], AIMessage) and getattr(cleaned[-1], "tool_calls", None): cleaned.append(msg) else: # 发现由于截断产生的 ToolMessage,直接过滤掉,防止 81001 错误。 print(f"[MaaS Guard] 检测到孤立的 ToolMessage,已自动过滤。") else: cleaned.append(msg) return cleaned # ================================================================================ # 第一部分:配置模块 # ================================================================================ class ModelConfig: """模型配置类""" def __init__(self): self.name = os.getenv("MODEL_NAME", "deepseek-v3.2") self.url = os.getenv("MODEL_URL", "https://api.modelarts-maas.com/openai/v1") self.api_key = os.getenv("MODEL_API_KEY") if not self.api_key: raise ValueError("[错误] 未找到 MODEL_API_KEY!请在 .env 文件中进行配置。") self.model_type = os.getenv("MODEL_TYPE", "openai") class AgentConfig: """Agent全局配置类,整合模型与平台组件配置""" def __init__(self): self.model = ModelConfig() # 记忆空间和区域配置(去除中文默认值) self.memory_space = os.getenv("MEMORY_SPACE_ID") if not self.memory_space: print("[Warning] 未配置 MEMORY_SPACE_ID,云端记忆库可能无法正常工作。") self.region = os.getenv("HUAWEICLOUD_SDK_REGION", "cn-southwest-2") # 全局递归控制(防止大模型无限调用工具陷入死循环) self.max_iterations = int(os.getenv("MAX_ITERATIONS", "50")) self.session_ttl = int(os.getenv("SESSION_TTL", "3600")) config = AgentConfig() # ================================================================================ # 第二部分:平台级记忆系统 (AgentArts Checkpoint) # ================================================================================ _global_checkpointer = None def get_checkpointer() -> AgentArtsMemorySessionSaver: """获取平台级 Checkpointer 实例""" global _global_checkpointer if _global_checkpointer is None and config.memory_space: _global_checkpointer = AgentArtsMemorySessionSaver( space_id=config.memory_space, region=config.region ) return _global_checkpointer def get_thread_config(session_id: str) -> Dict[str, Any]: return {"configurable": {"thread_id": session_id}} # ================================================================================ # 第三部分:工具定义 (MCP网关 + 安全沙箱) # ================================================================================ def _call_mcp_gateway(tool_name: str, arguments: Dict[str, Any]) -> str: """发起实际的 MCP 网关请求""" gateway_url = os.getenv("GATEWAY_ENDPOINT") if not gateway_url: return "网关调用失败: 未配置 GATEWAY_ENDPOINT" headers = { "Content-Type": "application/json", "Accept": "application/json, text/event-stream", "mcp-session-id": str(uuid.uuid4()), "Authorization": f"Bearer {os.getenv('GATEWAY_INBOUND_TOKEN', '')}" } payload = { "jsonrpc": "2.0", "id": "call-tool", "method": "tools/call", "params": {"name": tool_name, "arguments": arguments} } try: response = requests.post(gateway_url, headers=headers, json=payload, verify=False, timeout=30) raw_text = response.text if "data:" in raw_text: match = re.search(r'\{.*\}', raw_text, re.DOTALL) if match: parsed_json = json.loads(match.group()) return json.dumps(parsed_json.get("result", parsed_json), ensure_ascii=False) return json.dumps(response.json().get("result", response.json()), ensure_ascii=False) except Exception as e: return f"网关调用异常: {str(e)}" @tool def gaode_get_weather(city: str) -> str: """查询指定城市的天气。Args: city: 城市名称或标准adcode""" print(f"\n[MCP Gateway] 查询天气: {city}") return _call_mcp_gateway("gaodemap_maps_weather", {"city": city}) @tool def gaode_get_location(address: str) -> str: """将详细的地址转换为经纬度或位置基本信息。Args: address: 地址""" print(f"\n[MCP Gateway] 获取地理位置: {address}") return _call_mcp_gateway("gaodemap_maps_geo", {"address": address}) @tool def execute_python(code: str) -> str: """在平台沙箱环境中安全执行 Python 代码。Args: code: 待执行的Python代码""" from agentarts.sdk.tools import code_session print(f"\n[Sandbox] 正在安全沙箱中执行代码...") sandbox_name = os.getenv("SANDBOX_NAME") if not sandbox_name: return "沙箱调用失败: 未在环境变量中配置 SANDBOX_NAME" try: with code_session(config.region, sandbox_name) as client: res = client.invoke(operate_type="execute_code", arguments={"code": code, "language": "python"}) return json.dumps(res.get("result")) except Exception as e: return f"沙箱执行失败: {str(e)}" def get_all_tools(): return [gaode_get_weather, gaode_get_location, execute_python] # ================================================================================ # 第四部分:LangGraph 工作流构建 # ================================================================================ class AgentState(MessagesState): iteration: int def should_continue(state: AgentState) -> str: """条件边判断:如果 LLM 返回了工具调用,则继续执行 tools 节点""" last_message = state["messages"][-1] if hasattr(last_message, "tool_calls") and last_message.tool_calls: return "continue" return "end" class TravelAgent: def __init__(self): self._llm = ChatOpenAI( model=config.model.name, base_url=config.model.url, api_key=config.model.api_key, temperature=0.1, # 较低的温度值保障工具调用的 JSON 格式稳定 ) self._checkpointer = get_checkpointer() self.graph = self._build_graph() self.system_prompt = "你是一个智能体助手,可以调用高德mcp和代码沙箱" def _build_graph(self): def model_node(state: AgentState) -> AgentState: llm_with_tools = self._llm.bind_tools(get_all_tools()) # 在将历史消息发送给大模型接口前,进行安全性清洗 # 过滤掉由于记忆截断产生的孤立 ToolMessage,防止触发 ModelArts 81001 错误 safe_messages = clean_messages_for_llm(state["messages"]) response = llm_with_tools.invoke(safe_messages) return {"messages": [response], "iteration": state.get("iteration", 0) + 1} workflow = StateGraph(AgentState) workflow.add_node("agent", model_node) workflow.add_node("tools", ToolNode(get_all_tools())) workflow.set_entry_point("agent") workflow.add_conditional_edges("agent", should_continue, {"continue": "tools", "end": END}) workflow.add_edge("tools", "agent") return workflow.compile(checkpointer=self._checkpointer) def run(self, user_input: str, session_id: str) -> str: """主运行方法,集成记忆与工具循环""" if MemoryClient is not None and config.memory_space: try: MemoryClient().create_memory_session( space_id=config.memory_space, id=session_id, actor_id="user_123" ) except Exception as e: if "Session id already exists" not in str(e): print(f"[Warning] 云端记忆会话同步失败: {e}") thread_config = {"configurable": {"thread_id": session_id}} current_state = self.graph.get_state(thread_config) # 构造消息历史 if not current_state.values.get("messages"): messages = [SystemMessage(content=self.system_prompt), HumanMessage(content=user_input)] else: messages = [HumanMessage(content=user_input)] # 执行并限制最大迭代次数 result = self.graph.invoke( {"messages": messages}, config=thread_config, recursion_limit=config.max_iterations ) return result["messages"][-1].content @property def session_id(self) -> Optional[str]: return self._last_session_id - 创建.env环境变量配置文件。
# ========================================== # 1. 模型配置 (对接华为云MaaS模型) # ========================================== MODEL_NAME=deepseek-v3.2 MODEL_URL=https://api.modelarts-maas.com/openai/v1 MODEL_API_KEY=替换为真实的模型API Key MODEL_TYPE=openai # ========================================== # 2. AgentArts 记忆库配置 # ========================================== MEMORY_SPACE_ID=替换为记忆库 ID HUAWEICLOUD_SDK_MEMORY_API_KEY=替换为记忆库的API Kry HUAWEICLOUD_SDK_REGION=cn-southwest-2 # ========================================== # 3. 网关配置 # ========================================== # 网关地址 GATEWAY_ENDPOINT=替换为网关的URL # 网关的 apikey GATEWAY_INBOUND_TOKEN=替换为网关的API Key # ========================================== # 4. Agent 全局配置 # ========================================== MAX_ITERATIONS=50 SESSION_TTL=3600 # 本地测试启动 HTTP Server 时绑定的端口 AGENT_RUN_PORT=8080
获取模型API Key的方法请参考获取华为云MaaS服务模型API Key。
agent.py文件的示例中,使用华为云MaaS服务提供的deepseek-v3.2模型,如果使用其他模型注意MODEL_NAME的值,需填写为模型接口中model参数的值,可以参考如下方式获取。注意需要选择OpenAI兼容接口。
图12 获取OpenAI兼容接口的model值
- 创建main.py文件,用于本地运行智能体,并进行测试。
测试脚本中设计了4个对话,用于测试agent.py中集成的基础对话、MCP工具调用、沙箱工具,记忆恢复能力。
""" my-agent 本地多轮对话与工具测试脚本 """ import uuid from agent import TravelAgent def main(): print("正在初始化智能体助手 (集成 MCP 网关、安全沙箱与云端记忆)...") try: agent = TravelAgent() except Exception as e: print(f"\n[初始化失败] 请检查 .env 文件配置: {e}") return # 生成符合华为云记忆库校验规则的、标准格式的 UUID 字符串作为会话 ID session_id = str(uuid.uuid4()) print(f"当前分配的会话 ID (标准 UUID): {session_id}\n") # ===================================================================== # 第一轮测试:基础人设与大模型连通性 # 目的:验证模型能一句话完成符合人设的自我介绍 # ===================================================================== print("=" * 60) prompt1 = "你好,请用一句话做个自我介绍。" print(f"用户: {prompt1}") response1 = agent.run(prompt1, session_id=session_id) print(f"\nAgent 回复: \n{response1}") # ===================================================================== # 第二轮测试:高德 MCP 工具调用 (天气查询) # 目的:验证模型能否在公有云网关下,成功调用天气接口并返回真实天气。 # ===================================================================== print("\n" + "=" * 60) prompt2 = "帮我查一下现在杭州的天气怎么样?" print(f"用户: {prompt2}") response2 = agent.run(prompt2, session_id=session_id) print(f"\nAgent 回复: \n{response2}") # ===================================================================== # 第三轮测试:安全沙箱工具调用 (Python执行) # 目的:验证大模型能否利用 Python 代码沙箱安全计算复杂数理问题。 # ===================================================================== print("\n" + "=" * 60) prompt3 = "请用 Python 帮我计算一下 12345 乘以 6789 的结果是多少?" print(f"用户: {prompt3}") response3 = agent.run(prompt3, session_id=session_id) print(f"\nAgent 回复: \n{response3}") # ===================================================================== # 第四轮测试:长短期记忆校验 # 目的:测试在经历了复杂的工具调用后,云端记忆库的跨轮次记忆提取能力是否正常。 # ===================================================================== print("\n" + "=" * 60) prompt4 = "考考你的记性,我刚才让你查的是哪个城市的天气?" print(f"用户: {prompt4}") response4 = agent.run(prompt4, session_id=session_id) print(f"\nAgent 回复: \n{response4}") print("=" * 60) if __name__ == "__main__": main() - 将agent.py、.env、main.py上传至服务器中。
如果您使用MobaXterm、Xshell这类客户端软件,可以直接在软件界面上拖拽上传。

- 执行python main.py命令对智能体进行本地测试。
由于在main.py文件中已经设置好了问题,智能体本地运行时会调用大模型进行回复。有正常的回复即表示本地测试成功。

- 创建app.py文件,并上传至服务器中。该文件会将本地的Agent逻辑封装为符合AgentArts平台规范的Web服务。
import os from typing import Dict, Any from agentarts.sdk import AgentArtsRuntimeApp, RequestContext # 导入出行助手智能体 TravelAgent from agent import TravelAgent # 实例化平台运行时应用 app = AgentArtsRuntimeApp() try: # 实例化高级智能体 TravelAgent my_agent = TravelAgent() except Exception as e: print(f"Agent 初始化失败: {e}") my_agent = None # ================================================================================ # 第三部分:AgentArts 平台标准化接入 (核心) # ================================================================================ @app.entrypoint async def handler(payload: Dict[str, Any], context: RequestContext = None) -> Dict[str, Any]: """ AgentArts 平台标准 HTTP 暴露入口 """ if not my_agent: return {"response": "服务未正确初始化,请检查环境变量。", "status": "error"} # 获取用户输入 query = payload.get("message", "") # 从平台上下文中提取 session_id # 如果 context 存在且包含 session_id,则使用平台的;否则生成一个默认的 current_session_id = context.session_id if context and hasattr(context, 'session_id') else "default_web_session" try: # 将提取到的 session_id 传给 agent.run(),激活云端记忆 response = my_agent.run(query, session_id=current_session_id) return {"response": response, "status": "success"} except Exception as e: return {"response": f"执行出错: {str(e)}", "status": "error"} if __name__ == "__main__": # 平台托管时,必须读取 AGENT_RUN_PORT 环境变量以匹配容器端口契约 run_port = int(os.getenv("AGENT_RUN_PORT", 8080)) app.run(port=run_port) EOF - 执行python app.py启动http server,执行以下命令调用。通过该方法,在将代码真正推送到云端之前,在本地模拟云端环境,验证Agent的HTTP接口是否已经被正确封装且能正常通信。
执行python app.py回显效果如下。
打开一个新的终端窗口(保持原窗口运行),使用curl命令进行测试。测试完成后,可以使用Ctrl + C停止运行的进程。curl --location --request POST 'http://localhost:8080/invocations' \ --header 'Content-Type: application/json' \ --data-raw '{"message": "你好,请用一句话做个自我介绍,并查询下杭州的天气。"}'
- 部署智能体运行时。
按上述步骤完成后,基本代码开发已经完成,接下来准备将本地创建好的智能体部署托管到AgentArts平台。
首先准备依赖文件requirements.txt,内容可参考如下:
# ================================================================================ # AgentArts LangGraph Agent Demo - 依赖清单 # ================================================================================ # # 本项目基于 AgentArts 平台,使用 LangGraph 框架构建 AI Agent # 依赖分为两部分:基础框架依赖 + 平台 SDK 依赖 # # 安装方式: # pip install -r requirements.txt # # ================================================================================ # ================================================================================ # 第一部分:LangGraph & LangChain 核心框架 # ================================================================================ # LangGraph: AI Agent 工作流编排框架 # - 状态图定义与执行 # - Checkpoint 持久化机制 # - 条件边与节点路由 langgraph>=0.2.0 # LangChain: LLM 应用开发工具链 # - 消息类型定义 (HumanMessage, AIMessage, SystemMessage) # - 工具系统 (@tool 装饰器) # - LLM 客户端封装 langchain>=0.3.0 langchain-core>=0.3.0 langchain-community>=0.3.0 # ================================================================================ # 第二部分:LLM Provider 支持 # ================================================================================ # OpenAI 兼容接口 openai>=1.0.0 langchain-openai>=0.1.0 # Anthropic (Claude) 支持 anthropic>=0.18.0 langchain-anthropic>=0.1.0 # ================================================================================ # 第三部分:HTTP & 网络 # ================================================================================ requests>=2.31.0 httpx>=0.27.0 # ================================================================================ # 第四部分:工具与配置 # ================================================================================ # 环境变量管理 python-dotenv>=1.0.0 # 异步支持 aiofiles>=23.0.0 # JSON/YAML 支持 pyyaml>=6.0 # ================================================================================ # 第五部分:AgentArts 平台 SDK(核心依赖) # ================================================================================ agentarts-sdk
- 执行如下命令配置智能体。
agentarts configure --entrypoint app:app
执行后按照操作指引进行配置。
配置智能体名称(以小写字母开头,以小写字母或数字结尾,可以包含小写字母、数字和中划线)、服务部署区域(使用cn-southwest-2,仅支持此区域)、requirements.txt依赖文件、SWR Organization镜像组织名称(如果使用自定义的镜像组织名,需要在SWR服务控制台贵阳一region创建)

- 执行命令部署智能体。
agentarts launch
该命令会自动完成以下步骤:
- 本地构建Docker镜像。
- 将Docker镜像推送到华为云SWR镜像仓库。
- 部署到AgentArts运行时托管环境。

- 调用云端Agent进行会话。
agentarts invoke '{"message": "你好,请用一句话做个自我介绍。"}'
常见问题
- 执行agentarts launch命令时 出现AK/SK认证报错 执行以下命令配置华为云凭证,获取华为云凭证请参考认证鉴权。
export HUAWEICLOUD_SDK_AK="your-access-key" export HUAWEICLOUD_SDK_SK="your-secret-key"

- 执行agentarts launch命令出现运行时Runtime名称错误
智能体的名称需要以小写字母开头,以小写字母或数字结尾,可以包含小写字母、数字和中划线,长度为2-48个字符。
请重新执行agentarts configure --entrypoint app:app命令进行配置。

- 执行agentarts launch命令,在执行requirements.txt步骤中出现Read timed out ... files.pythonhosted.org报错。
问题现象及原因
Docker镜像在构建过程中,尝试从官方PyPI服务器下载在requirements.txt里写的那些Python包,由于网络环境原因造成超时。

解决方案
执行以下命令让其在安装依赖时使用华为云官方的Python镜像加速器。sed -i 's|pip install --no-cache-dir -r requirements.txt|pip install --no-cache-dir -r requirements.txt -i https://repo.huaweicloud.com/repository/pypi/simple --trusted-host repo.huaweicloud.com|g' Dockerfile
- 执行python main.py或者python app.py命令出现Memory记忆库报错。
请检查记忆库ID以及记忆库API Key配置是否正确,建议重新配置该值。

