进阶示例：构建全能出行助手（集成模型/记忆/沙箱/网关/高德mcp）

整体架构图

工作流程

步骤一：平台组件创建与配置

在AgentArts平台依次创建记忆库、沙箱工具和网关，并获取后续代码接入所需的各项配置信息。

1. 在AgentArts平台创建记忆库。
   1. 登录AgentArts智能体平台。
   2. 在左侧导航栏选择“开发中心 > 组件库 ”，进入“记忆库”界面。
   3. 单击“创建记忆库”。填写记忆库名称、描述；勾选“私网访问”、“公网访问”；勾选全部的“长期记忆提取策略”，其余参数使用默认配置。
      图1 创建记忆库
      
      图2 配置记忆策略
      
   4. 记忆库创建完成后，单击记忆库名称，进入“基本信息”界面，获取记忆库ID。
      图3 获取记忆库ID
      
   5. 在“API Key”处单击“更新”。
      图4 更新API Key
      
   6. 弹框中单击“确定”后，复制或下载API Key。
      图5 复制或下载API Key
      

2. 在AgentArts平台创建沙箱工具。
   1. 在AgentArts平台左侧导航栏中选择“开发中心 > 组件库”，并进入“沙箱工具”页面，单击“创建代码解释器”。
   2. 填写代码解释器相关配置。参考下表进行配置。

      表1 代码解释器配置说明

      参数	说明
      名称	可自定义。
      描述	可自定义。
      委托（可选）	（可选）授予的代理权限或代理功能，允许代表智能体与外部系统进行通信和交互。 可不填写，如需手动创建，请单击“创建委托”并参考如下配置创建。 委托名称：自定义。 信任主体类型：选择“云服务”。 云服务：搜索service.WorkloadSandboxMetadata。 其余参数使用默认值。 图6 创建委托
      入站身份认证	选择“API Key认证”。
      API Key名称	可自定义。
      日志记录	开启。
      出网网络配置	选择“公网访问”。

      图7 创建沙箱工具
      
   3. 配置完成后，单击“立即创建”创建沙箱。

      在沙箱工具列表中，记录名称。

      单击沙箱工具，进入“配置信息”页面，获取域名。

      在“配置信息”页面，单击URN链接，在新页面获取沙箱API Key。

      图8 获取沙箱工具名称
      
      图9 获取沙箱域名
      
      图10 获取沙箱API Key
      

3. 在AgentArts平台创建网关，配置高德MCP服务。
   1. 高德MCP如下，使用前，请参考高德开放平台文档，创建获取API Key。

      {
        "mcpServers": {
          "amap-maps-streamableHTTP": {
            "url": "https://mcp.amap.com/mcp?key=您在高德官网上申请的key"
          }
        }
      }

   2. 在AgentArts平台左侧导航栏中选择“开发中心 > 组件库”，并进入“网关”页面，单击“创建网关”。
   3. 填写网关基础信息并配置权限与身份认证。参考下表进行配置。

      表2 网关基础信息与权限身份认证配置

      参数	说明
      名称	可自定义。
      描述（可选）	网关的描述信息，可自定义。
      MCP 版本	2025-03-26
      委托	使用平台的默认值。
      入站身份认证	选择API Key认证。
      API Key名称	可自定义。
      日志记录	选择开启。
      工具检索	选择开启。
      高级配置	选择公网访问，开启会话保持。

      图11 填写网关基础信息与权限身份认证
      
   4. 单击“创建Target”，配置高德MCP服务。

      表3 高德Target配置

      参数	说明
      名称	填写为gaodemap。
      描述（可选）	可自定义。
      类型	选择“MCP”，使用现成的高德MCP服务，不需要使用直接对接外部REST API的方式。
      传输方式	选择“Streamable HTTP”，与高德MCP的调用方式保持一致。
      MCP地址	填写：https://mcp.amap.com/mcp
      出站身份认证	选择“API Key”，单击“创建出站身份”，参数配置如下： 身份名称：填写为gaodemcp。 认证类型：选择“API Key”。 API Key的值：填写高德开发平台中创建的API Key。 出站身份创建完成后，返回“创建Target”页面，选择已创建的出站身份。 位置：选择“查询参数”。 参数名称：填写为key。 前缀：不填。 图12 创建出站身份 图13 配置出站身份认证

   5. 配置完成后，单击“确定”创建Target。Target创建完成后，回到“创建网关”页面，高级配置选择“公网访问”后，单击“创建网关”。
   6. 网关创建完成后，回到网关列表页面。单击网关名称，记录网关URL及网关的API Key（注意是网关的API Key，不是高德mcp的key）。
      图14 网关列表
      
   7. 调测网关。

      MCP网关配置完成后，建议在“调测”页面进行一次测试，验证MCP工具是否可以正常调用。在网关详情页面单击“调测”标签，确认能够正常返回结果后再进行后续开发。

      图15 调测网关
      
   8. 在网关详情页面获取网关URL。
      图16 获取网关URL
      
   9. 在网关详情页面，单击URN链接。进入Agent Identity页面，获取网关的API Key。
      图17 单击URN链接
      
      图18 获取网关API Key
      

步骤二：编写Agent核心逻辑

使用LangGraph框架编写智能体的核心对话逻辑，并集成记忆库、MCP 网关、沙箱工具与华为云MaaS模型。

1. 执行如下命令安装langchain、langgraph、langchain-openai。

   pip install -U langchain langgraph langchain-openai

2. 创建.env环境变量配置文件。

   # ==========================================
   # 1. 模型配置 (对接华为云MaaS模型)
   # ==========================================
   MODEL_NAME=deepseek-v4-flash
   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 Key
   HUAWEICLOUD_SDK_REGION=cn-southwest-2
   
   # ==========================================
   # 3. 网关配置
   # ==========================================
   # 网关地址
   GATEWAY_ENDPOINT=替换为网关的URL
   # 网关的 apikey
   GATEWAY_INBOUND_TOKEN=替换为网关的API Key
   
   # ==========================================
   # 4. 沙箱配置
   # ==========================================
   SANDBOX_NAME=在AgentArts平台中创建的沙箱名称
   AGENTARTS_CODEINTERPRETER_DATA_ENDPOINT="http://您的沙箱域名"
   HUAWEICLOUD_SDK_CODE_INTERPRETER_API_KEY="您的沙箱 API Key"
   
   # ==========================================
   # 5. Agent 全局配置
   # ==========================================
   MAX_ITERATIONS=50
   SESSION_TTL=3600
   # 本地测试启动 HTTP Server 时绑定的端口
   AGENT_RUN_PORT=8080

   获取模型API Key的方法请参考获取华为云MaaS服务模型API Key。

   agent.py文件的示例中，使用华为云MaaS服务提供的deepseek-v4-flash模型，如果使用其他模型注意MODEL_NAME的值，需填写为模型接口中model参数的值，可以参考如下方式获取。注意需要选择OpenAI兼容接口。

   图19 获取OpenAI兼容接口的model值
   

3. 在本地创建agent.py文件，编辑代码如下。 示例代码中创建了出行助手智能体，并对接记忆库、网关、沙箱、身份认证、华为云MaaS服务提供的模型。

   """
   ================================================================================
   AgentArts LangGraph 智能体核心实现
   ================================================================================
   启动时自动从Gateway 拉取 tools/list，动态生成 LangChain Tool 对象。
   ================================================================================
   """
   import os
   import uuid
   import json
   import re
   from typing import List, Optional, Dict, Any, Type
   
   import requests
   import urllib3
   from dotenv import load_dotenv
   from pydantic import create_model, Field
   
   urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
   
   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 StructuredTool
   from langchain_openai import ChatOpenAI
   
   try:
       from agentarts.sdk.memory import MemoryClient
       from agentarts.sdk.memory.inner.config import TextMessage
   except ImportError:
       MemoryClient = None
       TextMessage = None
       print("[Warning] 未检测到 agentarts.sdk，将跳过云端记忆同步。")
   
   load_dotenv()
   
   
   # ================================================================================
   # 常量
   # ================================================================================
   _MAX_ITERATIONS = 12
   _FORCE_FINAL_REPLY_ITERATION = 10
   _TOOL_RESULT_MAX_LENGTH = 8000
   _HISTORY_K_MESSAGES = 20
   _REQUEST_TIMEOUT = 30
   _ACTOR_ID = "user_123"
   _ASSISTANT_ID = "travel_agent"
   _TOOL_CALL_TAG = "<tool_call>"
   
   _DEFAULT_MODEL_NAME = "deepseek-v4-flash"
   _DEFAULT_MODEL_URL = "https://api.modelarts-maas.com/openai/v1"
   _DEFAULT_REGION = "cn-southwest-2"
   _DEFAULT_MAX_ITERATIONS = 50
   
   _SYSTEM_PROMPT = (
       "你是一个专业的出行规划助手，具备以下能力：\n\n"
       "【地图服务 - 高德MCP】\n"
       "- 天气查询：帮用户判断出行日期是否合适\n"
       "- 地理编码：将地址转为坐标\n"
       "- 路径规划：计算驾车/步行/骑行/公交路线\n"
       "- 周边搜索：查找餐厅、酒店、景点等\n\n"
       "【代码沙箱】\n"
       "- 仅在用户明确要求计算费用/预算/总花费时使用\n"
       "- 不适用于：查位置、查路线、查天气、展示信息\n\n"
       "【云端记忆】\n"
       "- 记住用户提到的城市、日期、偏好\n"
       "- 跨轮次累积规划，支持增量调整\n\n"
       "工作原则：\n"
       "1. 规划出行时，先查天气判断可行性\n"
       "2. **何时调用 execute_python**：只有当用户明确要求计算费用、预算、总价、多方案费用对比时才调用。"
       "查位置、查路线、查天气、展示距离等任务直接用工具返回的结果回答，**不要调用 execute_python**。\n"
       "   - 对于不确定的价格（如门票、住宿、餐饮），用合理估计值\n"
       "   - 在 Python 代码中定义这些变量，用代码做加法/乘法计算总费用\n"
       "   - 禁止直接给出估算数字，必须通过沙箱计算\n"
       "3. 记住用户之前提到的所有信息，避免重复询问\n"
       "4. 多轮对话中，基于已有上下文做增量调整\n"
       "5. 路径规划时，尽量同时查询所有出行方式（驾车、步行、骑行、公交），给用户完整的对比"
   )
   
   
   # ================================================================================
   # 消息清洗（防止 ModelArts 81001 错误）
   # ================================================================================
   def clean_messages_for_llm(messages: List[BaseMessage]) -> List[BaseMessage]:
       cleaned = []
       expecting_tool_results = False
       for msg in messages:
           if isinstance(msg, AIMessage) and getattr(msg, "tool_calls", None):
               cleaned.append(msg)
               expecting_tool_results = True
           elif isinstance(msg, ToolMessage):
               if expecting_tool_results:
                   cleaned.append(msg)
               else:
                   print("[MaaS Guard] 检测到孤立的 ToolMessage，已自动过滤。")
           else:
               cleaned.append(msg)
               expecting_tool_results = False
       return cleaned
   
   
   # ================================================================================
   # 配置
   # ================================================================================
   class ModelConfig:
       def __init__(self):
           self.name = os.getenv("MODEL_NAME", _DEFAULT_MODEL_NAME)
           self.url = os.getenv("MODEL_URL", _DEFAULT_MODEL_URL)
           self.api_key = os.getenv("MODEL_API_KEY")
           if not self.api_key:
               raise ValueError("[错误] 未找到 MODEL_API_KEY！请在 .env 文件中进行配置。")
   
   
   class AgentConfig:
       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", _DEFAULT_REGION)
           self.max_iterations = int(os.getenv("MAX_ITERATIONS", str(_DEFAULT_MAX_ITERATIONS)))
   
   
   config = AgentConfig()
   
   
   # ================================================================================
   # Gateway 交互层（tools/list + tools/call）
   # ================================================================================
   def _mcp_headers() -> Dict[str, str]:
       return {
           "Content-Type": "application/json",
           "Accept": "application/json, text/event-stream",
           "mcp-session-id": str(uuid.uuid4()),
           "Authorization": f"Bearer {os.getenv('GATEWAY_INBOUND_TOKEN', '')}"
       }
   
   
   def _parse_mcp_response(raw_text: str) -> dict:
       """解析 MCP 响应：兼容 SSE (data: {...}) 和纯 JSON 两种格式"""
       if "data:" in raw_text:
           match = re.search(r'\{.*\}', raw_text, re.DOTALL)
           if match:
               return json.loads(match.group())
       return json.loads(raw_text)
   
   
   def _mcp_request(method: str, params: dict = None) -> dict:
       """通用 MCP JSON-RPC 请求"""
       gateway_url = os.getenv("GATEWAY_ENDPOINT")
       if not gateway_url:
           raise RuntimeError("未配置 GATEWAY_ENDPOINT")
       payload = {
           "jsonrpc": "2.0",
           "id": str(uuid.uuid4()),
           "method": method,
           "params": params or {}
       }
       resp = requests.post(gateway_url, headers=_mcp_headers(), json=payload, verify=False, timeout=_REQUEST_TIMEOUT)
       return _parse_mcp_response(resp.text)
   
   
   def list_mcp_tools() -> List[dict]:
       """第一步：调 tools/list 拉取 Gateway 上所有工具定义（支持 cursor 分页）"""
       all_tools = []
       cursor = None
       try:
           while True:
               params = {"cursor": cursor} if cursor else {}
               result = _mcp_request("tools/list", params)
               result_data = result.get("result", {})
               all_tools.extend(result_data.get("tools", []))
               cursor = result_data.get("nextCursor")
               if not cursor:
                   break
           print(f"[MCP] 从 Gateway 发现 {len(all_tools)} 个工具")
           return all_tools
       except Exception as e:
           print(f"[MCP] tools/list 失败: {e}")
           return all_tools
   
   
   _TRUNCATION_SUFFIX = "\n...(结果已截断，原始数据过长)"
   _MIN_TRUNCATION_RATIO = 0.8
   
   
   def _truncate_safely(text: str, max_length: int) -> str:
       """安全截断：在 JSON 结构边界处截断，避免破坏数据完整性"""
       if len(text) <= max_length:
           return text
       truncated = text[:max_length]
       last_brace = max(truncated.rfind('}'), truncated.rfind(']'))
       if last_brace > max_length * _MIN_TRUNCATION_RATIO:
           return truncated[:last_brace + 1] + _TRUNCATION_SUFFIX
       return truncated + _TRUNCATION_SUFFIX
   
   
   def call_mcp_tool(tool_name: str, arguments: Dict[str, Any]) -> str:
       """第四步：模型决定调某工具后，透传到 MCP tools/call"""
       try:
           result = _mcp_request("tools/call", {"name": tool_name, "arguments": arguments})
           text = json.dumps(result.get("result", result), ensure_ascii=False)
           return _truncate_safely(text, _TOOL_RESULT_MAX_LENGTH)
       except Exception as e:
           return f"网关调用异常: {e}"
   
   
   # ================================================================================
   # 第二步：把 MCP 工具定义动态转成 LangChain StructuredTool
   # ================================================================================
   _JSON_TYPE_MAP = {
       "string": str,
       "number": float,
       "integer": int,
       "boolean": bool,
       "array": list,
       "object": dict,
   }
   
   
   def _json_schema_to_pydantic(schema: dict, model_name: str = "Args") -> Type:
       """将 MCP inputSchema (JSON Schema) 转为 Pydantic Model 类"""
       properties = schema.get("properties", {})
       required = set(schema.get("required", []))
       fields = {}
       for param_name, param_info in properties.items():
           json_type = param_info.get("type", "string")
           py_type = _JSON_TYPE_MAP.get(json_type, str)
           desc = param_info.get("description", "")
           if param_name in required:
               fields[param_name] = (py_type, Field(..., description=desc))
           else:
               fields[param_name] = (Optional[py_type], Field(None, description=desc))
       return create_model(model_name, **fields) if fields else create_model(model_name)
   
   
   def _make_tool_caller(tool_name: str):
       """闭包工厂：每个工具绑定自己的 tool_name，防止循环变量捕获问题"""
       def caller(**kwargs):
           clean_args = {k: v for k, v in kwargs.items() if v is not None}
           for k, v in clean_args.items():
               if isinstance(v, (int, float)):
                   clean_args[k] = str(v)
           print(f"\n[Gateway] 调用工具: {tool_name} | 参数: {clean_args}")
           return call_mcp_tool(tool_name, clean_args)
       return caller
   
   
   def build_langchain_tool(mcp_tool_def: dict) -> StructuredTool:
       """将单个 MCP 工具定义转为 LangChain StructuredTool"""
       name = mcp_tool_def["name"]
       desc = mcp_tool_def.get("description", "") or f"MCP tool: {name}"
       schema = mcp_tool_def.get("inputSchema", {})
       args_model = _json_schema_to_pydantic(schema, f"{name}_Args")
       caller = _make_tool_caller(name)
       return StructuredTool.from_function(
           func=caller,
           name=name,
           description=desc,
           args_schema=args_model,
       )
   
   
   def discover_mcp_tools() -> List[StructuredTool]:
       """完整的工具发现流程：list -> build"""
       mcp_tools = list_mcp_tools()
       return [build_langchain_tool(t) for t in mcp_tools]
   
   
   # ================================================================================
   # 本地工具：代码解释器（非 MCP，保持独立定义）
   # ================================================================================
   def _execute_python(code: str) -> str:
       from agentarts.sdk.tools import code_session
       print(f"\n[Sandbox] 正在安全沙箱中执行代码...")
       sandbox_name = os.getenv("SANDBOX_NAME")
       api_key = os.getenv("SANDBOX_API_KEY") or os.getenv("HUAWEICLOUD_SDK_CODE_INTERPRETER_API_KEY")
       if not sandbox_name:
           return "沙箱调用失败: 未在环境变量中配置 SANDBOX_NAME"
       try:
           with code_session(config.region, sandbox_name, api_key=api_key) as client:
               res = client.invoke(operate_type="execute_code", api_key=api_key,
                                   arguments={"code": code, "language": "python"})
           return json.dumps(res.get("result"))
       except Exception as e:
           return f"沙箱执行失败: {e}"
   
   
   execute_python = StructuredTool.from_function(
       func=_execute_python,
       name="execute_python",
       description="在平台沙箱环境中安全执行 Python 代码。Args: code: 待执行的Python代码",
       args_schema=create_model("execute_python_Args", code=(str, Field(..., description="待执行的Python代码")))
   )
   
   
   # ================================================================================
   # LangGraph 工作流
   # ================================================================================
   class AgentState(MessagesState):
       iteration: int
   
   
   def should_continue(state: AgentState) -> str:
       if state.get("iteration", 0) >= _MAX_ITERATIONS:
           print(f"[Safety] 达到最大迭代次数({_MAX_ITERATIONS})，强制结束")
           return "end"
       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,
           )
           self._mcp_tools = discover_mcp_tools()
           self._all_tools = self._mcp_tools + [execute_python]
           print(f"[Agent] 共加载 {len(self._all_tools)} 个工具（MCP: {len(self._mcp_tools)}, 本地: 1）")
           self.graph = self._build_graph()
   
       def _build_graph(self):
           def model_node(state: AgentState) -> AgentState:
               messages = state["messages"]
               current_iteration = state.get("iteration", 0)
               if current_iteration >= _FORCE_FINAL_REPLY_ITERATION:
                   print("[Safety] 接近最大迭代次数，强制生成最终回复")
                   final_prompt = SystemMessage(content=(
                       "你已达到工具调用次数上限。请基于已有工具返回的结果，"
                       "直接给出最终的文字回复，不要再调用任何工具。"
                       "如果信息不足，请如实告知用户已获取的部分结果。"
                   ))
                   safe_messages = clean_messages_for_llm(messages + [final_prompt])
                   response = self._llm.invoke(safe_messages)
               else:
                   llm_with_tools = self._llm.bind_tools(self._all_tools)
                   safe_messages = clean_messages_for_llm(messages)
                   response = llm_with_tools.invoke(safe_messages)
               return {"messages": [response], "iteration": current_iteration + 1}
   
           workflow = StateGraph(AgentState)
           workflow.add_node("agent", model_node)
           workflow.add_node("tools", ToolNode(self._all_tools))
           workflow.set_entry_point("agent")
           workflow.add_conditional_edges("agent", should_continue, {"continue": "tools", "end": END})
           workflow.add_edge("tools", "agent")
           # 不使用 checkpointer，每轮从云端记忆库加载历史
           return workflow.compile()
   
       def run(self, user_input: str, session_id: str) -> str:
           thread_config = {"configurable": {"thread_id": session_id}}
           self._ensure_session(session_id)
           history = self._load_cloud_history(session_id)
           messages = self._build_messages(history, user_input)
           prev_count = len(messages) - 1
           result = self.graph.invoke(
               {"messages": messages, "iteration": 0},
               config=thread_config,
               recursion_limit=config.max_iterations
           )
           new_messages = result["messages"][prev_count:]
           self._save_to_cloud_memory(session_id, new_messages)
           return self._clean_response(result["messages"][-1].content)
   
       def _ensure_session(self, session_id: str):
           """确保云端记忆会话存在"""
           if MemoryClient is None or not config.memory_space:
               return
           try:
               MemoryClient().create_memory_session(
                   space_id=config.memory_space,
                   id=session_id,
                   actor_id=_ACTOR_ID
               )
           except Exception as e:
               if "Session id already exists" not in str(e):
                   print(f"[Warning] 云端记忆会话同步失败: {e}")
   
       def _build_messages(self, history: List[BaseMessage], user_input: str) -> List[BaseMessage]:
           """构建送入 LLM 的消息列表：系统提示 + 历史 + 当前输入"""
           messages = [SystemMessage(content=_SYSTEM_PROMPT)]
           messages.extend(history)
           messages.append(HumanMessage(content=user_input))
           return messages
   
       def _clean_response(self, response: str) -> str:
           """清理可能幻觉的工具调用标签"""
           if _TOOL_CALL_TAG in response:
               return response[:response.index(_TOOL_CALL_TAG)].strip()
           return response
   
       def _load_cloud_history(self, session_id: str) -> List[BaseMessage]:
           """从云端记忆加载历史消息"""
           if MemoryClient is None or not config.memory_space:
               return []
           try:
               client = MemoryClient()
               response = client.get_last_k_messages(
                   space_id=config.memory_space, session_id=session_id, k=_HISTORY_K_MESSAGES
               )
               msgs = None
               if isinstance(response, list):
                   msgs = response
               elif hasattr(response, 'messages'):
                   msgs = response.messages
               if msgs is None:
                   print(f"[Memory] 无法提取消息列表，{type(response).__name__}")
                   return []
               history = []
               for m in msgs:
                   role = getattr(m, "role", "")
                   content = getattr(m, "content", "")
                   if not content and hasattr(m, 'parts') and m.parts:
                       part = m.parts[0]
                       content = part.get('text', '') if isinstance(part, dict) else getattr(part, 'text', '')
                   if role == "user" and content:
                       history.append(HumanMessage(content=content))
                   elif role == "assistant" and content:
                       history.append(AIMessage(content=content))
               if history:
                   print(f"[Memory] 从云端加载 {len(history)} 条历史消息")
               return history
           except Exception as e:
               print(f"[Memory] 加载历史失败: {e}")
               return []
   
       def _save_to_cloud_memory(self, session_id: str, new_messages: List[BaseMessage]):
           """增量同步：只写入本轮新增的消息到 AgentArts 记忆库"""
           if MemoryClient is None or TextMessage is None or not config.memory_space:
               return
           try:
               text_messages = []
               for msg in new_messages:
                   if isinstance(msg, HumanMessage) and msg.content:
                       text_messages.append(TextMessage(role="user", content=msg.content, actor_id=_ACTOR_ID, assistant_id=_ASSISTANT_ID))
                   elif isinstance(msg, AIMessage) and msg.content:
                       text_messages.append(TextMessage(role="assistant", content=msg.content, actor_id=_ACTOR_ID, assistant_id=_ASSISTANT_ID))
               if text_messages:
                   MemoryClient().add_messages(
                       space_id=config.memory_space,
                       session_id=session_id,
                       messages=text_messages
                   )
                   print(f"[Memory] 增量同步 {len(text_messages)} 条新消息到云端")
           except Exception as e:
               print(f"[Memory] 云端同步失败（不影响对话）: {e}")

步骤三：本地功能验证

在本地直接运行智能体，确认智能体可以正常运行。

1. 创建main.py文件，用于本地运行智能体，并进行测试。 测试脚本为交互式对话，可直接输入问题测试agent.py中集成的基础对话、网关、沙箱工具、记忆恢复等能力。

   """
   出行规划助手 - 交互式场景演示
   """
   import sys
   import uuid
   from agent import TravelAgent
   
   _EXIT_COMMANDS = {"exit", "quit", "退出"}
   
   
   def safe_input(prompt="你: "):
       """
       兼容 Cloud Shell 终端编码的输入函数。
       Cloud Shell 可能以 GBK 编码发送中文字节，而 Python 默认按 UTF-8 解码会报 UnicodeDecodeError。
       策略：先尝试 UTF-8，失败后回退到 GBK，再失败则用 replace 模式兜底。
       """
       print(prompt, end="", flush=True)
       raw = sys.stdin.buffer.readline()
       if not raw:
           return ""
       try:
           return raw.decode("utf-8").strip()
       except UnicodeDecodeError:
           try:
               return raw.decode("gbk").strip()
           except UnicodeDecodeError:
               return raw.decode("utf-8", errors="replace").strip()
   
   
   def print_guide():
       print("\n" + "=" * 60)
       print("  杭州周末出行规划助手")
       print("=" * 60)
       print("""
     我可以帮你：
     - 查天气，判断出行是否合适
     - 查地点、规划路线
     - 计算行程预算
     - 记住你的偏好，持续优化方案
   
     建议对话流程：
     1. "我周末想去杭州玩，先帮我看看天气"
     2. "查一下西湖和灵隐寺的位置"
     3. "这两个地方之间怎么走？多远？"
     4. "帮我算一下两天行程的总预算"
     5. "如果改成三天呢？住宿多一晚"
   """)
       print("=" * 60 + "\n")
   
   
   def main():
       print("正在初始化智能体助手（集成网关、代码解释器与云端记忆）...")
       agent = TravelAgent()
       session_id = str(uuid.uuid4())
       print(f"会话 ID: {session_id}")
       print_guide()
       while True:
           try:
               user_input = safe_input()
               if not user_input:
                   continue
               if user_input.lower() in _EXIT_COMMANDS:
                   print("再见！祝旅途愉快～")
                   break
               print("\n" + "-" * 40)
               response = agent.run(user_input, session_id=session_id)
               print(f"\n助手: {response}\n")
               print("-" * 40 + "\n")
           except KeyboardInterrupt:
               print("\n再见！")
               break
   
   
   if __name__ == "__main__":
       main()

2. 将agent.py、.env、main.py上传至服务器项目目录中。

   

3. 执行python main.py命令对智能体进行本地测试。 由于main.py是交互式对话程序，运行后可直接输入问题，智能体会调用大模型进行回复。有正常的回复即表示本地测试成功。

   

交互场景与组件调用

问题 1："我周末想去杭州玩，先帮我看看天气"

• 用户意图：查询目的地天气，判断出行日期是否合适
• 调用组件：网关、记忆库
• 组件作用：
  • MCP Gateway：LLM 决策调用 maps_weather 工具，Gateway 转发请求到高德天气服务，返回杭州未来几天的天气数据（温度、天气状况、风向）。
  • 记忆库：首次对话，MemoryClient().create_memory_session() 在云端创建记忆会话，为跨会话持久化做准备；get_last_k_messages(k=20) 加载 0 条历史（首轮无历史）；对话后增量同步 3 条新消息（用户输入 + AI 工具调用 + AI 最终回复）到云端记忆库。

问题 2："查一下西湖和灵隐寺的位置"

• 用户意图：获取两个景点的地理坐标和距离
• 调用组件：网关、记忆库
• 组件作用：
  • MCP Gateway：LLM 连续调用 3 次工具——maps_geo（西湖坐标）、maps_geo（灵隐寺坐标）、maps_distance（两点驾车距离 4.4 公里）。LLM 自主决策调用顺序和参数，Gateway 透传执行。
  • 记忆库：对话前 get_last_k_messages(k=20) 从云端加载 3 条历史消息（Turn 1 的对话），LLM 能记住"之前查了杭州天气"的语境；对话后增量同步 4 条新消息到云端记忆库。

问题 3："这两个地方之间怎么走？多远？"

• 用户意图：获取多种出行方式的路线规划
• 调用组件：网关、记忆库
• 组件作用：
  • MCP Gateway：LLM 一次性调用 4 个路径规划工具——maps_direction_driving（驾车）、maps_direction_walking（步行）、maps_direction_bicycling（骑行）、maps_direction_transit_integrated（公交）。Gateway 并行执行，返回 4 种出行方式的路线数据。安全截断机制检查每个结果长度，超过 8000 字符时会在 JSON 结构边界处截断，避免破坏数据完整性，防止云端记忆写入失败。
  • 记忆库：对话前 get_last_k_messages(k=20) 从云端加载 7 条历史消息（Turn 1-2 的对话），LLM 记住前两轮的景点坐标和天气数据，回复中引用"结合周末雨天情况"给出出行建议；对话后增量同步 3 条新消息到云端记忆库。

问题 4："帮我算一下两天行程的总预算"

• 用户意图：精确计算两日行程的费用明细
• 调用组件：代码解释器、记忆库
• 组件作用：
  • 代码沙箱：LLM 不调用 MCP 工具，直接调用 execute_python，在沙箱中执行 Python 代码——定义住宿（经济型 300 元/晚）、餐饮、交通等变量，用代码做加法计算总费用，返回精确结果。LLM 基于沙箱返回的数值生成预算明细表格。
  • 记忆库：对话前 get_last_k_messages(k=20) 从云端加载 10 条历史消息（Turn 1-3 的对话），LLM 记住前 3 轮的景点和路线信息，在预算计算中引用之前查询的结果；对话后增量同步 3 条新消息到云端记忆库。

问题 5："如果改成三天呢？住宿多一晚"

• 用户意图：基于已有两日预算做增量调整
• 调用组件：代码解释器、记忆库
• 组件作用：
  • 代码沙箱：LLM 生成新的 Python 代码，复用 Turn 4 的变量定义，增加第三天的费用项（+1 晚住宿 300 元 +1 天餐饮 280 元 +1 天交通 40 元 + 额外景点 80 元），沙箱执行增量计算，返回 +700 元的差额。
  • 记忆库：对话前 get_last_k_messages(k=20) 从云端加载 13 条历史消息（Turn 1-4 的对话），LLM 记住 Turn 4 的两天预算方案，能做增量调整而非重新计算；对话后增量同步 3 条新消息到云端记忆库，保存完整 5 轮对话上下文，下次开新会话能加载历史继续规划。

步骤四：SDK封装与本地接口验证

将智能体封装为符合AgentArts平台规范的HTTP服务，并在本地模拟云端接口调用，确认封装正确后再执行后续的部署上云操作。

1. SDK封装，AgentArts平台通过标准HTTP协议与智能体通信，您需要用AgentArts SDK将Agent逻辑包一层，将其暴露为符合平台规范的/invocations接口，平台才能正确调度和管理您的智能体。 创建app.py文件，并上传至服务器中。该文件会将本地的Agent逻辑封装为符合AgentArts平台规范的Web服务。

   """
   AgentArts 平台 HTTP 封装层
   将 TravelAgent 暴露为符合平台规范的 /invocations 接口
   """
   import socket
   import uuid
   from typing import Dict, Any
   
   # ==========================================
   # DNS 补丁：容器内无法解析 AgentArts 公网域名
   # 必须在 import agent / sdk 之前执行
   # 通过 nslookup <域名> 获取对应 IP，替换下面的 XXX.XXX.XXX.XXX
   # ==========================================
   _DNS_OVERRIDES = {
       "memory.cn-southwest-2.huaweicloud-agentarts.com": "XXX.XXX.XXX.XXX",
       "gateway-gaodemcp-defaultgw-ge3wkzbthf.cn-southwest-2.huaweicloud-agentarts.com": "XXX.XXX.XXX.XXX",
       "defaultgw-ge3wkzbthf.cn-southwest-2.huaweicloud-agentarts.com": "XXX.XXX.XXX.XXX",
   }
   
   _orig_getaddrinfo = socket.getaddrinfo
   
   def _patched_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
       if host and host in _DNS_OVERRIDES:
           return _orig_getaddrinfo(_DNS_OVERRIDES[host], port, family, type, proto, flags)
       return _orig_getaddrinfo(host, port, family, type, proto, flags)
   
   socket.getaddrinfo = _patched_getaddrinfo
   
   from agentarts.sdk import AgentArtsRuntimeApp, RequestContext
   from agent import TravelAgent
   
   app = AgentArtsRuntimeApp()
   
   try:
       my_agent = TravelAgent()
   except Exception as e:
       print(f"Agent 初始化失败: {e}")
       my_agent = None
   
   
   def _resolve_session_id(context: RequestContext = None, payload: Dict[str, Any] = None) -> str:
       raw = None
       if context:
           raw = getattr(context, "session_id", None)
       if not raw and payload:
           raw = payload.get("session_id")
       if not raw:
           raw = str(uuid.uuid4())
       return str(raw)
   
   
   @app.entrypoint
   def invoke(payload: Dict[str, Any], context: RequestContext = None) -> Dict[str, Any]:
       message = payload.get("message", "")
       session_id = _resolve_session_id(context, payload)
   
       if my_agent is None:
           return {"response": "Agent 初始化失败", "status": "error", "session_id": session_id}
   
       try:
           response = my_agent.run(message, session_id=session_id)
           return {"response": response, "status": "success", "session_id": session_id}
       except Exception as e:
           return {"response": f"处理失败: {str(e)}", "status": "error", "session_id": session_id}
   
   
   if __name__ == "__main__":
       app.run(host="0.0.0.0", port=8080)

2. 接口验证，在正式推送镜像之前，先在本地模拟云端HTTP调用环境，验证Agent的接口封装是否正确、通信是否正常，可以大幅降低因接口问题导致云端部署后才发现错误的调试成本。 执行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": "你好，请用一句话做个自我介绍，并查询下杭州的天气。"}'

   

步骤五：云端部署

将本地智能体打包为Docker镜像并一键部署至AgentArts云端运行时。

1. 部署智能体运行时。 按上述步骤完成后，基本代码开发已经完成，接下来准备将本地创建好的智能体部署托管到AgentArts平台。 首先准备依赖文件requirements.txt，内容可参考如下：

   # ================================================================================
   # AgentArts LangGraph Agent Demo - 依赖清单
   # ================================================================================
   #
   # 本项目基于 AgentArts 平台，使用 LangGraph 框架构建 AI Agent
   #
   # 安装方式:
   #   pip install -r requirements.txt
   #
   # ================================================================================
   
   # 指定国内 PyPI 镜像源
   --index-url https://repo.huaweicloud.com/repository/pypi/simple
   --trusted-host repo.huaweicloud.com
   
   # ================================================================================
   # LangGraph & LangChain 核心框架
   # ================================================================================
   langgraph>=0.2.0
   langchain>=0.3.0
   langchain-core>=0.3.0
   langchain-openai>=0.1.0
   
   # ================================================================================
   # LLM Provider（ModelArts MaaS OpenAI 兼容接口）
   # ================================================================================
   openai>=1.0.0
   
   # ================================================================================
   # HTTP & 网络
   # ================================================================================
   requests>=2.31.0
   urllib3>=2.0.0
   
   # ================================================================================
   # 工具与配置
   # ================================================================================
   python-dotenv>=1.0.0
   pydantic>=2.0.0
   
   # ================================================================================
   # AgentArts 平台 SDK
   # ================================================================================
   agentarts-sdk>=0.1.0

2. 执行如下命令配置智能体。

   agentarts configure --entrypoint app:app

   执行后按照操作指引进行配置。 配置智能体名称（以小写字母开头，以小写字母或数字结尾，可以包含小写字母、数字和中划线）、服务部署区域（使用cn-southwest-2，仅支持此区域）、requirements.txt依赖文件、SWR Organization镜像组织名称（如果使用自定义的镜像组织名，需要在SWR服务控制台贵阳一region创建）。

   

3. 执行命令部署智能体。

   agentarts launch

   该命令会自动完成以下步骤：

   1. 本地构建Docker镜像。
   2. 将Docker镜像推送到华为云SWR镜像仓库。
   3. 部署到AgentArts运行时托管环境。

      

   4. 调用云端Agent进行会话。

      agentarts invoke --agent <你的agent名称> -s "<会话ID>" '{"message": "查一下西湖和灵隐寺的位置"}'

      

场景配置说明

本智能体采用解耦架构设计：工具动态发现、配置通过环境变量注入、System Prompt 独立定义。移植至其他业务场景时，仅需调整少量配置项，核心代码无需改动。

典型业务场景

• 客服助手

  重写System Prompt为客服话术规范 → MCP Gateway替换为工单/订单查询工具 → 保留execute_python用于退款金额计算 → 其余配置无需调整

• 数据分析助手

  重写System Prompt为数据分析流程 → MCP Gateway替换为SQL查询/数据导出工具 → 保留execute_python用于统计分析 → 其余配置无需调整

• 代码审查助手

  重写System Prompt为审查标准 → MCP Gateway替换为代码仓库/CI状态查询工具 → 保留execute_python用于复杂度/覆盖率计算 → 其余配置无需调整