Nanbeige4.1-3B实战教程：用600步工具链实现复杂任务自动分解执行

Nanbeige4.1-3B实战教程用600步工具链实现复杂任务自动分解执行1. 引言当小模型遇上“大”工具链如果你觉得只有几百亿参数的大模型才能玩转智能体Agent和工具调用那Nanbeige4.1-3B可能会改变你的看法。这是一个只有30亿参数的小模型但它有个“超能力”——支持长达600步的工具调用链。这意味着什么简单来说你可以给它一个复杂的任务比如“帮我分析这个季度的销售数据生成报告然后发邮件给团队”它能自己把这个任务拆解成一系列小步骤调用不同的工具比如读取文件、分析数据、生成图表、发送邮件一步步自动完成。听起来很酷对吧但问题来了怎么用起来600步的工具链听起来很强大但会不会很难上手今天这篇文章我就带你从零开始手把手教你部署Nanbeige4.1-3B并真正用上它那600步的工具链让这个“小身材大能量”的模型帮你自动化处理复杂任务。2. 环境准备10分钟搞定基础部署2.1 检查你的硬件和软件在开始之前我们先确认一下你的环境是否满足要求。Nanbeige4.1-3B虽然参数小但要想流畅运行特别是使用工具链功能还是有些基本要求的。硬件要求GPU推荐NVIDIA显卡显存至少6GB使用bfloat16精度内存系统内存建议16GB以上存储模型文件大约6GB预留10GB空间比较稳妥软件要求操作系统LinuxUbuntu 20.04/22.04推荐或Windows WSL2Python3.8或更高版本CUDA11.8或更高版本如果你用GPU的话如果你不确定自己的环境可以打开终端运行这几个命令检查一下# 检查Python版本 python3 --version # 检查CUDA版本如果有NVIDIA显卡 nvidia-smi # 检查内存 free -h2.2 一步一步安装依赖好了确认环境没问题后我们开始安装。我建议用conda来管理环境这样不会和你系统里其他的Python项目冲突。# 第一步创建专门的conda环境 conda create -n nanbeige python3.10 -y conda activate nanbeige # 第二步安装PyTorch根据你的CUDA版本选择 # 如果你有CUDA 11.8 pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 # 如果你没有GPU或者CUDA版本较低 pip install torch torchvision torchaudio # 第三步安装transformers和其他必要库 pip install transformers4.51.0 accelerate0.20.0 gradio4.0.0安装过程可能需要几分钟取决于你的网络速度。如果遇到网络问题可以试试国内的镜像源pip install -i https://pypi.tuna.tsinghua.edu.cn/simple torch transformers accelerate gradio2.3 下载模型文件模型文件有点大大约6GB我们需要先下载下来。你可以直接从Hugging Face下载或者如果你有国内镜像的话会更快。# download_model.py from huggingface_hub import snapshot_download # 下载模型 model_path snapshot_download( repo_idnanbeige/Nanbeige4.1-3B, local_dir./Nanbeige4.1-3B, local_dir_use_symlinksFalse ) print(f模型已下载到: {model_path})运行这个脚本它就会开始下载模型。下载时间取决于你的网速一般需要10-30分钟。如果你下载速度慢也可以手动下载访问Hugging Face的Nanbeige4.1-3B页面下载所有文件到本地目录确保目录结构完整3. 基础使用先和模型聊聊天在深入工具链之前我们先让模型跑起来感受一下它的基本能力。3.1 最简单的对话示例创建一个简单的Python脚本试试最基本的对话功能# basic_chat.py import torch from transformers import AutoModelForCausalLM, AutoTokenizer # 指定模型路径改成你下载的路径 model_path ./Nanbeige4.1-3B print(正在加载模型和分词器...) tokenizer AutoTokenizer.from_pretrained( model_path, trust_remote_codeTrue ) model AutoModelForCausalLM.from_pretrained( model_path, torch_dtypetorch.bfloat16, # 使用bfloat16节省显存 device_mapauto, # 自动分配到可用设备 trust_remote_codeTrue ) print(模型加载完成) # 准备对话 messages [ {role: user, content: 你好请介绍一下你自己} ] # 将对话格式化为模型能理解的输入 input_ids tokenizer.apply_chat_template( messages, return_tensorspt ).to(model.device) print(正在生成回复...) # 生成回复 outputs model.generate( input_ids, max_new_tokens200, # 最多生成200个token temperature0.7, # 创造性程度0.7比较平衡 top_p0.9, # 核采样参数 do_sampleTrue # 使用采样而不是贪心搜索 ) # 解码输出 response tokenizer.decode( outputs[0][len(input_ids[0]):], # 只取新生成的部分 skip_special_tokensTrue ) print(模型回复) print(response)运行这个脚本你应该能看到模型的自我介绍。如果一切正常恭喜你模型已经成功运行了3.2 试试不同的任务Nanbeige4.1-3B虽然小但能力很全面。我们试试几个不同类型的任务# test_capabilities.py def test_model(task_type, prompt): 测试模型的不同能力 messages [{role: user, content: prompt}] input_ids tokenizer.apply_chat_template( messages, return_tensorspt ).to(model.device) outputs model.generate( input_ids, max_new_tokens300, temperature0.6, top_p0.95 ) response tokenizer.decode( outputs[0][len(input_ids[0]):], skip_special_tokensTrue ) print(f\n{*50}) print(f任务类型: {task_type}) print(f输入: {prompt}) print(f输出: {response}) print(*50) # 测试代码生成 test_model(代码生成, 写一个Python函数计算列表中的最大值和最小值) # 测试逻辑推理 test_model(逻辑推理, 如果所有的猫都怕水而汤姆是一只猫那么汤姆怕水吗为什么) # 测试创意写作 test_model(创意写作, 用100字描述一个未来城市的早晨) # 测试知识问答 test_model(知识问答, 解释一下什么是机器学习用简单的语言说明)运行这个测试你会看到模型在不同任务上的表现。虽然只有3B参数但它的回答通常都很不错逻辑清晰语言流畅。4. 核心功能600步工具链实战现在来到最精彩的部分——工具链。这是Nanbeige4.1-3B的杀手锏功能能让它像真正的智能体一样工作。4.1 什么是工具链简单来说工具链就是让模型能够调用外部工具的能力。比如需要计算时调用计算器需要查天气时调用天气API需要读写文件时调用文件系统需要搜索信息时调用搜索引擎而600步的意思是模型可以把一个复杂任务分解成最多600个小步骤每个步骤调用一个工具然后根据上一步的结果决定下一步做什么。这就像有一个非常耐心的助手能帮你把复杂的工作拆解执行。4.2 配置工具链环境要使用工具链我们需要先定义一些工具。这里我创建几个常用的工具作为例子# tools.py import json import math from datetime import datetime import requests class ToolBox: 工具箱包含模型可以调用的各种工具 def __init__(self): self.tools { calculator: { description: 执行数学计算支持加减乘除、幂运算等, function: self.calculator }, get_time: { description: 获取当前日期和时间, function: self.get_current_time }, web_search: { description: 模拟网络搜索实际使用时可以接入真实搜索引擎API, function: self.simulate_search }, file_reader: { description: 读取文本文件内容, function: self.read_file }, file_writer: { description: 写入内容到文本文件, function: self.write_file }, data_analyzer: { description: 分析数据计算平均值、最大值、最小值等, function: self.analyze_data } } def calculator(self, expression: str) - str: 计算器工具 try: # 安全地评估数学表达式 # 注意实际生产环境应该用更安全的方式 result eval(expression, {__builtins__: {}}, {sqrt: math.sqrt, sin: math.sin, cos: math.cos, tan: math.tan, pi: math.pi, e: math.e}) return f计算结果: {result} except Exception as e: return f计算错误: {str(e)} def get_current_time(self, format_str: str %Y-%m-%d %H:%M:%S) - str: 获取当前时间 return f当前时间: {datetime.now().strftime(format_str)} def simulate_search(self, query: str) - str: 模拟搜索工具 # 这里只是模拟实际可以接入Google、百度等API responses { 天气: 今天晴天气温20-25度适合外出。, 新闻: 最新科技新闻AI助手越来越智能..., 股票: 上证指数今日上涨1.5%科技股表现强势。 } for key in responses: if key in query: return f搜索 {query} 的结果: {responses[key]} return f搜索 {query} 的结果: 相关信息暂时无法获取。 def read_file(self, filepath: str) - str: 读取文件工具 try: with open(filepath, r, encodingutf-8) as f: content f.read() return f文件内容:\n{content[:500]}... if len(content) 500 else f文件内容:\n{content} except Exception as e: return f读取文件失败: {str(e)} def write_file(self, filepath: str, content: str) - str: 写入文件工具 try: with open(filepath, w, encodingutf-8) as f: f.write(content) return f文件已保存到: {filepath} except Exception as e: return f写入文件失败: {str(e)} def analyze_data(self, data_str: str) - str: 数据分析工具 try: # 假设数据是逗号分隔的数字 numbers [float(x.strip()) for x in data_str.split(,)] if not numbers: return 数据为空 analysis { 数据量: len(numbers), 总和: sum(numbers), 平均值: sum(numbers) / len(numbers), 最大值: max(numbers), 最小值: min(numbers), 中位数: sorted(numbers)[len(numbers)//2] } return f数据分析结果:\n \n.join([f{k}: {v} for k, v in analysis.items()]) except Exception as e: return f数据分析失败: {str(e)} def get_tool_descriptions(self) - str: 获取所有工具的描述用于提示词 descriptions [] for name, info in self.tools.items(): descriptions.append(f{name}: {info[description]}) return \n.join(descriptions)4.3 创建工具链执行器有了工具箱我们还需要一个执行器来管理工具链的执行流程# toolchain_executor.py import re class ToolChainExecutor: 工具链执行器管理多步工具调用 def __init__(self, model, tokenizer, toolbox, max_steps600): self.model model self.tokenizer tokenizer self.toolbox toolbox self.max_steps max_steps self.conversation_history [] def parse_tool_call(self, text: str): 解析模型输出中的工具调用指令 # 匹配类似这样的格式: [TOOL: calculator] 12 34 [/TOOL] pattern r\[TOOL:\s*(\w)\](.*?)\[/TOOL\] matches re.findall(pattern, text, re.DOTALL) if matches: tool_name matches[0][0].strip() tool_input matches[0][1].strip() return tool_name, tool_input return None, None def execute_tool(self, tool_name: str, tool_input: str) - str: 执行单个工具 if tool_name not in self.toolbox.tools: return f错误: 工具 {tool_name} 不存在 try: tool_func self.toolbox.tools[tool_name][function] result tool_func(tool_input) return result except Exception as e: return f工具执行错误: {str(e)} def run(self, user_input: str, max_iterations10): 运行工具链处理用户输入 print(f\n用户输入: {user_input}) print(- * 50) # 初始化对话历史 self.conversation_history [ {role: system, content: f你是一个智能助手可以调用以下工具 {self.toolbox.get_tool_descriptions()} 当需要调用工具时请使用以下格式 [TOOL: 工具名称]工具输入参数[/TOOL] 工具执行后我会把结果返回给你格式为 [RESULT: 工具名称]执行结果[/RESULT] 请根据工具执行结果继续处理任务。如果一个工具无法完成可以调用其他工具。 任务完成后请给出最终答案。}, {role: user, content: user_input} ] for step in range(max_iterations): print(f\n步骤 {step 1}:) # 生成模型回复 input_ids self.tokenizer.apply_chat_template( self.conversation_history, return_tensorspt ).to(self.model.device) outputs self.model.generate( input_ids, max_new_tokens200, temperature0.6, top_p0.95, do_sampleTrue ) response self.tokenizer.decode( outputs[0][len(input_ids[0]):], skip_special_tokensTrue ) print(f模型回复: {response}) # 检查是否包含工具调用 tool_name, tool_input self.parse_tool_call(response) if tool_name: print(f检测到工具调用: {tool_name}) print(f工具输入: {tool_input}) # 执行工具 tool_result self.execute_tool(tool_name, tool_input) print(f工具结果: {tool_result}) # 将工具结果添加到对话历史 self.conversation_history.append({role: assistant, content: response}) self.conversation_history.append({ role: user, content: f[RESULT: {tool_name}]{tool_result}[/RESULT] }) else: # 没有工具调用任务完成 print(f任务完成最终回答: {response}) self.conversation_history.append({role: assistant, content: response}) return response print(达到最大迭代次数任务可能未完全完成) return 任务处理超时4.4 实战案例复杂任务自动分解现在让我们看一个完整的例子看看工具链如何自动分解复杂任务# complex_task_example.py def run_complex_task(): 运行一个复杂任务的例子 # 初始化所有组件 print(初始化模型和工具...) tokenizer AutoTokenizer.from_pretrained( ./Nanbeige4.1-3B, trust_remote_codeTrue ) model AutoModelForCausalLM.from_pretrained( ./Nanbeige4.1-3B, torch_dtypetorch.bfloat16, device_mapauto, trust_remote_codeTrue ) toolbox ToolBox() executor ToolChainExecutor(model, tokenizer, toolbox) # 案例1数学计算和数据分析 print(\n *60) print(案例1: 数学计算和数据分析) print(*60) task1 请帮我完成以下任务 1. 计算 (125 378) × 2.5 的结果 2. 获取当前时间 3. 分析数据23.5, 45.2, 67.8, 12.3, 89.1, 34.6 4. 把分析结果保存到文件 analysis_result.txt result1 executor.run(task1, max_iterations8) # 案例2综合信息处理 print(\n *60) print(案例2: 综合信息处理) print(*60) task2 我需要准备一个项目报告请帮我 1. 搜索最新的AI技术新闻 2. 计算我们团队这个月的平均工作效率数据85, 92, 78, 95, 88 3. 获取当前日期用于报告的时间戳 4. 把新闻摘要和数据分析结果合并保存为 report.txt result2 executor.run(task2, max_iterations10) # 案例3多步骤问题解决 print(\n *60) print(案例3: 多步骤问题解决) print(*60) task3 我正在规划一个活动需要 1. 计算活动预算场地费2500元物料费1200元餐饮费每人80元预计50人 2. 如果总预算不能超过8000元计算最多可以增加多少宣传费用 3. 获取当前时间确定活动日期 4. 把预算方案保存到 budget_plan.txt result3 executor.run(task3, max_iterations8) return result1, result2, result3 # 运行示例 if __name__ __main__: run_complex_task()运行这个脚本你会看到模型如何一步步分解任务首先识别需要调用计算器工具执行计算后调用时间工具然后调用数据分析工具最后调用文件写入工具保存结果整个过程完全自动你只需要给出一个复杂的任务描述模型就会自己规划执行步骤。5. 高级技巧优化工具链性能5.1 工具描述的优化工具链的效果很大程度上取决于工具描述的清晰度。好的描述能让模型更好地理解什么时候该用什么工具。# 优化后的工具描述示例 optimized_tools { calculate: { description: 使用计算器进行数学运算。支持的运算包括 - 基本运算: , -, *, / - 幂运算: ** 或 pow(x, y) - 平方根: sqrt(x) - 三角函数: sin(x), cos(x), tan(x)x为弧度 - 常数: pi (3.14159), e (2.71828) 示例输入: 3 4 * 2 或 sqrt(16) sin(pi/2) 注意请确保表达式是合法的数学表达式, function: calculator_function }, analyze_dataset: { description: 分析数值数据集提供统计信息。 输入格式逗号分隔的数字如 23.5, 45.2, 67.8, 12.3 输出包括数据量、总和、平均值、最大值、最小值、中位数 适用场景分析成绩、销售额、温度等数值数据, function: analyze_data_function } }5.2 多轮对话中的工具链工具链不仅适用于单次任务还可以在对话中持续使用class ConversationalToolAgent: 支持多轮对话的工具代理 def __init__(self, model, tokenizer, toolbox): self.model model self.tokenizer tokenizer self.toolbox toolbox self.conversation_context [] self.max_context_length 4000 # 控制上下文长度 def chat(self, user_message): 处理用户消息可能包含工具调用 # 添加上下文 self.conversation_context.append({role: user, content: user_message}) # 如果上下文太长清理早期记录 if len(self.conversation_context) 10: self.conversation_context self.conversation_context[-8:] # 构建系统提示 system_prompt f你是智能助手可以调用工具。可用工具 {self.toolbox.get_tool_descriptions()} 对话历史 {self.format_history()} 当前任务{user_message} 如果需要工具请使用 [TOOL: 工具名]输入[/TOOL] 格式。 # 生成回复这里简化了实际需要处理工具调用循环 messages [{role: system, content: system_prompt}] input_ids self.tokenizer.apply_chat_template( messages, return_tensorspt ).to(self.model.device) outputs self.model.generate( input_ids, max_new_tokens300, temperature0.7, top_p0.9 ) response self.tokenizer.decode( outputs[0][len(input_ids[0]):], skip_special_tokensTrue ) # 处理工具调用 tool_name, tool_input self.parse_tool_call(response) if tool_name: result self.toolbox.execute_tool(tool_name, tool_input) # 将结果加入上下文让模型继续 self.conversation_context.append({ role: assistant, content: f调用工具 {tool_name}结果{result} }) # 可以继续生成后续回复 return response def format_history(self): 格式化对话历史 return \n.join([f{msg[role]}: {msg[content][:100]}... for msg in self.conversation_context[-4:]])5.3 错误处理和重试机制在实际使用中工具调用可能会失败我们需要有错误处理机制class RobustToolChainExecutor(ToolChainExecutor): 增强版的工具链执行器包含错误处理 def execute_tool_with_retry(self, tool_name: str, tool_input: str, max_retries3): 带重试的工具执行 for attempt in range(max_retries): try: result self.execute_tool(tool_name, tool_input) # 检查结果是否合理 if 错误 in result or 失败 in result: if attempt max_retries - 1: print(f工具执行可能有问题重试 {attempt 1}/{max_retries}) continue return result except Exception as e: print(f工具执行异常 (尝试 {attempt 1}): {str(e)}) if attempt max_retries - 1: return f工具执行失败: {str(e)} return 工具执行失败已达到最大重试次数 def validate_tool_input(self, tool_name: str, tool_input: str): 验证工具输入是否合理 validation_rules { calculator: lambda x: all(c in 0123456789-*/.() sqrtcossintanpie for c in x), analyze_data: lambda x: any(c.isdigit() for c in x) and , in x, read_file: lambda x: x.endswith((.txt, .csv, .json, .py)), write_file: lambda x: len(x) 10 # 简单的内容长度检查 } if tool_name in validation_rules: if not validation_rules[tool_name](tool_input): return f工具输入验证失败: {tool_input} 不符合 {tool_name} 的输入要求 return 验证通过6. 实际应用场景6.1 场景一自动化数据分析报告假设你每天需要分析销售数据并生成报告可以这样自动化def auto_sales_report(): 自动化销售报告生成 executor get_tool_chain_executor() # 获取配置好的执行器 task 请生成今日销售报告 1. 读取 sales_data.csv 文件包含每日销售数据 2. 分析今日销售额、平均订单价、最畅销产品 3. 计算环比增长率需要读取昨日数据 4. 获取当前日期作为报告日期 5. 生成报告摘要保存为 daily_report.txt 6. 如果销售额超过目标10000元额外生成庆祝消息 result executor.run(task, max_iterations15) # 可以进一步处理结果 if 庆祝 in result: # 发送通知等 pass return result6.2 场景二智能客服助手在客服场景中工具链可以帮助处理复杂查询class CustomerServiceAgent: 客服助手集成工具链 def handle_customer_query(self, query: str): 处理客户查询 # 根据查询类型选择工具 if 订单状态 in query or 物流 in query: # 调用订单查询工具 return self.query_order_info(query) elif 价格 in query or 优惠 in query: # 调用价格计算工具 return self.calculate_price(query) elif 退货 in query or 退款 in query: # 调用退货流程工具 return self.handle_return(query) else: # 使用工具链处理复杂查询 task f客户查询{query} 请帮我 1. 理解客户问题 2. 搜索相关知识库 3. 提供准确回答 4. 如果有需要计算的内容使用计算器 5. 生成最终回复 return self.tool_chain_executor.run(task)6.3 场景三个人效率助手对于个人使用可以创建个性化的效率工具def personal_productivity_assistant(): 个人效率助手示例 # 创建个性化工具箱 personal_tools { schedule_meeting: { description: 安排会议输入时间参与者主题, function: schedule_meeting_func }, set_reminder: { description: 设置提醒输入时间事项, function: set_reminder_func }, summarize_text: { description: 总结文本内容输入长文本, function: summarize_text_func }, translate_text: { description: 翻译文本输入文本目标语言, function: translate_text_func } } # 集成到工具链 toolbox ToolBox() toolbox.tools.update(personal_tools) executor ToolChainExecutor(model, tokenizer, toolbox) # 处理复杂任务 task 今天的工作安排 1. 上午10点和技术团队开会讨论项目进度 2. 需要阅读并总结 attached_doc.txt 中的技术文档 3. 下午3点前完成项目报告计算需要的时间 4. 设置提醒下午4点提交报告 5. 把会议纪要和文档总结保存到 work_log.txt return executor.run(task)7. 总结通过这篇教程你应该已经掌握了Nanbeige4.1-3B工具链的核心用法。我们来回顾一下关键点7.1 核心收获小模型也能有大作为Nanbeige4.1-3B虽然只有30亿参数但通过600步的工具链它能处理相当复杂的任务工具链让AI更实用不再是简单的问答而是能真正帮你做事——计算、分析、读写文件、搜索信息部署其实很简单按照教程步骤10-30分钟就能从零搭建起可用的环境扩展性很强你可以轻松添加自己的工具定制专属的AI助手7.2 最佳实践建议根据我的使用经验给你几个实用建议对于初学者先从简单的工具开始比如计算器、时间获取测试时使用明确的指令让模型知道什么时候该用工具一次不要给太多步骤3-5步比较合适对于进阶用户设计清晰的工具描述这是工具链好用的关键实现错误处理和重试机制提高稳定性考虑上下文管理避免对话历史过长性能优化如果响应慢可以调整max_new_tokens参数对于复杂任务适当增加max_iterations使用bfloat16精度平衡速度和精度7.3 下一步探索方向掌握了基础之后你可以尝试集成真实API把工具换成真实的天气API、股票API、邮件发送等创建专业工具链针对你的工作领域设计专用工具多模型协作让Nanbeige4.1-3B和其他模型配合工作可视化界面用Gradio或Streamlit做个Web界面最重要的是开始动手实践。工具链的魅力在于你给它越多好用的工具它就能帮你做越多的事情。从今天开始试着让Nanbeige4.1-3B成为你的智能助手吧获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。