使用开源LLMS构建具有工具执行和结构化多智能体推理的层次规划AI代理的编码实现

本文教程中，我们使用开源指令模型构建了一个层次规划智能体。我们设计了一个包含规划智能体、执行智能体和聚合智能体的结构化多智能体架构，其中每个组件在解决复杂任务中承担专业角色。我们使用规划智能体将高级目标分解成可执行步骤，执行智能体使用推理或Python工具执行这些步骤，聚合智能体则将结果综合成一个连贯的最终响应。通过整合工具使用、结构化规划和迭代执行，我们创建了一个完全自主的智能体系统，展示了现代智能体如何以可扩展和模块化方式推理、规划和行动。

# 以下是代码示例，可复制到本地环境运行
!pip -q install -U transformers accelerate bitsandbytes sentencepiece

import json
import re
import io
import contextlib
from dataclasses import dataclass
from typing import Any, Dict, List, Optional

import torch
from transformers import AutoTokenizer, AutoModelForCausalLM

# 设置模型ID和设备
MODEL_ID = "Qwen/Qwen2.5-1.5B-Instruct"
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
print("Device:", DEVICE)

# 加载模型和分词器
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, use_fast=True)
model = None
if DEVICE == "cuda":
    try:
        model = AutoModelForCausalLM.from_pretrained(
            MODEL_ID,
            device_map="auto",
            torch_dtype="auto",
            load_in_4bit=True,
        )
        print("Loaded model in 4-bit.")
    except Exception as e:
        print("4-bit load failed, falling back to normal load:", str(e)[:200])
        model = AutoModelForCausalLM.from_pretrained(
            MODEL_ID,
            device_map="auto",
            torch_dtype="auto",
        )
else:
    model = AutoModelForCausalLM.from_pretrained(
        MODEL_ID,
        torch_dtype=torch.float32,
    ).to(DEVICE)

model.eval()

在接下来的内容中，我们将介绍如何使用这个模型建立层次规划智能体系统，包括：

使用开源Qwen指令模型，通过结构化多智能体架构来分解任务、规划和执行。
定义规划智能体、执行智能体和聚合智能体的功能和职责。
通过文本交互和Python执行工具实现智能体之间的通信和协同。

以下是代码示例，展示了如何训练和部署智能体系统：

def llm_chat(system: str, user: str, max_new_tokens: int = 500, temperature: float = 0.3) -> str:
   messages = [
       {"role": "system", "content": system.strip()},
       {"role": "user", "content": user.strip()},
   ]
   prompt = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
   inputs = tokenizer(prompt, return_tensors="pt").to(model.device)

   with torch.no_grad():
       out = model.generate(
           **inputs,
           max_new_tokens=max_new_tokens,
           do_sample=True if temperature > 0 else False,
           temperature=temperature,
           top_p=0.9,
           repetition_penalty=1.05,
           eos_token_id=tokenizer.eos_token_id,
       )

   text = tokenizer.decode(out[0], skip_special_tokens=True)
   return text.split(user.strip())[-1].strip()

# 以下是代码逻辑，包括定义智能体系统、编写任务描述、执行任务等
# ...

# 运行智能体系统
if __name__ == "__main__":
    demo_task = """
    Create a practical checklist to launch a small multi-agent system in Python for coordinating logistics:
    - One planner agent that decomposes tasks
    - Two executor agents (routing + inventory)
    - A simple memory store for past decisions
    Keep it lightweight and runnable in Colab.
    """

    # 运行层次规划智能体，执行任务
    _ = run_hierarchical_agent(demo_task, verbose=True)

    # 用户输入
    print("\n\nType your own task (or press Enter to skip):")
    user_task = input().strip()
    if user_task:
        _ = run_hierarchical_agent(user_task, verbose=True)

通过分层规划和执行，我们的智能体系统可以自主地理解和解决复杂任务，具有较高的灵活性和实用性。希望这些代码和方法能对您有所帮助。