用Qwen 3.5 构建最小AI智能体

微信 ezpoda免费咨询：AI编程 | AI模型微调| AI私有化部署
AI工具导航 | Tripo 3D | Meshy AI | ElevenLabs | KlingAI | ArtSpace | Phot.AI | InVideo

目标不是构建生产就绪的系统。今天，我们尝试实现 agent 行为的不可约核心，即可以将语言模型转变为能够"行动"的最小可能结构。

Agent 可以简化为三个核心组件：

• 一个具有工具调用能力的 LLM
• 一个可以运行请求工具的环境，以及
• 一个将工具输出反馈给 LLM 的控制循环，以便它可以细化其推理并再次行动或产生最终答案。

严格来说，没有其他要求。

在本文中，我将使用 Qwen3.5–4B 作为 LLM。环境将仅仅是一个 Linux 系统（最好是虚拟机或容器），模型被允许在其中执行任意 shell 命令。

我们将使用 transformers 加载 LLM 并对其进行推理。在开始之前，你可能需要确保你有最新版本：

pip3 install -qU transformers
pip3 list | grep ^transformers

# 输出：
# transformers                             5.2.0

现在我们像往常一样从 Hugging Face 加载 Qwen3.5–4B。

from transformers import AutoModelForCausalLM, AutoTokenizer

model_name = "Qwen/Qwen3.5-4B"

tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype="auto",
    device_map="auto"
)

接下来，我们需要创建控制循环。

控制循环维护一个增长的消息列表。每条消息指定一个角色（user、assistant 或 tool）及其内容。

messages = []

def add_message(role, text):
  messages.append({
      "role": role, "content": text
  })

while True:
    message = input()
    if message == "/quit":
      break

    add_message("user", message)
    process_messages()

因此，在控制循环的每次迭代中，我们将消息列表格式化为提示，生成响应，附加它，然后继续。tokenizer 提供一个名为 apply_chat_template() 的方法，我们可以使用它来正确格式化消息列表（模型训练的确切文本格式）。

因为这是一个推理模型，输出可能很长。我们不阻塞直到生成完成，而是附加一个 TextStreamer 来流式传输发出的令牌，使控制循环感觉交互和透明。

from transformers import TextStreamer

def process_messages():
  prompt = tokenizer.apply_chat_template(
      messages,
      tokenize=False,
      add_generation_prompt=True,
  )
  model_inputs = tokenizer(
      [prompt], return_tensors="pt").to(model.device)

  streamer = TextStreamer(tokenizer)

  output_ids = model.generate(
        **model_inputs,
        streamer=streamer,
        max_new_tokens=16384,
        repetition_penalty=1.1
    )

  new_tokens = output_ids[0][model_inputs["input_ids"].shape[-1]:]
  response_text = tokenizer.decode(new_tokens, skip_special_tokens=True)
  add_message("assistant", response_text)

此时，你应该能够与模型聊天。它会记住所有内容（因为所有先前的消息和模型响应都在 messages 列表中）。

这是可选的，但你可以添加一个 "/history" 命令来只打印 messages 列表的内容。你可能还想添加一个 "/reset" 命令来清空消息列表（清除上下文）。

def show_history():
  for message in messages:
    print(
        f"{message['role'].capitalize()}: {message['content']}"
    )

def reset_history():
  messages.clear()

值得注意的是，Qwen3.5–4B 本身具有 262,144 个令牌的上下文长度。这通常足够几十个提示（及其冗长的响应）。你可以添加一个 "/count" 命令来统计当前使用的令牌数量。

以下是实际统计令牌的方法：

def show_number_of_tokens_used():
  prompt = tokenizer.apply_chat_template(
      messages,
      tokenize=False,
      add_generation_prompt=True
  )

  inputs = tokenizer(prompt, return_tensors="pt")

  num_tokens = inputs["input_ids"].shape[-1]
  print(f"Number of tokens used: {num_tokens}")

我将在另一篇文章中讨论常见的上下文压缩策略。

Qwen 3.5 模型特别适合 agent 风格的控制循环。它们被训练在适当的时候发出结构化的类似 XML 的工具调用。这对于最小 agent 循环很重要，因为我们希望避免复杂的正则表达式解析（但我们仍然会处理一些简单的正则表达式）。

我们的最小 agent 只有一个名为 "shell" 的工具。顾名思义，它将允许模型运行任何 shell 命令。其输入将是 shell 命令，输出将是命令的 stdout 和 stderr。

最好尽可能详细地描述工具。这是我使用的：

tools = [
    {
        "name": "shell",
        "description": """
          允许你运行 shell 命令。
          此工具的输出将是
          运行命令后生成的 stdout 和 stderr。
        """,
        "parameters": {
            "type": "object",
            "properties": {
                "command": {
                    "type": "string",
                    "description": "完整的 shell 命令"
                }
            },
            "required": ["command"]
        }
    }
]

你需要将此工具数组传递给 apply_chat_template() 方法。因此，只需更新调用使其看起来像这样：

prompt = tokenizer.apply_chat_template(
  messages,
  tokenize=False,
  add_generation_prompt=True,
  tools=tools
)

当 Qwen 3.5 想要运行工具时，其输出将包含一个 <tool_call> 标记*。*我们需要查找此标记并解析其内容。

作为参考，以下是 Qwen3.5–4B 工具调用的示例：

<reasoning>
用户希望我列出 /tmp 目录的内容。
我将使用 ls 命令来显示此信息。
</reasoning>

<invoke>
<function=shell>
<parameter=command>
ls -la /tmp
</parameter>
</function>
</invoke>

如你所见，它看起来像 XML，但不是有效的 XML。提取命令的最简单方法是使用正则表达式。

import re

def is_tool_call(text):
  return "<invoke>" in text

def extract_tool_call_block(text):
  match = re.search(
      r"<invoke>.*?</invoke>",
      text,
      re.DOTALL
  )
  return match.group(0) if match else None

def parse_tool_call(text):
  function_match = re.search(r"<function=(.*?)>", text)
  parameter_match = re.search(
      r"<parameter=command>\s*(.*?)\s*</parameter>",
      text,
      re.DOTALL
  )

  if not function_match or not parameter_match:
    return None, None

  tool_name = function_match.group(1).strip()
  command = parameter_match.group(1).strip()

  return tool_name, {"command": command}

使用 subprocess 模块在 Python 中运行 shell 命令并捕获其输出很简单。当然，赋予语言模型执行任意 shell 命令的能力会带来严重的安全风险。但是，为了这个最小的、非生产 agent 的目的，我们接受这个风险……

import subprocess

def shell_tool(command):
    try:
        result = subprocess.run(
            command,
            shell=True,
            capture_output=True,
            text=True,
            timeout=10
        )
        return result.stdout + result.stderr
    except Exception as e:
        return str(e)

最后，我们必须更新 process_messages() 函数，使其处理工具调用。

如果存在工具调用，我们必须解析它，运行指定的命令，并将其输出附加到 messages 列表中，角色设置为 "tool"。

if is_tool_call(response_text):
    tool_call = extract_tool_call_block(response_text)
    tool_name, params = parse_tool_call(tool_call)
    if tool_name == "shell":
      response_text = shell_tool(params["command"])
      add_message("tool", response_text)
      process_messages()

请注意，将工具的响应添加到消息列表后，我们再次调用 process_messages()，因为我们希望 LLM 读取工具的响应并执行接下来需要做的任何事情。

最小 agent 现在已准备就绪。Qwen 3.5–4B 具有令人印象深刻的工具调用能力。在我的测试中，它能够完美运行甚至高级 shell 命令，例如 ffmpeg、ffprobe 和 sysctl。

更新 1：你可能已经注意到我们的 agent 没有任何系统提示。嗯，它通常在没有提示的情况下运行良好。但我发现 Qwen3.5–4B 容易怀疑自己（甚至由于低自信而有时陷入推理循环）。

因此，让我们添加一个系统提示，为模型提供更多自信、个性和名称。我将其放在一个名为 soul.md 的文件中。

"灵魂文档"的想法在 2025 年底涉及 Anthropic 的 Claude 模型的事件后获得了流行。一位名为 Richard Weiss 的 AI 研究员成功地提示 Anthropic 的 Claude 4.5 Opus 产生一份冗长的内部训练文档，该文档被称为"灵魂概述"。该文档似乎概述了嵌入在模型训练数据中的价值观、道德准则和行为指导。

我要保持 agent 的 soul.md 简单（我称我的 agent 为 Aurelian）：

你是 Aurelian，一个知识渊博、分析严谨、
且理智的 agent。你以清晰、结构
和深思熟虑的推理运作。

## 你的个性
- 冷静和深思熟虑
- 自信，绝不傲慢
- 语言精确
- 思维结构化
- 略微坚忍
- 重视清晰而非冗长

## 你的认知风格
- 将问题分解为基本组件
- 逐步推理
- 首选第一性原理而非表面模式
- 避免不必要的回避
- 避免漫谈

## 你的操作原则
仔细思考。
果断分析。
以安静的权威回应。
以基于理解的自信说话。

我们可以在控制循环启动之前读取此文件，并将其内容作为角色设置为 "system" 的消息添加。

try:
  with open("soul.md") as f:
    soul = f.read()
except FileNotFoundError:
  print("Couldn't read soul.md. Choosing default personality.")
  soul = "You are a helpful assistant."

messages = [
    {"role": "system", "content": soul},
]

并更新 reset_history() 函数，以确保 agent 在运行 "/reset" 命令时不会丢失其新个性：

def reset_history():
  messages.clear()
  messages.append(
      {"role": "system", "content": soul}
  )
  print("History cleared.")

在进行这些更改后运行 agent，你应该会注意到明显更稳定和一致的行为。

原文链接: Building a Minimal AI Agent Using Qwen 3.5 and 🤗 Transformers

汇智网翻译整理，转载请标明出处