高代码智能体开发与云端部署技巧-j9九游会登录

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

在开始之前，需要先明确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。