轻量应用服务器部署 Agent 工具箱应用镜像

操作场景
Agent 工具箱提供常见的 Agent 开发、托管依赖能力。工具箱中同时包含 Agent 沙箱、LangGraph、Langfuse、Qdrant 向量数据库，其中：
Agent 沙箱为 Agent 提供安全隔离的 Runtime（基于 Docker）、代码执行隔离（基于 LLM Sandbox）、浏览器控制能力（基于 Playwright），后两项能力提供 MCP 以调用。
﻿LangGraph 为 Agent 开发提供可自主决策、调用工具、持续迭代的轻量级工作流框架。
﻿Langfuse 为 Agent 提供从开发、测试到生产监控的全生命周期管理能力。
﻿Qdrant 向量数据库为 Agent 提供 Memory 能力，帮助实现 Agent 长短期记忆。
整个镜像基于 Ubuntu 22.04 LTS 64bit 构建。
部署 Agent 工具箱应用镜像
1. 登录 轻量应用服务器控制台，在服务器页面单击新建。或直接访问 轻量应用服务器购买页。
2. 在轻量应用服务器购买页面，选择所需配置完成轻量应用服务器购买。
应用创建方式：选择使用应用模板 > Agent 工具箱，建议购买内存配置 2GB 以上，系统盘配置 20GB 以上的镜像，其他参数配置请参见 购买方式。
说明：
应用模板即应用镜像。
查看镜像说明详情请参见 基本概念。
地域：建议选择与您当前所在地最近的地域，可降低访问时延、提高访问速度。
可用区：默认勾选随机分配，也可自行选择可用区。
注意:
推荐您选择“随机分配”可用区，同一地域不同可用区间的实例可通过内网互访。同地域下不同可用区间可实现故障隔离，若您的应用对于容灾能力有极高的要求，则可将不同实例部署在同一地域下的不同可用区内。但需注意不同实例跨可用区通信的时延较同一可用区会有一定增加。
实例创建成功后，不支持更换可用区。
实例套餐：产品组合售卖方式（包含套餐类型、套餐规格），不同套餐包含不同规格 CPU、内存、SSD 云硬盘、带宽或峰值带宽、流量包的组合。详情请参见 基础套餐。
服务器名称：自定义实例名称，若不填则默认使用所选镜像名称。批量创建实例时，连续命名后缀数字自动升序。例如，填入名称为 LH，数量选择3，则创建的3个实例名称为 LH1、LH2、LH3。
登录方式：您可通过该项设置实例的登录方式。
密码登录：
自动生成密码：自动生成的密码将会以站内信方式发送。
自定义密码：自定义设置实例的登录密码。
SSH密钥：为 Linux 实例绑定 SSH 密钥。
扫码登录：设置后支持通过微信扫码免密码登录 Linux 实例。
自动续费：按照实际需求配置即可。
共享存储：您可以选择实例创建过程中购买共享存储超值资源包，将自动挂载默认对象存储桶，详情请参见 介绍。注意事项如下：
2023年10月17日之后购买的实例，可以选择是否购买共享存储超值资源包，之前购买的实例可在 对象存储控制台 购买套餐包。
当前地域下已有生效共享存储超值资源包，后续购买实例时您可直接选择一键挂载存储桶，不会再提供共享存储超值资源包。
目前仅应用模板和操作系统镜像中的 Linux 镜像支持共享存储。
时长：表示轻量应用服务器的使用时长。
说明:
您可勾选账户余额足够时，实例到期后按月自动续费，开启实例自动续费功能。当实例创建成功后，可参见 自动续费 修改已有的自动续费设置。
台数：表示需购买轻量应用服务器的数量。
3. 单击立即购买。
4. 核对配置信息后，单击提交订单，并根据页面提示完成支付。可用代金券使用说明请参见 代金券使用说明。
5. 完成支付后，即可进入 轻量应用服务器控制台 查收您的轻量应用服务器实例，实例通常需要 1-3 分钟完成创建。
6. 待实例创建完成后，在目标服务器实例右上角单击
﻿
 > 查看详情，进入该实例的详情页，即可进行后续操作。
使用 Agent 工具箱应用镜像
Agent 沙箱、 Langfuse、 Qdrant 向量数据库
说明：
Agent 工具箱应用镜像中 Agent 沙箱、 Langfuse、 Qdrant 向量数据库的部署和使用方式，建议参见下述镜像的部署实践教程进行操作。
﻿部署 Agent 沙箱应用模板﻿
﻿部署 Langfuse 应用模板﻿
﻿部署 Qdrant 向量数据库应用模板﻿
LangGraph
﻿LangGraph 是一个用于构建有状态、可循环的多智能体应用的库，让您可以用“图”的方式来组织和控制复杂的 LLM 工作流。
环境准备
在您的实例中运行以下命令安装依赖库。
pip install langgraph langchain langchain_openai
构建一个带工具的智能 Agent
我们将在实例中利用 Python 代码构建一个 Agent：它能决定是直接回答问题，还是先使用工具（例如搜索）来获取信息，然后再回答。
说明：
代码中将会用到硅基流动的 API 密钥。获取方式请参见 官方帮助文档。
第一步：定义状态
# 状态是整个图的“记忆”。它在节点之间传递和更新。
class AgentState(TypedDict):
    """
    定义了图的状态。
    `messages`: 一个消息列表，记录了对话的完整历史。
    `Annotated[..., operator.add]` 的作用是让新的消息附加到现有列表的末尾，而不是覆盖它。
    """
    messages: Annotated[List[BaseMessage], operator.add]
第二步：定义工具和节点
# 1. 定义我们的工具：一个简单的搜索函数
# 我将使用您环境中已有的 `concise_search` 函数
# 如果您没有这个函数，可以替换成任何其他函数，或mock一个
@tool
def concise_search(query: str):
    """在网上搜索给定的查询。"""
    # 在真实场景中，这里会调用真正的搜索引擎 API
    print(f"---正在搜索: {query}---")
     # 为了演示，我们返回一个固定的字符串
    if "langgraph" in query.lower():
        return f"关于 '{query}' 的搜索结果是：LangGraph 是一个用于构建有状态、可循环的多智能体应用的库。"
    else:
        return f"关于 '{query}' 的搜索结果未找到。"
﻿
tools = [concise_search]
﻿
# 将工具绑定到模型上，这样模型就知道它有哪些工具可用
# .bind_tools() 是一个方便的方法，它会将工具的定义以 OpenAI function-calling 的格式传给模型
model = ChatOpenAI(
    temperature=0,
    streaming=True,
    model="xxx",  # 硅基流动支持的模型
    base_url=SILICONFLOW_BASE_URL,
    api_key=SILICONFLOW_API_KEY
).bind_tools(tools)
﻿
﻿
# 2. 定义节点函数
# 节点1: 调用模型
def call_model(state: AgentState):
    """调用LLM，决定是直接回答还是调用工具"""
    messages = state['messages']
    response = model.invoke(messages)
    # 将模型的回复（可能是回答，也可能是工具调用请求）添加到状态中
    return {"messages": [response]}
﻿
# 节点2: 调用工具
def call_tool(state: AgentState):
    """节点2: 执行工具"""
    print("---节点: Tool---")
    # Agent 节点的输出 (AIMessage) 是列表中的最后一条消息
    last_message = state['messages'][-1]
﻿
    # 检查是否有工具调用请求
    if not last_message.tool_calls:
        # 如果没有工具调用，则不执行任何操作
        return {}
﻿
    # 我们只处理第一个工具调用
    tool_call = last_message.tool_calls[0]
    tool_name = tool_call['name']
﻿
    # 找到与名称匹配的工具
    action = None
    for t in tools:
        if t.name == tool_name:
            action = t
            break
﻿
    if action:
        # 执行工具
        result = action.invoke(tool_call['args'])
        # 将结果包装成 ToolMessage
        tool_message = ToolMessage(content=str(result), tool_call_id=tool_call['id'])
        return {"messages": [tool_message]}
    else:
        raise ValueError(f"未找到工具: {tool_name}")
﻿
第三步：定义条件边
# 这是一个函数，它根据当前状态决定接下来要执行哪个节点。
def should_continue(state: AgentState):
    """
    决策中心。
    - 如果模型的最新回复中有工具调用，则路由到 'call_tool' 节点。
    - 否则，结束流程。
    """
    print("---决策: Should Continue?---")
    last_message = state['messages'][-1]
﻿
    if last_message.tool_calls:
        print("决策: 是 -> 调用工具")
        return "call_tool"
    else:
        print("决策: 否 -> 结束")
        return "end"
第四步：组装图
# 定义一个新的图
workflow = StateGraph(AgentState)
﻿
# 添加节点
workflow.add_node("agent", call_model)
workflow.add_node("call_tool", call_tool)
﻿
# 设置入口点
workflow.set_entry_point("agent")
﻿
# 添加条件边
workflow.add_conditional_edges(
    "agent", # 从 agent 节点出发
    should_continue, # 使用 should_continue 函数做判断
    {
        "call_tool": "call_tool", # 如果返回 "call_tool"，则去 call_tool 节点
        "end": END               # 如果返回 "end"，则结束
    }
)
﻿
# 添加从工具节点返回到 Agent 节点的普通边
# 这样，工具执行完后，流程会回到 Agent，让它根据工具结果继续思考
workflow.add_edge('call_tool', 'agent')
﻿
# 编译成可执行的应用
app = workflow.compile()
第五步：运行和测试
# 定义初始输入
inputs = {"messages": [HumanMessage(content="LangGraph 是什么?")]}
print("\\n---开始运行 Agent---")
﻿
try:
  # 使用 .stream() 方法可以实时查看每一步的输出
  for output in app.stream(inputs, {"recursion_limit": 5}):  # 设置递归限制以防无限循环
    # stream() 会返回每个节点的名字和它的输出
    for key, value in output.items():
      print(f"\\n## 节点 '{key}' 的输出 ##")
      # `value` 是该节点更新后的完整状态
      # 我们只打印最新的消息以保持简洁
      latest_message = value['messages'][-1]
      print(latest_message)
﻿
      print("\\n" + "="*60)
      print("✅ Agent 运行成功完成！")
      print("🔗 使用的是硅基流动 API")
﻿
    except Exception as e:
      print(f"\\n❌ 运行出错: {e}")
      print("请检查：")
      print("  - 硅基流动 API Key 是否正确设置")
      print("  - 网络连接是否正常")
      print("  - API 配额是否充足")
预期结果
完成上述步骤后，即可运行代码，查看运行效果。示例效果如下：
✅ 硅基流动 API 配置成功。
🔗 API 地址: https://api.siliconflow.cn/v1
🔑 API Key: 您的API Key
📋 支持的功能: 工具调用、流式输出
﻿
============================================================
🚀 硅基流动 LangGraph Agent 演示
============================================================
💬 用户问题: LangGraph 是什么?
﻿
---开始运行 Agent---
---节点: Agent---
---决策: Should Continue?---
决策: 是 -> 调用工具
﻿
## 节点 'agent' 的输出 ##
content='' additional_kwargs={'tool_calls': [{'index': 0, 'id': '0198cc8756fc48650fa67087fad33cc4', 'function': {'arguments': ' {"query": "LangGraph 是什么"}', 'name': 'concise_search'}, 'type': 'function'}]} response_metadata={'finish_reason': 'tool_calls', 'model_name': 'Qwen/Qwen2.5-7B-Instruct'} id='run--44e8eaef-8970-47b6-ac31-7219d8e137ea-0' tool_calls=[{'name': 'concise_search', 'args': {'query': 'LangGraph 是什么'}, 'id': '0198cc8756fc48650fa67087fad33cc4', 'type': 'tool_call'}] usage_metadata={'input_tokens': 2148, 'output_tokens': 65, 'total_tokens': 2213, 'input_token_details': {}, 'output_token_details': {}}
---节点: Tool---
---正在搜索: LangGraph 是什么---
﻿
## 节点 'call_tool' 的输出 ##
content="关于 'LangGraph 是什么' 的搜索结果是：LangGraph 是一个用于构建有状态、可循环的多智能体应用的库。" tool_call_id='0198cc8756fc48650fa67087fad33cc4'
---节点: Agent---
---决策: Should Continue?---
决策: 否 -> 结束
﻿
## 节点 'agent' 的输出 ##
content='LangGraph 是一个用于构建有状态、可循环的多智能体应用的库。' additional_kwargs={} response_metadata={'finish_reason': 'stop', 'model_name': 'Qwen/Qwen2.5-7B-Instruct'} id='run--05896a6b-16fb-4119-a19e-9f9602f27ff2-0' usage_metadata={'input_tokens': 5313, 'output_tokens': 209, 'total_tokens': 5522, 'input_token_details': {}, 'output_token_details': {}}
﻿
============================================================
✅ Agent 运行成功完成！
完整代码如下：
import os
from typing import TypedDict, Annotated, List
import operator
﻿
# --- 核心 LangChain 和 LangGraph 库 ---
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, ToolMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END
﻿
# --- 0. 环境设置 ---
# API 配置 : 这里以国内 硅基流动 为例
SILICONFLOW_API_KEY = "sk-nmg.."
SILICONFLOW_BASE_URL = "https://api.siliconflow.cn/v1"
﻿
# 设置环境变量以兼容 OpenAI 客户端
os.environ["OPENAI_API_KEY"] = SILICONFLOW_API_KEY
os.environ["OPENAI_BASE_URL"] = SILICONFLOW_BASE_URL
﻿
# 检查配置
print("✅ 硅基流动 API 配置成功。")
print(f"🔗 API 地址: {SILICONFLOW_BASE_URL}")
print(f"🔑 API Key: {SILICONFLOW_API_KEY[:20]}...")
﻿
﻿
# --- 1. 定义状态 (State) ---
# 状态是整个图的“记忆”。它在节点之间传递和更新。
class AgentState(TypedDict):
    """
    定义了图的状态。
    `messages`: 一个消息列表，记录了对话的完整历史。
    `Annotated[..., operator.add]` 的作用是让新的消息附加到现有列表的末尾，而不是覆盖它。
    """
    messages: Annotated[List[BaseMessage], operator.add]
﻿
﻿
# --- 2. 定义工具和节点 (Tools & Nodes) ---
﻿
# 2.1 定义工具
# 工具是 Agent 可以使用的能力。
@tool
def concise_search(query: str):
    """在网上搜索给定的查询，并返回简洁的结果。"""
    # 这是一个模拟的搜索工具。在真实应用中，这里会调用一个真正的搜索 API。
    print(f"---正在搜索: {query}---")
    # 为了演示，我们返回一个固定的字符串
    if "langgraph" in query.lower():
        return f"关于 '{query}' 的搜索结果是：LangGraph 是一个用于构建有状态、可循环的多智能体应用的库。"
    else:
        return f"关于 '{query}' 的搜索结果未找到。"
﻿
﻿
tools = [concise_search]
﻿
# 2.2 定义模型
# 使用硅基流动的模型，支持工具调用
# 硅基流动支持多种模型，这里使用 Qwen2.5-7B-Instruct，它支持工具调用
model = ChatOpenAI(
    temperature=0,
    streaming=True,
    model="Qwen/Qwen2.5-7B-Instruct",  # 硅基流动支持的模型
    base_url=SILICONFLOW_BASE_URL,
    api_key=SILICONFLOW_API_KEY
).bind_tools(tools)
﻿
print(f"🤖 使用模型: Qwen/Qwen2.5-7B-Instruct (硅基流动)")
print("📋 支持的功能: 工具调用、流式输出")
﻿
﻿
# 2.3 定义节点函数
# 节点是图中的“工作单元”，每个节点都是一个函数或 Runnable。
﻿
def call_model(state: AgentState):
    """节点1: 调用 LLM"""
    print("---节点: Agent---")
    messages = state['messages']
    response = model.invoke(messages)
    # 返回一个字典，其键与 AgentState 中的键匹配。
    # LangGraph 会自动用这个返回值更新状态。
    return {"messages": [response]}
﻿
﻿
def call_tool(state: AgentState):
    """节点2: 执行工具"""
    print("---节点: Tool---")
    # Agent 节点的输出 (AIMessage) 是列表中的最后一条消息
    last_message = state['messages'][-1]
﻿
    # 检查是否有工具调用请求
    if not last_message.tool_calls:
        # 如果没有工具调用，则不执行任何操作
        return {}
﻿
    # 我们只处理第一个工具调用
    tool_call = last_message.tool_calls[0]
    tool_name = tool_call['name']
﻿
    # 找到与名称匹配的工具
    action = None
    for t in tools:
        if t.name == tool_name:
            action = t
            break
﻿
    if action:
        # 执行工具
        result = action.invoke(tool_call['args'])
        # 将结果包装成 ToolMessage
        tool_message = ToolMessage(content=str(result), tool_call_id=tool_call['id'])
        return {"messages": [tool_message]}
    else:
        raise ValueError(f"未找到工具: {tool_name}")
﻿
﻿
# --- 3 定义条件边 (Conditional Edge) ---
# 这是一个函数，它根据当前状态决定接下来要执行哪个节点。
def should_continue(state: AgentState):
    """
    决策中心。
    - 如果模型的最新回复中有工具调用，则路由到 'call_tool' 节点。
    - 否则，结束流程。
    """
    print("---决策: Should Continue?---")
    last_message = state['messages'][-1]
﻿
    if last_message.tool_calls:
        print("决策: 是 -> 调用工具")
        return "call_tool"
    else:
        print("决策: 否 -> 结束")
        return "end"
﻿
﻿
# --- 4. 组装图 (Assemble the Graph) ---
# 创建一个新的状态图实例
workflow = StateGraph(AgentState)
﻿
# 添加节点
workflow.add_node("agent", call_model)
workflow.add_node("call_tool", call_tool)
﻿
# 设置图的入口点
workflow.set_entry_point("agent")
﻿
# 添加条件边，这是图的逻辑核心
workflow.add_conditional_edges(
    "agent",  # 从 "agent" 节点出发
    should_continue,  # 使用 "should_continue" 函数做决策
    {
        "call_tool": "call_tool",  # 如果决策函数返回 "call_tool"，则流向 "call_tool" 节点
        "end": END  # 如果返回 "end"，则结束 (END 是一个特殊标识符)
    }
)
﻿
# 添加一条普通边
# 从 "call_tool" 节点执行完毕后，总是返回到 "agent" 节点，形成循环
workflow.add_edge('call_tool', 'agent')
﻿
# 编译图，生成一个可执行的 LangChain Runnable
app = workflow.compile()
﻿
# --- 6. 运行和测试 ---
def run_langgraph_demo():
    """运行硅基流动 LangGraph 演示"""
    print("\\n" + "="*60)
    print("🚀 硅基流动 LangGraph Agent 演示")
    print("="*60)
﻿
    # 定义初始输入
    inputs = {"messages": [HumanMessage(content="LangGraph 是什么?")]}
﻿
    print("💬 用户问题: LangGraph 是什么?")
    print("\\n---开始运行 Agent---")
﻿
    try:
        # 使用 .stream() 方法可以实时查看每一步的输出
        for output in app.stream(inputs, {"recursion_limit": 5}):  # 设置递归限制以防无限循环
            # stream() 会返回每个节点的名字和它的输出
            for key, value in output.items():
                print(f"\\n## 节点 '{key}' 的输出 ##")
                # `value` 是该节点更新后的完整状态
                # 我们只打印最新的消息以保持简洁
                latest_message = value['messages'][-1]
                print(latest_message)
﻿
        print("\\n" + "="*60)
        print("✅ Agent 运行成功完成！")
        print("🔗 使用的是硅基流动 API")
﻿
    except Exception as e:
        print(f"\\n❌ 运行出错: {e}")
        print("请检查：")
        print("  - 硅基流动 API Key 是否正确设置")
        print("  - 网络连接是否正常")
        print("  - API 配额是否充足")
﻿
if __name__ == "__main__":
    run_langgraph_demo()
﻿
﻿
如需了解 LangGraph 的更多信息，请参见 LangGraph 官方文档。
部署 Agent 工具箱应用镜像

本页目录：

操作场景

部署 Agent 工具箱应用镜像

使用 Agent 工具箱应用镜像

Agent 沙箱、 Langfuse、 Qdrant 向量数据库

LangGraph

环境准备

构建一个带工具的智能 Agent

第一步：定义状态

第二步：定义工具和节点

第三步：定义条件边

第四步：组装图

第五步：运行和测试

预期结果

完整代码如下：
