高代码智能体开发与云端部署技巧

高代码开发的两种构建模式

在开始之前，需要先明确AgentArts高代码开发支持的两种构建模式，根据团队现状选择合适路径：

选择原则：如果团队已有完整的Agent代码，只需要一个安全的云端运行环境，选择“代码云上托管”；如果需要集成AgentArts云端的跨会话记忆、安全沙箱执行能力，选择“从零开始构建”。

搭建运行时：从本地代码到云端托管

运行时是什么

智能体运行时是智能体的云端执行底座。它基于容器化技术实现进程级隔离，支持基于QPS/并发数的自动扩缩容，并深度集成IAM实现细粒度的身份认证和网络管控。开发者只需关注业务逻辑，底层基础设施的运维完全由平台托管。

关键技术约束

在开始编码前，必须了解以下硬约束，否则部署后会遇到运行失败：

代码结构

任何高代码智能体都需要通过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)

请求上下文：RequestContext包含user_id、session_id、request_id、workload_access_token等属性，由平台在调用时自动注入。在调用栈更深层的代码中，可通过全局上下文AgentArtsRuntimeContext.get_session_id()在任意位置获取，无需层层传参。

实践案例：从初始化项目到云端托管部署

华为云AgentArts官方文档

基础示例：创建基础对话智能体

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

扩展能力1：记忆库

什么场景需要记忆库

当智能体需要跨会话记住用户的偏好、历史对话或业务状态时（如智能客服记住客户之前的投诉、个人助手记住用户的饮食偏好、问诊助手记住患者的病史），就需要接入记忆库。

记忆库提供短期记忆（当前会话的临时上下文，7~365天过期）和长期记忆（持久化存储到向量数据库，跨会话可检索）的双层架构，支持按空间、会话、用户三个维度进行数据隔离。

记忆库的工作原理

1. 用户对话 → Agent将对话内容写入短期记忆。
2. 触发条件满足（空闲时间/Token累计量/消息数任一达到阈值） → 短期记忆传递至记忆提取模块。
3. 记忆提取执行三步处理：Embedding（向量嵌入） → Consolidation（巩固降维） → Extraction（关键信息抽取）。
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}}
)

多租户场景实践：为每个租户创建独立的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。