# 创建环境
conda create -n agent_camp3 python=3.10 -y
# 激活环境
conda activate agent_camp3
# 安装 torch
conda install pytorch==2.1.2 torchvision==0.16.2 torchaudio==2.1.2 pytorch-cuda=12.1 -c pytorch -c nvidia -y
# 安装其他依赖包
pip install termcolor==2.4.0
pip install lmdeploy==0.5.2
# 安装lagent
mkdir -p /root/agent_camp3
cd /root/agent_camp3
git clone https://github.com/InternLM/lagent.git
cd lagent && git checkout 81e7ace && pip install -e . && cd ..
pip install griffe==0.48.0
第一步，通过 lmdeploy 部署一个 api_server :
lmdeploy serve api_server /share/new_models/Shanghai_AI_Laboratory/internlm2_5-7b-chat --model-name internlm2_5-7b-chat
第二步，启动 Lagent 的Web Demo：
cd /root/agent_camp3/lagent
streamlit run examples/internlm2_agent_web_demo.py
第三步，映射这两个端口到本地：
ssh -p 47155 [email protected] -CNg -L 8501:127.0.0.1:8501 -L 23333:127.0.0.1:23333 -o StrictHostKeyChecking=no
然后打开web UI，填上模型名称、端口，选上插件。然后就可以玩耍了：
从 Lagent Web Demo 的代码中浅尝逻辑。
首先，从代码中可以看到插件是在初始化时就注册好的：
# SessionState.init_state() 中
action_list = [
ArxivSearch(), # 预先注册了 ArxivSearch 插件
]
st.session_state['plugin_map'] = {
action.name: action
for action in action_list
}Chatbot 的初始化也和插件有关：
# StreamlitUI.initialize_chatbot() 中
return Internlm2Agent(
llm=model,
protocol=Internlm2Protocol(
tool=dict(
begin='{start_token}{name}\n',
start_token='<|action_start|>',
name_map=dict(
plugin='<|plugin|>',
interpreter='<|interpreter|>'
),
belong='assistant',
end='<|action_end|>\n',
),
),
max_turn=7
)这里的关键是定义了一个特殊的格式来标记工具调用，使用了特殊的 token：
<|action_start|>- 动作开始<|plugin|>- 插件调用<|action_end|>- 动作结束
让我用一个序列图来展示完整的交互过程：
让我梳理一下完整的 LLM-插件交互周期：
- LLM 决定使用插件：
- LLM 根据用户输入和系统提示词，判断是否需要使用插件
- 如果需要，生成包含特殊token和JSON格式参数的响应
- Agent 解析和执行：
- 检测到特殊token时，解析JSON参数
- 查找对应的插件并执行
- 收集插件的执行结果
- 继续对话：
- 将插件执行结果作为新的上下文
- LLM 生成新的响应，可能继续调用插件
- 直到生成完整的答案或达到最大轮次
- 多轮交互：
- LLM 可能会根据插件返回的结果继续提问或调用其他插件
- 每一轮交互都会被记录在会话历史中
- 整个过程是动态的，直到LLM认为已经生成了完整的答案
- 关于判断是否调用插件：
# 是的，LLM的输出会有两种情况：
# 情况1：普通回复，不需要调用插件
'''
这是一个普通的回答，不需要调用任何插件...
'''
# 情况2：需要调用插件的回复
'''
让我帮你搜索相关信息
<|action_start|><|plugin|>
{
"name": "arxiv_search",
"parameters": {
"query": "...",
"max_results": 5
}
}
<|action_end|>
'''- 关于插件信息的传递，从代码中我们可以看到这是通过系统提示词实现的：
# StreamlitUI.setup_sidebar() 中
meta_prompt = st.sidebar.text_area('系统提示词', value=META_CN)
plugin_prompt = st.sidebar.text_area('插件提示词', value=PLUGIN_CN)
# PLUGIN_CN 就是告诉 LLM 可用插件信息的系统提示词
# META_CN 是基础的系统提示词让我们画个图说明这个提示词系统：
- 重要细节：
- 提示词的组成：
# 一个典型的提示词结构：
# 1. 系统基础提示词（META_CN）
'''
你是一个AI助手...
'''
# 2. 插件提示词（PLUGIN_CN）
'''
你可以使用以下插件：
1. arxiv_search: 搜索学术论文
参数格式：
{
"name": "arxiv_search",
"parameters": {
"query": "搜索关键词",
"max_results": "最大结果数"
}
}
调用格式：
<|action_start|><|plugin|>
{json格式的参数}
<|action_end|>
示例：...
'''
# 3. 历史对话上下文
[
{"role": "user", "content": "之前的对话..."},
{"role": "assistant", "content": "之前的回复..."}
]
# 4. 当前用户输入
{"role": "user", "content": "当前问题"}- 关于每轮对话：
- 初始化时：
# StreamlitUI.initialize_chatbot() 中设置协议
protocol=Internlm2Protocol(
tool=dict(
begin='{start_token}{name}\n',
start_token='<|action_start|>',
name_map=dict(
plugin='<|plugin|>',
interpreter='<|interpreter|>'
),
belong='assistant',
end='<|action_end|>\n',
),
)- 每轮对话中：
- 系统提示词（META_CN）和插件提示词（PLUGIN_CN）会一直保持
- 会带上完整的对话历史
- LLM 可以参考这些信息来决定是否使用插件
- 工作流程：
从代码里看，实际的插件提示词模板是在 PLUGIN_CN 变量中定义的，但完整的插件信息是在运行时动态填充的。让我们看看这个过程：
- 首先，在 sidebar 设置时，我们选择了要启用哪些插件
- 然后，当有插件被选择时，会创建 ActionExecutor
- 实际的工具说明和格式应该是在
ActionExecutor这个类中生成的。这个类会：
- 收集所有可用插件的信息
- 生成它们的描述和调用格式
- 将这些信息填充到提示词模板中
让我模拟一下这个过程：
# 这个过程大概是这样的（简化版）：
class ArxivSearch:
name = "arxiv_search"
description = "搜索学术论文"
parameters = {
"query": "搜索关键词",
"max_results": "最大结果数"
}
class ActionExecutor:
def __init__(self, actions):
self.actions = actions
def get_tools_description(self):
tools_desc = []
for action in self.actions:
desc = f"""
{action.name}: {action.description}
参数格式：
{{
"name": "{action.name}",
"parameters": {{
# 这里列出参数说明
}}
}}
"""
tools_desc.append(desc)
return "\n".join(tools_desc)
# 当实际使用时
action_executor = ActionExecutor([ArxivSearch()])
tools_description = action_executor.get_tools_description()
# 最终的提示词会变成：
"""
你可以使用如下工具：
arxiv_search: 搜索学术论文
参数格式：
{
"name": "arxiv_search",
"parameters": {
"query": "搜索关键词",
"max_results": "最大结果数"
}
}
如果你已经获得足够信息，请直接给出答案. 避免不必要的工具调用! 同时注意你可以使用的工具，不要随意捏造！
"""这里有个重要的设计模式：
关键点是：
- 插件注册：
- 每个插件（如 ArxivSearch）都需要定义自己的名称、描述和参数格式
- 这些信息会被用来：
- 生成给 LLM 的工具说明
- 验证 LLM 生成的调用格式是否正确
- 运行时：
- ActionExecutor 收集所有已启用插件的信息
- 动态生成完整的工具说明
- 这些说明会被填充到提示词模板中
- 提示词结构：
基础提示词（告诉 LLM 它的角色）
+
动态生成的工具说明（告诉它可用的工具）
+
调用格式说明（告诉它如何调用）
+
使用建议（告诉它何时调用）
所以，总结一下：
- LLM 学会调用工具的方式是通过完整的提示词系统
- 具体的工具信息是由 ActionExecutor 在运行时动态生成的
- 这种设计允许我们：
- 动态添加/删除工具
- 确保 LLM 只调用实际存在的工具
- 标准化工具的调用格式
第一步，写MagicMaker插件本身的代码
第二步，在原来的 web demo 代码中引入这个MagicMaker
可惜 MagicMaker 的API接口目前无法正常响应。
MagicMaker 代码逻辑流程图：
我自定义了一个 帮助用户诊断深度学习环境 的插件，检查深度学习环境（PyTorch、CUDA、系统信息等）
- 检查 PyTorch 版本和配置
- 检查 CUDA 可用性和版本
- 检查 cuDNN 版本和状态
- 获取 GPU 信息
- 获取系统环境信息
-
环境检查插件介绍：
- 功能：检查深度学习环境（PyTorch、CUDA、系统信息等）
- 核心实现：
class EnvChecker(BaseAction): @tool_api def check_torch_env(self) -> dict: # 1. 收集 PyTorch 信息（版本、CUDA可用性等） # 2. 收集系统信息（Python版本、操作系统等） # 3. 收集 CUDA 系统信息（GPU信息、驱动版本等） # 4. 格式化输出 return {"text": formatted_output}
- 错误处理：捕获可能的异常，返回友好的错误信息
- 输出：以分类整理的文本形式展示所有环境信息
-
如何写自定义插件：
步骤 1: 创建插件类
from lagent.actions.base_action import BaseAction, tool_api class MyPlugin(BaseAction): def __init__(self): super().__init__()
步骤 2: 实现核心方法
@tool_api # 必须使用这个装饰器 def my_function(self, param: str) -> dict: try: result = do_something(param) return {"text": result} # 或 {"image": path} 等 except Exception as e: return {"error": str(e)}
步骤 3: 注册插件
# 在主程序中 from my_plugin import MyPlugin action_list = [ ArxivSearch(), MyPlugin(), # 添加新插件 ]
关键原则：
- 继承 BaseAction
- 使用 @tool_api 装饰器
- 返回标准格式字典（包含 text/image/error 等键）
- 做好错误处理
- 遵循统一的文档规范
就这样，一个基本的插件就完成了！
import sys
import subprocess
from typing import Dict, Optional
import platform
from lagent.actions.base_action import BaseAction, tool_api
class EnvChecker(BaseAction):
"""检查深度学习环境的插件"""
@tool_api
def check_torch_env(self) -> dict:
"""获取PyTorch相关环境信息"""
try:
# 检查PyTorch
import torch
torch_info = {
"torch_version": torch.__version__,
"cuda_available": torch.cuda.is_available(),
"cuda_device_count": torch.cuda.device_count() if torch.cuda.is_available() else 0,
}
# 如果CUDA可用，获取更多信息
if torch.cuda.is_available():
torch_info.update({
"cuda_device_names": [torch.cuda.get_device_name(i) for i in range(torch.cuda.device_count())],
"current_device": torch.cuda.current_device(),
"cuda_version": torch.version.cuda,
"cudnn_version": torch.backends.cudnn.version() if torch.backends.cudnn.is_available() else "Not available",
"cudnn_enabled": torch.backends.cudnn.enabled,
})
# 获取系统信息
sys_info = {
"python_version": sys.version.split()[0],
"os": platform.platform(),
"cpu_info": self._get_cpu_info(),
}
# 获取CUDA系统环境
cuda_sys_info = self._get_cuda_sys_info()
# 组合所有信息
full_info = {
"torch_info": torch_info,
"system_info": sys_info,
"cuda_system_info": cuda_sys_info
}
# 格式化输出信息
output = self._format_env_info(full_info)
return {"text": output}
except Exception as e:
return {"error": f"Error checking environment: {str(e)}"}
def _get_cpu_info(self) -> str:
"""获取CPU信息"""
if platform.system() == "Linux":
try:
with open("/proc/cpuinfo") as f:
for line in f:
if "model name" in line:
return line.split(":")[1].strip()
except:
pass
return platform.processor() or "Unknown"
def _get_cuda_sys_info(self) -> Dict[str, Optional[str]]:
"""获取系统CUDA环境信息"""
cuda_info = {}
# 检查 nvidia-smi
try:
nvidia_smi = subprocess.check_output(["nvidia-smi", "--query-gpu=gpu_name,driver_version,memory.total",
"--format=csv,noheader"]).decode()
cuda_info["gpu_info"] = nvidia_smi.strip()
except:
cuda_info["gpu_info"] = "nvidia-smi not available"
# 检查 nvcc
try:
nvcc_output = subprocess.check_output(["nvcc", "--version"]).decode()
cuda_info["nvcc_info"] = nvcc_output.strip()
except:
cuda_info["nvcc_info"] = "nvcc not available"
return cuda_info
def _format_env_info(self, info: Dict) -> str:
"""格式化环境信息为易读的文本"""
output = []
# PyTorch信息
output.append("=== PyTorch Environment ===")
torch_info = info["torch_info"]
output.append(f"PyTorch Version: {torch_info['torch_version']}")
output.append(f"CUDA Available: {torch_info['cuda_available']}")
output.append(f"GPU Count: {torch_info['cuda_device_count']}")
if torch_info['cuda_available']:
output.append("\n=== CUDA Device Information ===")
for i, name in enumerate(torch_info['cuda_device_names']):
output.append(f"GPU {i}: {name}")
output.append(f"Current Device: {torch_info['current_device']}")
output.append(f"CUDA Version: {torch_info['cuda_version']}")
output.append(f"cuDNN Version: {torch_info['cudnn_version']}")
output.append(f"cuDNN Enabled: {torch_info['cudnn_enabled']}")
# 系统信息
output.append("\n=== System Information ===")
sys_info = info["system_info"]
output.append(f"Python Version: {sys_info['python_version']}")
output.append(f"Operating System: {sys_info['os']}")
output.append(f"CPU: {sys_info['cpu_info']}")
# CUDA系统信息
output.append("\n=== CUDA System Information ===")
cuda_sys_info = info["cuda_system_info"]
if cuda_sys_info["gpu_info"] != "nvidia-smi not available":
output.append("GPU Information (from nvidia-smi):")
output.append(cuda_sys_info["gpu_info"])
if cuda_sys_info["nvcc_info"] != "nvcc not available":
output.append("\nNVCC Information:")
output.append(cuda_sys_info["nvcc_info"])
return "\n".join(output)使用方法：
- 创建文件：
# 在 lagent/actions/ 目录下创建新文件
touch lagent/actions/env_checker.py-
将上面的代码保存到
env_checker.py文件中。 -
修改主程序添加插件：
from lagent.actions import ActionExecutor, ArxivSearch
from lagent.actions.env_checker import EnvChecker
# 在 action_list 中添加新插件
action_list = [
ArxivSearch(),
EnvChecker(),
]最终效果：
在这个位置打印信息：
就可以在终端看到：
[{
'name': 'ArxivSearch',
'description': 'Search information from Arxiv.org. Useful for when you need to answer questions about Physics, Mathematics, Computer Science, Quantitative Biology, Quantitative Finance, Statistics, Electrical Engineering, and Economics from scientific articles on arxiv.org.',
'api_list': [{
'name': 'get_arxiv_article_information',
'description': 'Run Arxiv search and get the article meta information.',
'parameters': [{
'name': 'query',
'type': 'STRING',
'description': 'the content of search query'
}],
'required': ['query'],
'return_data': [{
'name': 'content',
'description': 'a list of 3 arxiv search papers',
'type': 'STRING'
}],
'parameter_description': 'If you call this tool, you must pass arguments in the JSON format {key: value}, where the key is the parameter name.'
}]
}, {
'name': 'MagicMaker',
'description': '',
'api_list': [{
'name': 'generate_image',
'description': 'Run magicmaker and get the generated image according to the keywords.',
'parameters': [{
'name': 'keywords',
'type': 'STRING',
'description': 'the keywords to generate image'
}],
'required': ['keywords'],
'parameter_description': 'If you call this tool, you must pass arguments in the JSON format {key: value}, where the key is the parameter name.'
}]
}, {
'name': 'EnvChecker',
'description': '检查深度学习环境的插件',
'api_list': [{
'name': 'check_torch_env',
'description': '获取PyTorch相关环境信息',
'parameters': [],
'required': [],
'parameter_description': 'If you call this tool, you must pass arguments in the JSON format {key: value}, where the key is the parameter name.'
}]
}]
解析一下这个结构：
[{
# 插件基本信息
'name': 'EnvChecker', # 从类名自动获取
'description': '检查深度学习环境的插件', # 从类的docstring获取
# API列表（插件提供的方法）
'api_list': [{
'name': 'check_torch_env', # 方法名
'description': '获取PyTorch相关环境信息', # 从方法的docstring获取
'parameters': [], # 方法参数列表（这里是空的，因为没有参数）
'required': [], # 必需的参数（这里是空的）
'parameter_description': 'If you call this tool...' # 参数说明
}]
}]所以当我们写插件时：
-
系统会自动收集：
- 类名 → name
- 类文档 → description
- 方法信息 → api_list
-
如果想让描述更精确，我们可以：
class EnvChecker(BaseAction):
"""检查深度学习环境的插件，可以获取PyTorch、CUDA等配置信息""" # 更详细的描述
@tool_api
def check_torch_env(self, scope: str = "all") -> dict: # 添加参数
"""获取PyTorch相关环境信息
Args:
scope: 检查范围，可选 'all', 'torch', 'cuda'
"""框架会自动从类和方法的定义中收集必要信息！这是一个很贴心的设计。
打开 Lagent Web Demo 页面时报错：
Traceback (most recent call last):
File "/root/.conda/envs/lagent/lib/python3.10/site-packages/streamlit/runtime/scriptrunner/exec_code.py", line 88, in exec_func_with_error_handling
result = func()
File "/root/.conda/envs/lagent/lib/python3.10/site-packages/streamlit/runtime/scriptrunner/script_runner.py", line 579, in code_to_exec
exec(code, module.__dict__)
File "/root/agent_camp4/lagent/examples/internlm2_agent_web_demo.py", line 8, in <module>
from lagent.actions import ActionExecutor, ArxivSearch, IPythonInterpreter
File "/root/agent_camp4/lagent/lagent/__init__.py", line 2, in <module>
from .actions import * # noqa: F401, F403
File "/root/agent_camp4/lagent/lagent/actions/__init__.py", line 3, in <module>
from .action_executor import ActionExecutor
File "/root/agent_camp4/lagent/lagent/actions/action_executor.py", line 4, in <module>
from .base_action import BaseAction
File "/root/agent_camp4/lagent/lagent/actions/base_action.py", line 14, in <module>
from class_registry import AutoRegister, ClassRegistry
ImportError: cannot import name 'AutoRegister' from 'class_registry' (/root/.conda/envs/lagent/lib/python3.10/site-packages/class_registry/__init__.py)
解决：
pip install class_registry