# 高代码智能体开发与云端部署技巧
在前两章中，提示词编写和多智能体编排均在AgentArts平台页面操作完成。但在企业真实落地场景中，大量团队已在本地基于LangChain、LangGraph等主流框架积累了成熟的Agent代码资产，这些代码无法直接粘贴进画布节点。
AgentArts的高代码开发能力正是为解决这一问题而设计：开发者在本地用自己熟悉的框架写代码，通过标准化SDK即插即用地接入平台提供的记忆库、沙箱工具和MCP网关等云端组件，最终以容器镜像形式一键部署至智能体运行时（Agent Runtime，一个安全隔离、弹性伸缩的托管执行环境）。
本章按照"先把运行时跑起来，再按需挂载扩展能力"的工程路径，依次覆盖：从本地代码到云端部署的完整流程、记忆库的跨会话持久化集成、沙箱工具的安全代码执行、MCP网关的外部系统连接。
#### 高代码开发的两种构建模式
在开始之前，需要先明确AgentArts高代码开发支持的两种构建模式，根据团队现状选择合适路径：
表1高码开发Agent模式 
| 构建模式   | 适用场景                 | 平台组件依赖            | 说明                                                     |
|:---|:---|:---|:---|
| 从零开始构建 | 需要快速集成企业服务、降低重复开发。   | 使用平台记忆库、网关、沙箱等组件。 | 本地用LangChain/LangGraph写Agent核心逻辑，通过SDK集成云上能力组件，打包镜像部署。 |
| 代码云上托管 | 已有成熟代码资产、需要完全控制底层逻辑。 | 不依赖平台组件库。         | 本地完全自主开发，平台仅作运行时托管环境，提供部署和监控能力。                        |
   
选择原则：如果团队已有完整的Agent代码，只需要一个安全的云端运行环境，选择"代码云上托管"；如果需要集成AgentArts云端的跨会话记忆、安全沙箱执行能力，选择"从零开始构建"。
#### 搭建运行时：从本地代码到云端托管
**运行时是什么**
智能体运行时是智能体的云端执行底座。它基于容器化技术实现进程级隔离，支持基于QPS/并发数的自动扩缩容，并深度集成IAM实现细粒度的身份认证和网络管控。开发者只需关注业务逻辑，底层基础设施的运维完全由平台托管。
**关键技术约束**
在开始编码前，必须了解以下硬约束，否则部署后会遇到运行失败：
表2约束说明 
| 约束项      | 要求                      | 说明                                     |
|:---|:---|:---|
| 操作系统架构   | 必须使用ARM64系统构建镜像         | X86系统制作的镜像在调用运行时时会直接失败                 |
| 监听地址     | 0.0.0.0                 | 容器内必须绑定所有网络接口                          |
| HTTP端口   | 8080                    | HTTP协议智能体的标准端口                         |
| MCP端口    | 8000                    | MCP协议智能体的标准端口                          |
| Python版本 | ≥ 3.10                  | SDK最低要求                                |
| Docker版本 | ≥ 18.06                 | 华为云SWR基础版不支持OCI镜像格式，Docker 27+需关闭OCI支持 |
| 部署区域     | 仅cn-southwest-2（西南-贵阳一） | 当前唯一支持区域                               |
   
**代码结构**
任何高代码智能体都需要通过AgentArtsRuntimeApp类将业务逻辑封装为标准HTTP服务。以下是一个精简的代码说明：
```
# app.py
from agentarts.sdk import AgentArtsRuntimeApp, RequestContext
app = AgentArtsRuntimeApp()
@app.entrypoint
async def handler(payload: dict, context: RequestContext = None) -> dict:
    """智能体主入口，处理所有到达 /invocations 端点的请求"""
    message = payload.get("message", "")
    session_id = context.session_id if context else None
    # 在此处编写业务逻辑（调用大模型、执行工具等）
    return {"response": f"收到: {message}", "session_id": session_id}
if __name__ == "__main__":
    app.run(port=8080)
```
表3核心装饰器说明 
| 装饰器             | 端点                | 用途                                                    |
|:---|:---|:---|
| @app.entrypoint | POST /invocations | 主入口，支持返回dict（JSON响应）、生成器（SSE流式响应）、异步生成器。              |
| @app.ping       | GET /ping         | 健康检查，返回PingStatus.HEALTHY / HEALTHY_BUSY / UNHEALTHY。 |
| @app.websocket  | /ws               | WebSocket双向通信（可选）。                                    |
| @app.async_task | 无端点               | 注册后台异步任务，执行期间ping自动设为HEALTHY_BUSY。                    |
   
请求上下文：RequestContext包含user_id、session_id、request_id、workload_access_token等属性，由平台在调用时自动注入。在调用栈更深层的代码中，可通过全局上下文AgentArtsRuntimeContext.get_session_id()在任意位置获取，无需层层传参。
#### 实践案例：从初始化项目到云端托管部署
华为云AgentArts官方文档
[基础示例：创建基础对话智能体](https://support.huaweicloud.com/highcode-agentarts/agentarts_10_056.html)
[进阶示例：构建全能出行助手（集成模型/记忆/沙箱/网关/高德mcp）](https://support.huaweicloud.com/highcode-agentarts/agentarts_10_057.html)
#### 扩展能力1：记忆库
**什么场景需要记忆库**
当智能体需要跨会话记住用户的偏好、历史对话或业务状态时（如智能客服记住客户之前的投诉、个人助手记住用户的饮食偏好、问诊助手记住患者的病史），就需要接入记忆库。
记忆库提供短期记忆（当前会话的临时上下文，7\~365天过期）和长期记忆（持久化存储到向量数据库，跨会话可检索）的双层架构，支持按空间、会话、用户三个维度进行数据隔离。
**记忆库的工作原理**
1. 用户对话 → Agent将对话内容写入短期记忆。
2. 触发条件满足（空闲时间/Token累计量/消息数任一达到阈值） → 短期记忆传递至记忆提取模块。
3. 记忆提取执行三步处理：Embedding（向量嵌入） → Consolidation（巩固降维） → Extraction（关键信息抽取）。
4. 抽取结果存入向量数据库，成为可检索的长期记忆。
**记忆策略**
表4记忆策略说明 
| 策略   | 提取内容             | 典型应用           |
|:---|:---|:---|
| 总结   | 压缩会话中的主题、任务、决策记录 | 会议总结、工作汇报      |
| 语义记忆 | 与上下文无关的事实性知识和概念  | 个人信息、专业术语、教育经历 |
| 用户偏好 | 用户行为模式、关键偏好和选择   | 饮食偏好、购物习惯、沟通风格 |
| 情景记忆 | 有时空序列的具体事件和场景细节  | 旅行经历、就诊记录、项目事件 |
   
还可创建自定义策略：选择基础模型，分别定义抽取（Extraction）、整合（Consolidation）和反思（Reflection）三个阶段的提示词模板。
**SDK集成记忆库**
Memory SDK提供两种使用模式：
**Client模式**，适合需要完整控制的场景（如后台服务批量管理多用户记忆）：
```
from agentarts.sdk.memory import MemoryClient
from agentarts.sdk.memory.inner.config import TextMessage, MemorySearchFilter
# 实例化客户端（API Key 通过环境变量 HUAWEICLOUD_SDK_MEMORY_API_KEY 配置，或直接传入）
with MemoryClient(api_key="your-memory-api-key") as client:
    # 1. 为特定用户创建独立会话（实现用户级数据隔离）
    session_data = client.create_memory_session(
        space_id="your-space-id",       # 控制台记忆库列表中获取
        actor_id="user-001",            # 用户唯一标识
        assistant_id="assistant-001"    # 绑定的智能体标识
    )
    session_id = session_data.id
    # 2. 写入对话消息
    client.add_messages(
        space_id="your-space-id",
        session_id=session_id,
        messages=[
            TextMessage(role="user", content="我对机器学习很感兴趣", actor_id="user-001"),
            TextMessage(role="assistant", content="机器学习是AI的核心分支...", actor_id="assistant-001")
        ]
    )
    # 3. 等待长期记忆生成后，按语义搜索相关记忆
    results = client.search_memories(
        space_id="your-space-id",
        filters=MemorySearchFilter(query="机器学习", top_k=3)
    )
```
**Session模式**，适合绑定特定用户/会话的交互场景（如对话式应用）：
```
from agentarts.sdk.memory.session import MemorySession
# 创建会话对象后，所有操作自动在该用户的上下文中执行
session = MemorySession(
    space_id="your-space-id",
    actor_id="user-001",
    assistant_id="assistant-001"
)
# 直接在会话上下文中操作，无需反复传入 space_id 和 session_id
session.add_messages(messages)
session.search_memories(filters=MemorySearchFilter(query="偏好", top_k=5))
```
**在LangGraph中集成记忆（Checkpoint持久化）：**
对于基于LangGraph构建的智能体，SDK还提供了AgentArtsMemorySessionSaver，可直接作为LangGraph的checkpointer使用，实现状态图的跨会话持久化：
```
from agentarts.sdk.integration.langgraph import AgentArtsMemorySessionSaver
checkpointer = AgentArtsMemorySessionSaver(
    space_id="your-space-id",
    region="cn-southwest-2"
)
# 在编译 LangGraph 工作流时传入 checkpointer
graph = workflow.compile(checkpointer=checkpointer)
# 使用 thread_id 隔离不同会话的状态
result = graph.invoke(
    {"messages": [HumanMessage(content="你好")]},
    config={"configurable": {"thread_id": session_id}}
)
```
表5记忆库隔离维度 
| 维度          | 标识符        | 隔离效果                     |
|:---|:---|:---|
| 空间（Space）   | space_id   | 不同记忆库之间完全物理隔离，适合多租户/多业务线 |
| 会话（Session） | session_id | 同一空间内不同会话的消息和记忆相互独立      |
| 用户（Actor）   | actor_id   | 同一空间内不同用户的偏好和记忆相互独立      |
   
多租户场景实践：为每个租户创建独立的Space（记忆库），租户内的每个用户通过actor_id区分，每次对话通过session_id隔离。这样从物理层面杜绝了跨租户数据串扰。
#### 扩展能力2：沙箱工具
**什么场景需要沙箱**
当智能体需要执行大模型生成的代码（如数据分析脚本、数学计算、文件处理）时，绝不能直接在主进程或宿主机上运行，一段恶意代码就可能导致系统被入侵。沙箱工具（代码解释器）提供了一个完全隔离的容器环境，智能体在其中执行代码，即使代码有问题也不会影响宿主环境。
沙箱支持四种操作：执行代码（execute_code）、执行命令（execute_command）、读取文件（read_files）、写入文件（write_files）。
**通过SDK创建沙箱**
```
from agentarts.sdk.tools import CodeInterpreter
client = CodeInterpreter(region="cn-southwest-2", auth_type="API_KEY")
code_interpreter = client.create_code_interpreter(
    name="my-sandbox",
    api_key_name="my-sandbox-key"
)
```
**在智能体中集成沙箱工具**
通过code_session上下文管理器连接沙箱，将其封装为LangChain工具：
```
import json
from agentarts.sdk.tools import code_session
from langchain_core.tools import tool
@tool
def execute_python_tool(code: str, description: str = "") -> str:
    """在安全沙箱中执行 Python 代码"""
    api_key = "your-sandbox-api-key"
    with code_session("cn-southwest-2", "your-sandbox-name", api_key=api_key) as code_client:
        response = code_client.invoke(
            operate_type="execute_code",
            api_key=api_key,
            arguments={
                "code": code,
                "language": "python",
                "clear_context": False  # 保持会话上下文，多次执行共享状态
            }
        )
    return json.dumps(response["result"])
```
将工具绑定到大模型后，模型在需要执行代码时会自动调用此工具：
```
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="deepseek-v4-flash", api_key="...", base_url="...")
llm_with_tools = llm.bind_tools([execute_python_tool])
```
#### 扩展能力3：网关
**什么场景需要网关**
当智能体需要调用企业内部系统（如ERP、CRM、OA）或外部第三方服务（如高德地图、天气接口）时，需要通过网关建立安全的连接通道。网关是连接AgentArts与外部系统的统一接口层，具备两大核心能力：
- REST → MCP自动转换：上传OpenAPI规范文件（3.0或Swagger 2.0），网关自动将企业存量REST API转换为大模型可感知的标准MCP工具，无需修改原有接口。
- MCP服务安全接入：为已有的MCP服务（如高德MCP）提供统一的入站认证和网络隔离。
**创建网关并关联Target**
在AgentArts平台"开发中心 \> 组件库 \> 网关"中可以创建网关。
网关本身只是一个统一的入站接入层，负责接收智能体发来的请求、完成入站身份认证，并将请求转发给后端服务，但它自己并不知道要把请求转发到哪里。Target就是告诉网关"请求该往哪里发"的配置单元，每个Target对应一个具体的下游服务端点。一个网关可以关联多个Target，智能体通过tools/list拿到的工具列表，实际上就是该网关下所有Target暴露出来的工具的聚合。
Target支持两种类型，覆盖企业接入外部系统的两类典型需求。
- 第一类是MCP Server类型，用于对接已有的MCP服务（如高德地图、企业自研MCP工具），只需填写MCP服务地址和传输方式（SSE或Streamable HTTP），再配置出站身份认证（如高德MCP需要在查询参数中传入key），网关会直接将请求转发给该MCP服务，无需任何协议转换。
- 第二类是REST API类型，专门解决企业存量REST接口的接入问题。上传一份符合OpenAPI 3.0或Swagger 2.0规范的OAS文件，网关会自动解析其中每个接口的operationId并将其转换为MCP工具，大模型可以像调用MCP工具一样直接调用这些REST接口，无需改动原有后端系统任何代码。需要注意的是，OAS文件的servers字段只能配置一个HTTPS URL，且所有operation必须有唯一的operationId，否则网关无法正确生成工具名称。
**在智能体中调用网关**
网关基于标准JSON-RPC 2.0协议通信，支持两个核心方法：
- tools/list：获取网关提供的所有可用工具
- tools/call：调用特定工具
以下为在LangGraph智能体中集成MCP网关的示例（以高德天气查询为例）：
```
import requests, json, re, uuid
def _call_mcp_gateway(tool_name: str, arguments: dict) -> str:
    """通用 MCP 网关调用函数"""
    gateway_url = "your-gateway-url"       # 网关详情页获取
    api_key = "your-gateway-api-key"       # 网关 URN 链接中获取
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json, text/event-stream",
        "mcp-session-id": str(uuid.uuid4()),
        "Authorization": f"Bearer {api_key}"
    }
    payload = {
        "jsonrpc": "2.0",
        "id": "call-tool",
        "method": "tools/call",
        "params": {"name": tool_name, "arguments": arguments}
    }
    response = requests.post(gateway_url, headers=headers, json=payload, verify=False, timeout=30)
    # 处理 SSE 或 JSON 格式响应
    raw_text = response.text
    if "data:" in raw_text:
        match = re.search(r'\{.*\}', raw_text, re.DOTALL)
        if match:
            parsed = json.loads(match.group())
            return json.dumps(parsed.get("result", parsed), ensure_ascii=False)
    return json.dumps(response.json().get("result", response.json()), ensure_ascii=False)
```
将网关调用封装为LangChain工具后，绑定到大模型即可实现自动调用：
```
from langchain_core.tools import tool
@tool
def gaode_get_weather(city: str) -> str:
    """查询指定城市的天气。Args: city: 城市名称"""
    return _call_mcp_gateway("gaodemap_maps_weather", {"city": city})
```
请求头注意：Streamable HTTP要求Accept头必须同时包含application/json和text/event-stream，否则服务器返回406。