运行时智能体离线评估实践（流式）

准备环境

本地环境：

• 操作系统：推荐使用Linux Ubuntu 22.04 ARM64，服务器可访问公网。

  本文命令如无特殊说明，均以Ubuntu为例。为保证镜像构建与云端部署的兼容性，建议使用Ubuntu系统。

  因此，为了减少部署失败和额外排查成本，建议直接使用Ubuntu 22.04 ARM64制作AgentArts镜像。

  若使用CentOS、EulerOS等非Ubuntu系统，请将以下命令中的apt替换为yum或dnf，并自行确认Docker 18.06+、Python 3.10+；本文不逐条展开各发行版安装差异。

  

  需使用ARM64系统制作的Agent镜像，使用X86系统制作的镜像在调用智能体运行时会调用失败。

  可以使用华为云ECS服务购买服务器，购买时需要选择鲲鹏架构（ARM64），系统选择Ubuntu。

  

• 安装Python：请确保Python 3.10及以上版本已安装。

  大多数Linux发行版（如Ubuntu）都预装了Python，您可以先通过python3 --version检查。如未安装，可以使用如下命令安装：

  sudo apt update
  sudo apt install python3

  非Ubuntu/Debian系统请使用yum或dnf安装Python 3.10+及pip。

• 安装Docker：请确保Docker 18.06及以上版本已安装。如未安装，可以使用如下命令安装：

  # 查询 Docker 版本
  docker --version
  
  # 安装Docker
  sudo apt update
  sudo apt install docker.io

  安装完成后，建议立即配置国内镜像加速器。在后续执行agentarts launch构建镜像时，Docker需要拉取python:3.10-slim基础镜像，默认从docker.io官方仓库下载，可能超时失败。提前配置加速器可避免此问题。

  执行以下命令创建Docker配置目录：

  sudo mkdir -p /etc/docker

  执行以下命令配置国内镜像源（直接复制并回车执行）：

  sudo tee /etc/docker/daemon.json <<-'EOF'
  {
    "registry-mirrors":[
      "https://docker.m.daocloud.net",
      "https://dockerproxy.net",
      "https://mirror.baidubce.com"
    ]
  }
  EOF

  依次执行以下命令重启Docker服务，使配置生效：

  sudo systemctl daemon-reload
  sudo systemctl restart docker

  

  华为云SWR基础版不支持OCI镜像格式。如果您使用的是Docker 27及以上版本，由于该版本默认使用OCI镜像格式推送到SWR时会报错，可以通过设置以下环境变量来关闭OCI支持。

  export DOCKER_BUILDKIT=0
  # 或者
  export BUILDKIT_USE_OCI_MEDIA_TYPES=0

• 执行以下命令安装运行时SDK（建议在Python虚拟环境中安装，以避免与系统包产生冲突）。

  # 安装依赖包
  apt install python3.12-venv
  
  # 创建并激活虚拟环境 (linux) 
  python3 -m venv venv
  source venv/bin/activate
  
  # 安装sdk
  pip install agentarts-sdk

  若配置镜像加速器，后续在执行agentarts launch命令时，仍无法拉取 python:3.10-slim，可参考本文结尾的“常见问题”中的“直接修改Dockerfile基础镜像地址”。

  

  • 执行pip install agentarts-sdk命令下载缓慢、超时可以更改为使用如下命令

    pip install agentarts-sdk -i https://repo.huaweicloud.com/repository/pypi/simple --trusted-host repo.huaweicloud.com

  • 如果系统缺少python3-venv包，导致无法创建虚拟环境，请按照命令回显提示安装python3-venv包。

    

• 执行以下命令配置华为云凭证，获取华为云凭证请参考认证鉴权。

  export HUAWEICLOUD_SDK_AK="your-access-key"
  export HUAWEICLOUD_SDK_SK="your-secret-key"

准备智能体

1. 创建项目目录。

   在终端执行以下命令，创建项目目录并进入：

   mkdir my-first-agent && cd my-first-agent

2. 配置环境变量。

   在项目根目录下创建.env文件，用于存放模型调用所需的配置信息。

   执行vi .env命令创建.env文件，写入以下内容，并将model_api_key替换为您实际的模型API Key。

   # .env
   
   # 必填，您的华为云 MaaS API 密钥
   MODEL_API_KEY=model_api_key
   
   # 模型名称
   MODEL_NAME=deepseek-v4.1-flash
   
   # 模型服务地址
   MODEL_URL=https://api.modelarts-maas.com/openai/v1
   
   # Agent 服务监听端口（可选，默认 8080）
   AGENT_RUN_PORT=8080

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

   pip install -U langchain langgraph langchain-openai -i https://pypi.tuna.tsinghua.edu.cn/simple --trusted-host pypi.tuna.tsinghua.edu.cn

4. 编写流式输出Agent核心逻辑。
   1. 在项目根目录执行vi agent.py命令，写入以下内容并保存。

      # agent.py
      import os
      from dotenv import load_dotenv
      from langchain_openai import ChatOpenAI
      from langchain_core.messages import HumanMessage, SystemMessage
      
      # 加载 .env 文件中的环境变量
      load_dotenv()
      
      class MyAgent:
          """基础对话智能体，同时支持普通调用 + 同步流式 + 异步流式"""
      
          def __init__(self):
              self.llm = ChatOpenAI(
                  model=os.getenv("MODEL_NAME", "deepseek-v4.1-flash"),
                  openai_api_key=os.getenv("MODEL_API_KEY"),
                  openai_api_base=os.getenv(
                      "MODEL_URL", "https://api.modelarts-maas.com/openai/v1"
                  ),
                  streaming=True,
              )
              self.system_prompt = (
                  "您是轻量化中英翻译助手，专注日常办公、学习交流场景，仅提供中英文精准翻译服务。"
              )
      
          def chat(self, message: str) -> str:
              """接收用户消息，调用大模型，返回回复内容"""
              messages = [
                  SystemMessage(content=self.system_prompt),
                  HumanMessage(content=message),
              ]
              response = self.llm.invoke(messages)
              return response.content
      
          def chat_stream(self, message: str):
              """同步流式对话生成器"""
              messages = [
                  SystemMessage(content=self.system_prompt),
                  HumanMessage(content=message),
              ]
              for chunk in self.llm.stream(messages):
                  if chunk.content:
                      yield chunk.content
      
          async def achat_stream(self, message: str):
              """异步流式对话生成器，专门给AgentArts异步handler使用"""
              messages = [
                  SystemMessage(content=self.system_prompt),
                  HumanMessage(content=message),
              ]
              async for chunk in self.llm.astream(messages):
                  if chunk.content:
                      yield chunk.content
      
      # 创建 Agent 单例，供 app.py 引用
      agent = MyAgent()

   2. 验证Agent逻辑是否正常。

      在终端执行以下命令，直接调用agent.py进行快速验证，确认模型调用链路没有问题：

      python3 -c "from agent import agent; print(agent.chat('今天天气怎么样？'))"

      

      如果模型调用正常，终端会输出模型的回复内容。

      如果此步骤报错，请优先排查以下问题：

      • .env文件中的MODEL_API_KEY是否已替换为实际值。
      • 当前网络环境是否可以访问api.modelarts-maas.com。

5. 封装为HTTP服务。

   AgentArts运行时通过标准HTTP接口与您的Agent通信。本步骤使用AgentArts运行时SDK将Agent逻辑封装为符合平台规范的HTTP服务。

   在项目根目录执行vi app.py命令，写入以下内容并保存。

   # app.py
   import os
   from typing import Dict, Any, AsyncGenerator
   from dotenv import load_dotenv
   from agentarts.sdk import AgentArtsRuntimeApp, RequestContext
   from agent import agent
   
   load_dotenv()
   
   app = AgentArtsRuntimeApp()
   
   @app.entrypoint
   async def handler(payload: Dict[str, Any], context: RequestContext = None) -> AsyncGenerator[Dict[str, Any], None]:
       """AgentArts高代码 异步流式入口，支持yield分片"""
       message = payload.get("message", "")
       if not message:
           yield {"response": "请求体中缺少 message 字段", "status": "error"}
           return
   
       try:
           async for chunk_text in agent.achat_stream(message):
               yield {
                   "response": chunk_text,
                   "status": "streaming"
               }
           # 结束标记
           yield {"response": "", "status": "done"}
       except Exception as e:
           yield {"response": f"执行出错: {str(e)}", "status": "error"}
   
   
   if __name__ == "__main__":
       port = int(os.getenv("AGENT_RUN_PORT", 8080))
       app.run(host="0.0.0.0", port=port)

   

   • AgentArtsRuntimeApp是AgentArts SDK提供的运行时封装类，负责处理平台与Agent之间的通信协议，以及健康检查接口的注册。
   • @app.entrypoint装饰器将handler函数注册为平台的HTTP调用入口，平台每次调用Agent时都会触发此函数。

本地验证

在部署到云端之前，先在本地确认服务启动正常、接口响应符合预期。

通过SDK一键部署

SDK方式会自动完成镜像构建、推送SWR、部署运行时的全部操作，适合快速验证场景。

1. 参考前置条件安装SDK，并配置华为云凭证。
2. 准备依赖文件。

   在项目根目录执行vi requirements.txt命令，写入以下内容并保存。

   # ================================================================================
   # 指定国内 PyPI 镜像源，加速镜像构建时的依赖下载
   # ================================================================================
   --index-url https://repo.huaweicloud.com/repository/pypi/simple
   --trusted-host repo.huaweicloud.com
   
   # ================================================================================
   # 核心依赖
   # ================================================================================
   
   # LangChain：LLM 应用开发工具链
   langchain>=0.3.0
   langchain-core>=0.3.0
   langchain-openai>=0.2.0
   
   # 环境变量加载
   python-dotenv>=1.0.0
   
   # AgentArts SDK：运行时封装与 HTTP 服务标准化
   agentarts-sdk

3. 初始化部署配置。
   执行如下命令配置智能体。

   agentarts configure --entrypoint app:app

   执行后按照操作指引进行配置。

   表2 配置参数

   参数	说明
   Agent name	智能体名称。 以小写字母开头，以小写字母或数字结尾，可以包含小写字母、数字和中划线。
   Region	服务部署区域。 cn-southwest-2，仅支持此区域。
   Dependency file	依赖文件默认为requirements.txt。
   SWR Organization	镜像组织名称（建议自定义的镜像组织名，可以在SWR服务控制台贵阳一region创建）。
   SWR Repository	不填写（推荐）。 SWR Repository项直接回车留空，工具会自动生成仓库名称，部署agentarts launch时自动创建镜像仓库，避免仓库不存在报错。 如需手动指定，请填写SWR中已存在的镜像仓库名称；若尚未创建仓库，请留空，切勿填写不存在的仓库名。

   

4. 部署智能体。

   agentarts launch

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

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

5. 验证构建的镜像是否能正常响应。

   agentarts invoke --agent translation-assistant-stream '{"message": "帮我翻译一下 什么是AgentArts？"}'