最近在尝试基于大语言模型构建一个能够自主执行复杂任务的智能体(Agent)时,遇到了一个关键瓶颈:模型对工具(Tools)的调用能力。许多小参数模型(例如7B、13B)在理解指令、规划步骤后,往往在“动手”执行工具调用这一步失败,导致智能体逻辑链断裂,沦为“纸上谈兵”。这促使我将目光投向了参数规模更大、工具调用能力更强的模型。经过一番调研和测试,最终选定了通义千问的Qwen3.6-27B模型作为智能体开发的核心引擎,并成功完成了从环境部署、基础功能验证到初步的测试开发(Test Development)阶段。
本文将完整记录我基于 Qwen3.6-27B 模型,从零开始搭建一个可工作的智能体开发环境,并推进到 P1(第一阶段)功能开发的全过程。内容涵盖模型选型与部署、智能体基础框架搭建、工具调用能力测试、以及一个简单的测试开发案例。无论你是想了解如何高效部署大模型服务,还是对智能体(Agent)开发感兴趣,希望构建一个能“听懂指令并执行操作”的AI助手,这篇文章都能提供一套可复现的实操指南。
1. 为什么选择 Qwen3.6-27B 进行智能体开发?
在开始动手之前,我们需要明确技术选型的理由。智能体开发对底层大模型有几个核心要求:
- 强大的工具调用(Tool Calling)能力:这是智能体的“手”和“脚”。模型必须能准确理解何时、如何使用外部工具(如搜索、计算、调用API),并生成符合工具要求的参数。
- 优秀的指令遵循(Instruction Following)和规划(Planning)能力:智能体需要将复杂任务分解为可执行的子步骤,并严格遵循预设的规则和格式。
- 足够大的上下文窗口(Context Length):智能体在运行过程中会积累大量的对话历史、工具调用结果和中间思考过程,需要模型有强大的“记忆力”来处理长上下文。
- 合理的性能与资源消耗:模型需要在可接受的推理延迟和硬件成本下提供服务。
Qwen3.6-27B恰好在这几个方面表现突出:
- 专为智能体优化:根据官方介绍和社区反馈,Qwen3.6-27B 相比前代,在智能体编码和工具调用能力上有显著增强。27B的参数规模在效果和效率之间取得了很好的平衡,避免了“小模型调用不成功,超大模型部署困难”的尴尬。
- 超长上下文:支持高达262,144tokens 的上下文长度,足以应对复杂的多轮交互和长文档分析任务。
- 原生多模态支持:为未来扩展图像、音频等多模态智能体能力预留了空间。
- 活跃的生态与工具链:拥有完善的模型仓库(如 ModelScope)、高效的推理框架(如 vLLM)支持,部署和集成相对顺畅。
因此,选择 Qwen3.6-27B 作为智能体开发的“大脑”,是一个兼顾能力、成本和工程可行性的方案。
2. 环境准备与模型部署
智能体开发的第一步,是为 Qwen3.6-27B 提供一个稳定、高效的推理服务。我们选择使用vLLM框架进行部署,它以其高效的内存管理和推理速度著称。
2.1 硬件与基础环境
根据官方文档,部署 Qwen3.6-27B 需要较强的算力支持。以下是两种常见的硬件配置方案:
- 方案A(推荐):1台 Atlas 800 A3 训练服务器(配置 16 张 64GB 显存的 NPU)。
- 方案B:1台 Atlas 800 A2 训练服务器(配置 8 张 64GB 显存的 NPU)。
对于大多数开发者和中小团队,可能无法直接获取上述硬件。一个可行的替代方案是使用云计算服务商提供的GPU/NPU实例,例如配备了多张 A100/A800 或 昇腾910B 的云服务器。核心原则是确保有足够的显存(VRAM/HBM)来加载整个27B参数的模型(量化后约需 30-40GB,非量化版本需更多)。
本文的演示环境基于一台拥有 8 张 80GB A100 GPU 的云服务器,操作系统为 Ubuntu 22.04 LTS。使用 vLLM 的 GPU 版本进行部署,其操作逻辑与 NPU 版本(vLLM-Ascend)类似。
2.2 使用 Docker 快速部署 vLLM 服务
为了环境隔离和便捷性,我们使用 Docker 进行部署。这里以部署量化版本Qwen3.6-27B-w8a8为例,它能在保证精度的同时显著降低显存占用。
步骤1:拉取合适的 Docker 镜像vLLM 提供了官方镜像。我们需要根据硬件选择 tag。对于 NVIDIA GPU,我们使用vllm/vllm-openai:latest或指定版本。
# 拉取 vLLM 官方镜像(包含 OpenAI 兼容 API) docker pull vllm/vllm-openai:latest步骤2:准备模型权重将模型权重下载到宿主机的一个目录中,例如/data/models/。你可以从 ModelScope 或 Hugging Face 下载。
# 使用 modelscope 库下载(需提前安装:pip install modelscope) from modelscope import snapshot_download model_dir = snapshot_download('Qwen/Qwen3.6-27B-Instruct', cache_dir='/data/models/') # 或者直接使用 huggingface-cli # huggingface-cli download Qwen/Qwen3.6-27B-Instruct --local-dir /data/models/Qwen3.6-27B-Instruct步骤3:启动 Docker 容器并运行 vLLM 服务以下命令启动一个容器,将模型目录挂载进去,并开放 8000 端口用于 API 访问。
# 设置环境变量 export MODEL_PATH=/data/models/Qwen3.6-27B-Instruct export PORT=8000 # 运行 Docker 容器 docker run --rm --gpus all \ -p ${PORT}:8000 \ -v ${MODEL_PATH}:/model \ -v /tmp/vllm:/tmp/vllm \ # 可选的,用于缓存 vllm/vllm-openai:latest \ --model /model \ --served-model-name qwen3.6-27b \ --max-model-len 16384 \ # 根据你的需求调整,最大支持262144 --tensor-parallel-size 2 \ # 张量并行大小,根据你的GPU数量调整。8卡A100可设为8。 --quantization awq \ # 使用AWQ量化,如果下载的是AWQ量化模型。如果是w8a8,可能需要其他参数。 --trust-remote-code关键参数解释:
--gpus all: 将宿主机的所有GPU暴露给容器。-p 8000:8000: 将容器的8000端口映射到宿主机。-v ${MODEL_PATH}:/model: 将宿主机模型目录挂载到容器的/model路径。--model /model: 指定模型路径。--tensor-parallel-size:非常重要。它指定了将模型参数拆分到多少张GPU上进行张量并行推理。必须小于等于你的GPU数量,且通常是2的幂次方(如1,2,4,8)。设置不当会导致OOM(内存不足)或性能低下。--quantization: 指定量化方法,如awq,gptq等,可以大幅减少显存占用。请确保下载的模型权重是对应的量化版本。--trust-remote-code: 信任并运行模型自带的定制化代码(Qwen模型需要)。
步骤4:验证服务是否正常服务启动后,你可以使用curl命令测试 OpenAI 兼容的 API 接口。
curl http://localhost:8000/v1/completions \ -H "Content-Type: application/json" \ -d '{ "model": "qwen3.6-27b", "prompt": "请用一句话介绍你自己。", "max_tokens": 100, "temperature": 0.7 }'如果返回包含生成的文本,说明模型服务部署成功。
2.3 (备选)从源码安装 vLLM
如果你需要更灵活的自定义或调试,可以从源码安装 vLLM。
# 1. 创建并激活 Python 虚拟环境(推荐) python -m venv vllm-env source vllm-env/bin/activate # 2. 安装 PyTorch (请根据你的CUDA版本从官网获取对应命令) pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 # 3. 安装 vLLM pip install vllm # 4. 直接使用 vllm 命令启动服务 vllm serve Qwen/Qwen3.6-27B-Instruct \ --model /path/to/your/model \ --served-model-name qwen3.6 \ --max-model-len 16384 \ --tensor-parallel-size 2 \ --quantization awq \ --trust-remote-code3. 智能体(Agent)基础框架搭建
有了模型服务,接下来我们需要构建智能体的“身体”——即能够理解用户指令、规划任务、调用工具并整合结果的框架。这里我们使用LangChain和LangGraph这两个流行的库,它们提供了构建智能体所需的高级抽象。
3.1 安装依赖
创建一个新的 Python 项目,并安装必要的库。
# 创建项目目录 mkdir qwen-agent-dev && cd qwen-agent-dev python -m venv .venv source .venv/bin/activate # Linux/Mac # .venv\Scripts\activate # Windows # 安装核心库 pip install langchain langchain-community langgraph pip install openai # 用于连接 vLLM 的 OpenAI 兼容接口 pip install requests # 用于工具调用 pip install python-dotenv # 管理环境变量3.2 连接 Qwen3.6-27B 模型服务
在config.py或环境变量中配置模型服务的地址。
# config.py import os from dotenv import load_dotenv load_dotenv() # vLLM 服务地址 (假设在本地运行) VLLM_API_BASE = os.getenv("VLLM_API_BASE", "http://localhost:8000/v1") VLLM_API_KEY = os.getenv("VLLM_API_KEY", "not-needed") # vLLM 默认不需要key MODEL_NAME = os.getenv("MODEL_NAME", "qwen3.6-27b")然后,创建一个工具来初始化 LangChain 的 LLM 对象。
# llm_client.py from langchain_openai import ChatOpenAI from config import VLLM_API_BASE, VLLM_API_KEY, MODEL_NAME def get_qwen_llm(temperature=0.1, max_tokens=2000): """ 创建连接到本地 vLLM 服务的 LLM 实例。 由于 vLLM 提供了与 OpenAI 兼容的 API,我们可以直接使用 ChatOpenAI。 """ llm = ChatOpenAI( base_url=VLLM_API_BASE, # 指向你的 vLLM 服务 api_key=VLLM_API_KEY, model=MODEL_NAME, temperature=temperature, max_tokens=max_tokens, timeout=60, ) return llm # 测试连接 if __name__ == "__main__": llm = get_qwen_llm() try: response = llm.invoke("你好,请回复‘服务连接成功’。") print(f"模型回复: {response.content}") except Exception as e: print(f"连接失败: {e}")3.3 定义智能体的工具(Tools)
工具是智能体与外界交互的接口。我们定义几个简单的工具作为示例。
# tools/calculator.py from langchain.tools import tool import math @tool def calculator(expression: str) -> str: """ 一个简单的计算器工具。输入一个数学表达式字符串,返回计算结果。 例如: `calculator("3 + 5 * 2")` -> `13` """ try: # 警告:使用 eval 存在安全风险,仅用于演示。生产环境应使用安全表达式解析器(如 ast.literal_eval 或第三方库)。 # 这里为了演示简单处理,实际请勿在开放环境中使用。 result = eval(expression, {"__builtins__": None}, {"math": math}) return f"计算结果: {result}" except Exception as e: return f"计算错误: {e}" # tools/web_search.py (模拟) from langchain.tools import tool import requests @tool def search_web(query: str) -> str: """ 模拟网络搜索工具。输入搜索关键词,返回模拟的搜索结果摘要。 在实际项目中,你应该接入真实的搜索引擎API(如 SerperAPI、Google Custom Search)。 """ # 这里用一个模拟的API响应来演示 mock_responses = { "今天的天气": "北京:晴,15-25°C;上海:多云,18-28°C。", "Qwen3.6 发布": "通义千问 Qwen3.6 系列模型于近期发布,在代码、数学、推理等能力上均有显著提升。", "Python 最新版本": "Python 的最新稳定版本是 3.12。", } # 简单匹配,实际应用应更智能 for key in mock_responses: if key in query: return f"搜索 ‘{query}’ 的结果:{mock_responses[key]}" return f"未找到关于 ‘{query}’ 的模拟结果。请尝试其他关键词。" # tools/current_time.py from langchain.tools import tool from datetime import datetime @tool def get_current_time(timezone: str = "Asia/Shanghai") -> str: """ 获取当前时间。可以指定时区(默认为上海/北京时间)。 参数 timezone: 时区字符串,例如 'UTC', 'America/New_York'。 """ try: # 这里简化处理,实际应使用 pytz 或 zoneinfo 库 if timezone != "Asia/Shanghai": return f"演示工具仅支持 ‘Asia/Shanghai’ 时区,当前时间(上海)是:{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") return f"当前时间({timezone})是:{current_time}" except Exception as e: return f"获取时间失败: {e}" # 在 __init__.py 或主文件中汇总工具 def get_all_tools(): from .calculator import calculator from .web_search import search_web from .current_time import get_current_time return [calculator, search_web, get_current_time]3.4 构建智能体执行图(Agent with LangGraph)
LangGraph 允许我们以“图”的形式定义智能体的工作流,非常适合处理带有循环(如思考-行动-观察)的智能体逻辑。我们构建一个经典的ReAct (Reason + Act)风格智能体。
# agent/graph.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from llm_client import get_qwen_llm from tools import get_all_tools # 1. 定义智能体的状态结构 class AgentState(TypedDict): messages: Annotated[List, operator.add] # 对话消息历史 current_step: str # 当前步骤描述(可选,用于调试) # 2. 创建 LLM 和工具 llm = get_qwen_llm(temperature=0.1, max_tokens=1000) tools = get_all_tools() llm_with_tools = llm.bind_tools(tools) # 3. 定义提示词模板 system_prompt = """你是一个有帮助的AI助手,可以调用工具来解决问题。 请遵循以下步骤: 1. 理解用户的问题。 2. 如果需要使用工具,请精确地调用一个合适的工具,并等待工具返回结果。 3. 根据工具返回的结果和对话历史,决定下一步是继续调用工具还是直接给出最终答案。 4. 最终答案应清晰、完整,并基于你获得的所有信息。 你可以使用的工具: {tools_descriptions} 注意:一次只调用一个工具。""" prompt = ChatPromptTemplate.from_messages([ ("system", system_prompt), MessagesPlaceholder(variable_name="messages"), ]) # 4. 定义节点函数 def should_continue(state: AgentState) -> str: """根据最新的AI消息,决定下一步是调用工具还是结束。""" last_message = state['messages'][-1] if isinstance(last_message, AIMessage) and last_message.tool_calls: # 如果AI消息包含工具调用,则去执行工具 return "call_tool" else: # 否则,结束本次交互 return "end" def call_model(state: AgentState): """调用LLM,生成回复或工具调用请求。""" # 准备工具描述 tools_descriptions = "\n".join([f"- {tool.name}: {tool.description}" for tool in tools]) formatted_prompt = prompt.invoke({"messages": state['messages'], "tools_descriptions": tools_descriptions}) # 调用绑定了工具的LLM response = llm_with_tools.invoke(formatted_prompt.to_messages()) # 将LLM的响应添加到消息历史中 return {"messages": [response]} def call_tool(state: AgentState): """执行AI消息中指定的工具调用。""" last_message = state['messages'][-1] results = [] if isinstance(last_message, AIMessage) and last_message.tool_calls: for tool_call in last_message.tool_calls: tool_name = tool_call['name'] tool_args = tool_call['args'] # 根据工具名找到对应的工具函数 tool_to_use = next((tool for tool in tools if tool.name == tool_name), None) if tool_to_use: print(f"[Agent] 正在调用工具: {tool_name},参数: {tool_args}") try: # 执行工具 tool_result = tool_to_use.invoke(tool_args) results.append(ToolMessage(content=str(tool_result), tool_call_id=tool_call['id'])) except Exception as e: results.append(ToolMessage(content=f"工具调用出错: {e}", tool_call_id=tool_call['id'])) else: results.append(ToolMessage(content=f"未知工具: {tool_name}", tool_call_id=tool_call['id'])) else: # 如果没有工具调用,返回空结果 results = [ToolMessage(content="无工具调用需要执行。", tool_call_id="none")] return {"messages": results} # 5. 构建并编译图 def create_agent_graph(): workflow = StateGraph(AgentState) # 添加节点 workflow.add_node("model", call_model) workflow.add_node("tool", call_tool) # 设置入口点 workflow.set_entry_point("model") # 添加条件边 workflow.add_conditional_edges( "model", should_continue, { "call_tool": "tool", "end": END, } ) workflow.add_edge("tool", "model") # 工具执行完后,回到模型进行下一步推理 # 编译图 graph = workflow.compile() return graph # 创建图实例 agent_graph = create_agent_graph()3.5 运行你的第一个智能体
现在,让我们写一个主程序来测试这个智能体。
# main.py from agent.graph import agent_graph from langchain_core.messages import HumanMessage def run_agent_query(query: str): """向智能体提问并打印完整交互过程。""" print(f"\n{'='*50}") print(f"用户问题: {query}") print(f"{'='*50}") # 初始化状态 initial_state = { "messages": [HumanMessage(content=query)], "current_step": "start" } # 运行图 final_state = None for step, output in enumerate(agent_graph.stream(initial_state, stream_mode="values")): # `output` 是每次状态更新后的 `AgentState` last_message = output['messages'][-1] if isinstance(last_message, AIMessage): print(f"[步骤 {step+1}] AI 思考/回复: {last_message.content}") if last_message.tool_calls: print(f" 计划调用工具: {[tc['name'] for tc in last_message.tool_calls]}") elif isinstance(last_message, ToolMessage): print(f"[步骤 {step+1}] 工具返回: {last_message.content[:100]}...") # 截断显示 final_state = output print(f"\n{'='*50}") if final_state and final_state['messages']: final_ai_message = [m for m in final_state['messages'] if isinstance(m, AIMessage)][-1] print(f"最终答案: {final_ai_message.content}") print(f"{'='*50}\n") if __name__ == "__main__": # 测试几个问题 test_queries = [ "北京现在的天气怎么样?", "计算一下 (15 + 7) * 3 等于多少?", "请先搜索‘Python 最新版本’,然后告诉我现在几点了。", ] for query in test_queries: run_agent_query(query) input("按回车键继续下一个问题...")运行python main.py,你应该能看到智能体逐步思考、调用工具并给出最终答案的完整过程。这标志着你的智能体“大脑”(Qwen3.6-27B)和“身体”(LangGraph框架)已经成功连接并可以协同工作。
4. 测试开发(Test Development)阶段实践
在智能体基础功能跑通后,就进入了关键的“测试开发”阶段。这个阶段的目标不是简单的功能测试,而是构建一套自动化、可重复的测试体系,以确保智能体在各种场景下的行为符合预期,并为后续的 P1 功能开发提供质量保障。
4.1 智能体测试的挑战与策略
测试一个基于 LLM 的智能体比测试传统软件更复杂,因为其输出具有非确定性。我们的测试策略需要分层:
- 单元测试(Unit Testing):测试单个工具函数的功能是否正确。这是确定性的。
- 集成测试(Integration Testing):测试智能体工作流(图)是否能正确路由、调用工具并整合结果。
- 端到端测试(E2E Testing):用一组预定义的、覆盖不同场景的“提示词-期望”对来测试整个智能体系统。我们需要评估其输出的相关性、正确性和安全性。
4.2 构建一个简单的测试套件
我们使用pytest框架来组织测试。
首先,安装 pytest:pip install pytest。
然后,创建测试目录和文件。
# tests/test_tools.py import sys import os sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) from tools.calculator import calculator from tools.current_time import get_current_time def test_calculator_basic(): """测试计算器工具的基本功能。""" assert calculator.invoke("3 + 5") == "计算结果: 8" assert "计算结果: 6" in calculator.invoke("2 * 3") # 测试错误处理 assert "计算错误" in calculator.invoke("10 / 0") def test_calculator_with_math(): """测试计算器工具使用 math 模块。""" result = calculator.invoke("math.sqrt(16)") assert "计算结果: 4.0" in result def test_get_current_time(): """测试获取时间工具。""" result = get_current_time.invoke({}) assert "当前时间" in result assert "Asia/Shanghai" in result or "上海" in result# tests/test_agent_integration.py import sys import os sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) from langchain_core.messages import HumanMessage from agent.graph import agent_graph def test_agent_with_single_tool(): """测试智能体能够正确调用单一工具并返回结果。""" query = "123乘以456等于多少?" initial_state = {"messages": [HumanMessage(content=query)], "current_step": "start"} final_state = agent_graph.invoke(initial_state) # 检查最终消息是 AI 回复 final_message = final_state['messages'][-1] assert final_message.type == "ai" # 检查回复中是否包含计算结果(非精确匹配,因为LLM会组织语言) content = final_message.content.lower() # 123 * 456 = 56088 assert any(str(num) in content for num in [56088, "56088"]) or "计算" in content print(f"测试通过: {query} -> 回复包含计算结果") def test_agent_with_chained_tools(): """测试智能体能够按顺序调用多个工具。""" query = "先告诉我现在的时间,再计算一下 99 的平方。" initial_state = {"messages": [HumanMessage(content=query)], "current_step": "start"} final_state = agent_graph.invoke(initial_state) final_message = final_state['messages'][-1] content = final_message.content # 检查是否提到了时间和计算结果 # 这是一个宽松的检查,因为LLM的表述可能多样 assert ("时间" in content or "当前" in content) and ("9801" in content or "99" in content or "平方" in content) print(f"测试通过: {query} -> 回复涉及时间和计算") def test_agent_no_tool_needed(): """测试不需要工具调用的纯对话。""" query = "你好,请做一下自我介绍。" initial_state = {"messages": [HumanMessage(content=query)], "current_step": "start"} final_state = agent_graph.invoke(initial_state) final_message = final_state['messages'][-1] assert final_message.type == "ai" assert len(final_message.content) > 10 # 回复不应为空 print(f"测试通过: {query} -> 得到了AI回复")4.3 运行测试并分析
在项目根目录下运行:pytest tests/ -v
你会看到测试结果。对于集成测试,由于依赖真实的 LLM 调用,可能会比较慢,且结果有一定随机性。这正是测试智能体的难点。在实际项目中,我们可能需要:
- Mock LLM 响应:在集成测试中,用一个固定的、模拟的 LLM 来替换真实的模型调用,使测试变得确定和快速。
- 评估指标:使用更复杂的评估方法,如使用另一个 LLM(评判员)来评估输出是否满足要求,或计算输出与期望的嵌入向量相似度。
- 持续集成(CI):将测试套件集成到 CI/CD 流程中,定期运行。
通过这个测试开发阶段,我们验证了智能体核心流程的稳定性,为进入 P1 阶段(实现具体业务功能)打下了坚实的基础。
5. 迈向 P1 阶段:开发一个“天气查询与建议”智能体
假设我们的 P1 阶段目标是开发一个能为用户提供天气查询和出行建议的智能体。这需要集成真实的天气 API。
5.1 集成真实天气 API 工具
我们使用一个免费的天气 API(例如 Open-Meteo)来替换之前的模拟搜索工具。
# tools/real_weather.py from langchain.tools import tool import requests from datetime import datetime @tool def get_real_weather(city: str) -> str: """ 获取指定城市的真实天气信息。 参数 city: 城市名称,例如 'Beijing', 'Shanghai'。 """ # 这里使用 Open-Meteo 免费 API (无需密钥) # 首先,需要将城市名转换为经纬度(这里简化,使用预设字典) city_coordinates = { "beijing": {"latitude": 39.9042, "longitude": 116.4074}, "shanghai": {"latitude": 31.2304, "longitude": 121.4737}, "guangzhou": {"latitude": 23.1291, "longitude": 113.2644}, "shenzhen": {"latitude": 22.5431, "longitude": 114.0579}, "hangzhou": {"latitude": 30.2741, "longitude": 120.1551}, } city_lower = city.lower().strip() if city_lower not in city_coordinates: return f"抱歉,暂不支持城市 '{city}' 的查询。支持的城市有:{', '.join(city_coordinates.keys())}" coord = city_coordinates[city_lower] url = f"https://api.open-meteo.com/v1/forecast" params = { "latitude": coord["latitude"], "longitude": coord["longitude"], "current_weather": True, "timezone": "auto" } try: response = requests.get(url, params=params, timeout=10) response.raise_for_status() data = response.json() current = data.get("current_weather", {}) temperature = current.get("temperature") windspeed = current.get("windspeed") weathercode = current.get("weathercode") # 简化天气代码转换 weather_map = { 0: "晴", 1: "大部晴朗", 2: "局部有云", 3: "阴天", 45: "有雾", 48: "有雾", 51: "小雨", 53: "中雨", 55: "大雨", 61: "小雨", 63: "中雨", 65: "大雨", 80: "阵雨", 81: "大阵雨", 82: "强阵雨", 95: "雷暴", } weather_desc = weather_map.get(weathercode, "未知") return f"{city}当前天气:{weather_desc},温度 {temperature}°C,风速 {windspeed} km/h。" except requests.exceptions.RequestException as e: return f"获取天气信息失败:{e}"更新tools/__init__.py,将新工具加入列表。
5.2 更新智能体提示词与测试
更新系统提示词,明确告知智能体新的工具能力。
# 在 agent/graph.py 中更新 system_prompt system_prompt = """你是一个智能天气与生活助手。你可以通过工具获取实时天气、进行计算、查询时间。 请根据用户问题,智能地调用工具组合来提供最佳答案。 例如: 用户:“北京天气如何?适合出门吗?” 你应该:1. 调用 get_real_weather 工具查询北京天气。2. 根据天气情况(如温度、是否下雨)给出出行建议。 记住:一次只调用一个工具,根据结果再决定下一步。 可用工具: {tools_descriptions} """编写一个新的端到端测试用例。
# tests/test_p1_weather_agent.py import sys import os sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) from langchain_core.messages import HumanMessage from agent.graph import agent_graph def test_weather_query(): """测试智能体使用真实天气API。""" query = "上海现在的天气怎么样?" initial_state = {"messages": [HumanMessage(content=query)], "current_step": "start"} final_state = agent_graph.invoke(initial_state) final_message = final_state['messages'][-1] content = final_message.content # 检查回复中是否包含上海和天气关键词 assert "上海" in content assert any(word in content for word in ["天气", "温度", "°C", "晴", "云", "雨"]) print(f"P1 天气查询测试通过: {query}") print(f"AI回复: {content}") def test_weather_advice(): """测试智能体结合天气给出建议。""" query = "我看杭州今天下雨,我还需要带伞吗?" initial_state = {"messages": [HumanMessage(content=query)], "current_step": "start"} final_state = agent_graph.invoke(initial_state) final_message = final_state['messages'][-1] content = final_message.content # 检查回复是否合理(应该建议带伞或确认下雨) # 这是一个语义检查,比较宽松 assert any(word in content.lower() for word in ["带伞", "雨伞", "下雨", "需要", "建议"]) print(f"P1 天气建议测试通过: {query}") print(f"AI回复: {content}")运行新的测试:pytest tests/test_p1_weather_agent.py -v -s(-s用于打印 print 语句)。
至此,我们已经完成了一个具备真实天气查询功能的 P1 阶段智能体雏形。你可以在此基础上继续扩展,例如增加日程管理、邮件发送、数据库查询等更复杂的工具,构建功能更强大的智能体应用。
6. 常见问题与排查思路
在基于 Qwen3.6-27B 开发智能体的过程中,你可能会遇到以下典型问题:
| 问题现象 | 可能原因 | 排查与解决思路 |
|---|---|---|
| 模型服务启动失败,OOM(内存不足) | 1. 模型太大,显存不足。 2. --tensor-parallel-size设置过小。3. 未使用量化模型。 | 1. 使用nvidia-smi或npu-smi检查显存占用。2. 增大 --tensor-parallel-size至 GPU/NPU 数量。3. 换用量化版本(如 Qwen3.6-27B-w8a8或 AWQ/GPTQ 量化版)。4. 减小 --max-model-len。 |
| 智能体调用工具时失败或格式错误 | 1. 工具定义不符合 LangChain 格式。 2. LLM 的提示词未清晰指导工具调用。 3. 模型工具调用能力不足。 | 1. 检查工具函数的@tool装饰器和参数类型提示。2. 优化系统提示词,明确工具使用格式和时机。 3. 在调用 llm.bind_tools(tools)时,确保tools列表正确传入。4. 考虑使用更擅长工具调用的模型或进行微调。 |
| 智能体陷入死循环或重复调用工具 | 1. 状态判断逻辑 (should_continue) 有误。2. 工具返回结果未能让LLM识别为“已完成”。 | 1. 在call_model节点后打印 AI 消息,检查其tool_calls属性。2. 优化 should_continue逻辑,可以加入最大步数限制。3. 在工具返回结果中增加明确的完成标识。 |
| 测试不稳定,时而过时而不过 | 1. LLM 输出的非确定性。 2. 测试断言过于严格(精确字符串匹配)。 | 1. 为测试设置固定的随机种子 (temperature=0)。2. 采用模糊匹配、关键词匹配或语义相似度(如余弦相似度)进行断言。 3. 对非核心的“闲聊”类测试,可以只检查是否有回复,不检查具体内容。 |
| API 调用速度慢 | 1. 模型推理速度慢。 2. 网络延迟。 3. 工具本身是慢速 API(如网络请求)。 | 1. 考虑使用量化、更小的模型或性能更强的硬件。 2. 为工具调用设置超时和重试机制。 3. 对智能体工作流进行性能剖析,找出瓶颈。 |
7. 最佳实践与工程建议
- 环境隔离:始终使用虚拟环境(如
venv,conda)或 Docker 容器来管理项目依赖,避免污染系统环境。 - 配置管理:将模型端点、API密钥、超时设置等写入配置文件(如
.env)或配置中心,不要硬编码在代码中。 - 错误处理与日志:在智能体的每个关键节点(模型调用、工具执行、状态转换)添加详细的日志记录和健壮的错误处理(try-except)。这有助于快速定位生产环境的问题。
- 版本控制:对模型版本、框架版本、工具版本进行严格管理。任何升级都可能引入不兼容性。
- 提示词工程:系统提示词是智能体的“灵魂”。花时间精心设计和迭代你的提示词,使其目标明确、格式清晰、能有效引导模型使用工具。可以将提示词模板化,便于管理和 A/B 测试。
- 工具设计原则:
- 单一职责:每个工具只做一件事。
- 强类型:使用 Pydantic 模型来定义工具的输入参数,这能帮助 LLM 生成更准确的参数。
- 安全性:永远不要相信来自 LLM 的未经净化的输入(如
eval)。对工具进行输入验证和权限控制。
- 性能优化:
- 缓存:对频繁且结果不变的工具调用(如某些查询)实施缓存。
- 异步:如果工具是 I/O 密集型(如网络请求),考虑使用异步调用以提高整体吞吐量。
- 批处理:如果 vLLM 服务支持,可以尝试将多个用户请求批量发送给模型,以提高 GPU/NPU 利用率。
- 监控与评估:建立监控指标,如请求延迟、工具调用成功率、用户满意度等。定期用一组标准问题集(评估基准)来评估智能体性能的波动。
从 Qwen3.6-27B 的部署,到 LangGraph 智能体框架的搭建,再到测试体系的建立和 P1 功能开发,我们走完了一个智能体项目从零到一的核心路径。这条路径上的每一步——模型服务化、工具定义、工作流编排、测试验证——都是构建可靠、实用 AI 智能体的关键基石。接下来,你可以沿着这个框架,深入探索更复杂的工具集成(如数据库、企业内部 API)、多智能体协作、记忆(Memory)管理以及面向生产环境的部署与运维,将你的 AI 助手变得越来越强大和智能。