1. 项目缘起当“派活”遇上AI一个真实的生产力痛点最近在团队里我经常遇到一个挺烦人的场景手头正忙着写代码或者开会突然钉钉上就弹出一条消息是同事或者老板发来的一个链接、一个文档或者干脆就是一句“这个需求你看看”。这种“随时派活”的模式在快节奏的协作里很常见但打断深度工作流不说更麻烦的是很多“活”本身并不复杂比如“把这份会议纪要总结成三个要点”、“查一下上周项目A的进度数据”、“给这个英文技术名词找个准确的中文翻译”。为了这些小事我得停下手里的事打开浏览器、登录系统、复制粘贴、整理格式……一套流程下来十几分钟就没了思路也断了。我一直琢磨着能不能让钉钉这个“消息中心”变得更智能一点让它不仅能接收任务还能自己“消化”一部分简单任务直接把结果返回来。这不就是AI Agent智能体最擅长的事吗一个能理解自然语言指令、调用工具、并返回结果的自动化程序。市面上成熟的AI Agent框架不少但要么部署复杂要么对国内生态支持不够友好特别是和钉钉、飞书这类国内主流办公软件深度集成的方案公开的、能直接拿来用的并不多。直到我遇到了nanobot和通义千问这个组合我把它戏称为“国产小龙虾方案”——就像夏夜吃小龙虾食材模型和烹饪方式框架都是本土的搭配起来格外对味。nanobot是一个轻量级、可扩展的Python AI Agent框架而通义千问是阿里云出品的大语言模型对中文场景的理解和生成能力非常出色。更重要的是它们都能很好地跑在我自己的服务器上数据安全可控并且通过钉钉的机器人接口可以无缝嵌入到日常的聊天流中。这个项目的核心目标很简单在钉钉群里一个机器人用自然语言给它“派活”它能在后台自动调用通义千问进行分析、思考、执行比如搜索、总结、计算然后把最终结果直接回复到群里。下面我就把从零搭建这个“智能小助理”的完整过程、核心原理以及我踩过的坑毫无保留地分享出来。2. 技术栈选型为什么是 nanobot 通义千问在开始动手之前我们先拆解一下这个方案里的几个关键角色以及我为什么做出这样的选择。这有助于你理解整个系统的运作逻辑未来替换其中某个组件时也能心中有数。2.1 nanobot轻量灵活的AI Agent“骨架”nanobot 不是一个功能大而全的“重型”框架它的设计哲学是轻量和模块化。你可以把它理解为一个AI Agent的“骨架”或者“路由器”。它的核心职责是接收输入比如从钉钉机器人回调接口收到的用户消息。编排流程决定如何处理这条消息。是直接调用大模型还是先调用某个工具Tool管理上下文维护与用户对话的历史让AI能理解上下文语境。返回输出将大模型或工具产生的结果格式化后返回给调用方钉钉。我选择nanobot的主要原因有三点Python原生易于上手和调试Agent的核心逻辑用Python编写与我团队的技术栈一致调试和扩展非常方便。它的代码结构清晰没有过于复杂的抽象层。对“工具”调用支持友好nanobot内置了对OpenAI格式函数调用Function Calling的良好支持这使得为Agent定义各种技能Skills——比如调用搜索引擎API、查询数据库、执行命令行——变得非常直观。虽然我们用的是通义千问但其API也兼容OpenAI的格式这让整合工作简化了不少。社区活跃文档尚可作为一个较新的项目nanobot的社区在不断成长遇到问题时有地方可以讨论。其文档虽然不算极其详尽但核心概念和示例足以让人快速入门。注意nanobot本身不提供大模型能力它只是一个调度框架。你需要为它配备一个“大脑”这就是大语言模型LLM。2.2 通义千问本土化场景下的“最强大脑”为什么不用ChatGPT或Claude原因很现实网络稳定性、数据合规性、成本以及对中文语境的理解深度。通义千问作为国内头部模型在这几点上优势明显出色的中文能力在中文理解、生成、特别是涉及国内文化、网络用语、专业术语的场景下表现更加自然和准确。让它处理会议纪要、总结中文文档比用英文模型翻译后再处理要靠谱得多。稳定的API服务通过阿里云平台调用网络延迟低服务稳定有SLA保障不用担心随时可能中断。可控的成本与数据安全API调用按量计费清晰可控。更重要的是数据通过阿里云国内节点传输符合很多企业对数据不出境的要求。你也可以选择部署通义千问的开源模型如Qwen2.5到自己的GPU服务器上实现完全私有化。兼容OpenAI API通义千问的API设计很大程度上兼容了OpenAI的格式这意味着大量为ChatGPT设计的工具链和框架包括nanobot可以几乎无缝迁移极大地降低了开发门槛。在这个方案里通义千问扮演着“思考者”和“执行者”的角色。nanobot将用户的问题和可用的工具列表交给它它来思考“用户想干什么我需要用到哪个工具工具返回结果后我又该如何组织语言回复用户”2.3 钉钉机器人无缝嵌入工作流的“交互界面”钉钉机器人提供了标准的Webhook接入方式。我们创建一个自定义机器人后会得到一个带有签名的Webhook URL。当用户在群里这个机器人并发送消息时钉钉服务器会将消息内容以HTTP POST请求的形式推送到我们配置的服务器地址即我们的Agent服务。我们的nanobot服务就需要提供一个HTTP接口来接收这个回调解析出消息内容交给nanobot核心处理最后再将处理结果通过钉钉机器人提供的API发送回群里。这个过程是异步的为了不超时我们通常需要先快速回复一个“收到正在处理”然后再异步执行耗时任务并推送最终结果。这个组合形成了一个清晰的管道钉钉输入/输出 - nanobot流程编排 - 通义千问思考/规划 - 各种工具执行 - 通义千问总结 - nanobot格式化 - 钉钉输出。3. 环境搭建与核心组件部署理论讲完了我们开始动手。这一部分会非常详细包括环境准备、服务部署、配置细节确保你能跟着一步步复现。3.1 基础环境准备你需要准备一台具有公网IP的服务器或使用内网穿透工具用于部署我们的Agent服务。这里以Ubuntu 22.04 LTS为例。# 1. 更新系统并安装基础依赖 sudo apt update sudo apt upgrade -y sudo apt install -y python3-pip python3-venv git curl # 2. 创建项目目录并进入 mkdir -p ~/projects/dingtalk-agent cd ~/projects/dingtalk-agent # 3. 创建并激活Python虚拟环境强烈推荐避免包冲突 python3 -m venv venv source venv/bin/activate # 看到命令行前缀变成 (venv) 即表示激活成功3.2 安装与配置 nanobotnanobot可以通过pip直接安装。同时我们还需要安装用于构建HTTP服务接收钉钉回调的框架这里选择轻量级的FastAPI。# 在虚拟环境中安装核心包 pip install nanobot-sdk fastapi uvicorn httpx python-multipart # httpx用于异步HTTP客户端python-multipart用于解析钉钉POST请求接下来创建项目的基本结构dingtalk-agent/ ├── main.py # FastAPI应用主入口处理钉钉回调 ├── agent_core.py # nanobot Agent核心逻辑定义 ├── tools/ # 存放自定义工具Skills │ └── __init__.py │ └── web_search_tool.py ├── config.py # 配置文件存放API密钥等 ├── requirements.txt └── (其他日志等目录)首先创建config.py来管理敏感信息# config.py import os from dotenv import load_dotenv load_dotenv() # 从 .env 文件加载环境变量 # 通义千问 API 配置 (以阿里云灵积平台为例) DASHSCOPE_API_KEY os.getenv(DASHSCOPE_API_KEY) # 从环境变量读取 DASHSCOPE_API_BASE https://dashscope.aliyuncs.com/compatible-mode/v1 # 兼容OpenAI的端点 # 钉钉机器人配置 DINGTALK_WEBHOOK os.getenv(DINGTALK_WEBHOOK) # 钉钉机器人的Webhook地址 DINGTALK_SECRET os.getenv(DINGTALK_SECRET) # 钉钉机器人的加签密钥 # 其他工具API密钥如搜索引擎 SERPER_API_KEY os.getenv(SERPER_API_KEY) # 用于谷歌搜索的替代品然后在项目根目录创建.env文件切记不要提交到GitDASHSCOPE_API_KEYyour_dashscope_api_key_here DINGTALK_WEBHOOKhttps://oapi.dingtalk.com/robot/send?access_tokenxxx DINGTALK_SECRETyour_secret_here SERPER_API_KEYyour_serper_api_key_here3.3 创建 nanobot Agent 核心这是大脑的连接处。在agent_core.py中我们将初始化通义千问作为LLM并定义Agent。# agent_core.py import asyncio from typing import List, Optional from nanobot import Nanobot, LLM from nanobot.schema import Message import httpx from config import DASHSCOPE_API_KEY, DASHSCOPE_API_BASE class QwenLLM(LLM): 自定义通义千问LLM类适配nanobot框架 def __init__(self, api_key: str, base_url: str, model: str qwen-max): self.api_key api_key self.base_url base_url self.model model self.client httpx.AsyncClient( base_urlbase_url, headers{Authorization: fBearer {api_key}}, timeout30.0 ) async def generate(self, messages: List[Message], **kwargs) - str: 调用通义千问兼容API生成回复 # 将nanobot的Message格式转换为OpenAI API格式 openai_messages [{role: msg.role, content: msg.content} for msg in messages] payload { model: self.model, messages: openai_messages, stream: False, **kwargs # 可以传递 temperature, max_tokens 等参数 } try: response await self.client.post(/chat/completions, jsonpayload) response.raise_for_status() data response.json() return data[choices][0][message][content] except Exception as e: return f调用模型时出错: {str(e)} async def close(self): await self.client.aclose() def create_agent(): 创建并返回一个配置好的nanobot Agent实例 # 1. 初始化通义千问LLM llm QwenLLM( api_keyDASHSCOPE_API_KEY, base_urlDASHSCOPE_API_BASE, modelqwen-max # 可根据需要选择 qwen-plus, qwen-max 等 ) # 2. 创建Agent并传入LLM agent Nanobot(llmllm) # 3. (后续步骤) 在这里注册工具Skills # 例如agent.register_tool(web_search_tool) return agent # 创建一个全局Agent实例避免每次请求都创建 global_agent None async def get_agent(): 获取或创建全局Agent单例模式 global global_agent if global_agent is None: global_agent create_agent() return global_agent async def close_agent(): 关闭Agent释放资源 global global_agent if global_agent: await global_agent.llm.close() global_agent None这段代码的关键点在于QwenLLM类它实现了nanobot框架期望的LLM接口内部使用httpx异步客户端调用通义千问的兼容端点。这样nanobot就可以像使用OpenAI一样使用通义千问了。3.4 为Agent添加“技能”自定义工具Tools一个只会聊天的Agent用处有限关键是它能“做事”。我们需要为它定义工具。这里以一个“网络搜索”工具为例它能让Agent回答实时性问题。首先安装搜索工具依赖这里用Serper API一个性价比高的谷歌搜索API替代品pip install aiohttp创建tools/web_search_tool.py# tools/web_search_tool.py import aiohttp import json from typing import Dict, Any from nanobot import Tool from config import SERPER_API_KEY class WebSearchTool(Tool): 一个简单的网络搜索工具 name web_search description 使用此工具在互联网上搜索最新信息。当用户询问需要实时数据、新闻、最新事件或你不知道的信息时使用此工具。输入应为搜索查询字符串。 async def run(self, query: str) - str: 执行搜索并返回格式化结果 if not SERPER_API_KEY: return 错误未配置搜索API密钥。 url https://google.serper.dev/search headers { X-API-KEY: SERPER_API_KEY, Content-Type: application/json } payload json.dumps({q: query, gl: cn}) # gl参数可指定国家/地区 async with aiohttp.ClientSession() as session: try: async with session.post(url, headersheaders, datapayload) as resp: if resp.status 200: data await resp.json() # 简化处理提取有机搜索结果 results data.get(organic, []) if not results: return f未找到关于 {query} 的相关结果。 # 拼接前3个结果 summary f关于 {query} 的搜索结果\n\n for i, item in enumerate(results[:3], 1): title item.get(title, 无标题) link item.get(link, #) snippet item.get(snippet, 无摘要) summary f{i}. **{title}**\n {snippet}\n 链接{link}\n\n return summary else: return f搜索API请求失败状态码{resp.status} except Exception as e: return f搜索过程中发生异常{str(e)} # 工具实例 web_search_tool WebSearchTool()这个工具类继承自nanobot.Tool必须定义name、description和run方法。description非常重要它是给大模型看的“说明书”模型通过阅读它来决定是否以及如何调用这个工具。描述要清晰准确。现在我们需要在agent_core.py的create_agent函数中注册这个工具# agent_core.py (更新部分) from tools.web_search_tool import web_search_tool def create_agent(): llm QwenLLM(...) # 同上 agent Nanobot(llmllm) # 注册工具 agent.register_tool(web_search_tool) return agent这样当用户问“今天北京天气怎么样”时nanobot会将对话历史和工具描述发给通义千问通义千问可能会判断需要调用web_search工具并生成类似{tool_name: web_search, arguments: {query: 北京今天天气}}的调用请求。nanobot收到后执行工具将结果返回给通义千问最后由通义千问组织成自然语言回复。4. 钉钉机器人集成与消息处理Agent准备好了现在要让它能听得到、回得了钉钉的消息。4.1 创建钉钉群自定义机器人打开任意钉钉群 - 点击右上角设置图标 - 「智能群助手」。点击「添加机器人」- 「自定义机器人」。设置机器人名字如“AI小助理”、头像并选择添加到哪个群。安全设置至关重要务必选择“加签”。复制生成的secret即DINGTALK_SECRET。同时你也会获得一个Webhook地址其中包含access_token即DINGTALK_WEBHOOK。将secret和完整的Webhook地址填入你的.env文件。4.2 实现钉钉回调接口与签名验证钉钉为了安全要求对机器人收到的回调进行签名验证。我们需要在main.py中实现这个逻辑。# main.py import hashlib import hmac import base64 import urllib.parse import time import json from fastapi import FastAPI, Request, HTTPException, BackgroundTasks from fastapi.responses import JSONResponse import httpx from agent_core import get_agent, close_agent from config import DINGTALK_SECRET, DINGTALK_WEBHOOK app FastAPI(titleDingTalk AI Agent) def verify_dingtalk_signature(timestamp: str, sign: str, secret: str, body: bytes) - bool: 验证钉钉机器人回调签名 if not timestamp or not sign: return False # 钉钉要求如果时间戳与当前时间相差超过1小时则拒绝 if abs(int(time.time() * 1000) - int(timestamp)) 3600 * 1000: return False string_to_sign f{timestamp}\n{secret} hmac_code hmac.new( secret.encode(utf-8), string_to_sign.encode(utf-8), digestmodhashlib.sha256 ).digest() my_sign base64.b64encode(hmac_code).decode(utf-8) return hmac.compare_digest(my_sign, sign) async def send_dingtalk_message(text: str, at_all: bool False, at_user_ids: list None): 通过钉钉机器人Webhook发送消息 headers {Content-Type: application/json} payload { msgtype: text, text: { content: text }, at: { isAtAll: at_all, atUserIds: at_user_ids or [] } } async with httpx.AsyncClient() as client: try: # 注意钉钉要求对Webhook URL进行签名 timestamp str(int(time.time() * 1000)) secret DINGTALK_SECRET string_to_sign f{timestamp}\n{secret} hmac_code hmac.new(secret.encode(utf-8), string_to_sign.encode(utf-8), hashlib.sha256).digest() sign urllib.parse.quote_plus(base64.b64encode(hmac_code).decode(utf-8)) webhook_url f{DINGTALK_WEBHOOK}timestamp{timestamp}sign{sign} resp await client.post(webhook_url, jsonpayload, headersheaders) resp.raise_for_status() except Exception as e: print(f发送钉钉消息失败: {e}) async def process_agent_task(content: str, sender_id: str): 后台任务调用Agent处理消息并回复 try: agent await get_agent() # 这里简化处理将用户消息直接传给Agent # 在实际中你可能需要构建更复杂的对话历史管理 response await agent.run(content) # 将Agent的回复发送回钉钉并发送者 await send_dingtalk_message(f{response}\n\n—— 来自AI小助理, at_user_ids[sender_id]) except Exception as e: error_msg f处理您的请求时出现错误{str(e)} await send_dingtalk_message(error_msg, at_user_ids[sender_id]) app.post(/dingtalk/callback) async def dingtalk_callback(request: Request, background_tasks: BackgroundTasks): 钉钉机器人消息回调入口 # 1. 获取签名和参数 timestamp request.headers.get(timestamp) sign request.headers.get(sign) body_bytes await request.body() # 2. 验证签名 if not verify_dingtalk_signature(timestamp, sign, DINGTALK_SECRET, body_bytes): raise HTTPException(status_code403, detail签名验证失败) # 3. 解析消息体 try: body json.loads(body_bytes.decode(utf-8)) msg_type body.get(msgtype) text_content body.get(text, {}).get(content, ).strip() sender_id body.get(senderId) # 发送者钉钉ID conversation_id body.get(conversationId) # 4. 只处理文本消息并过滤掉机器人自己的消息防止循环 if msg_type ! text or [AI小助理] in text_content: return JSONResponse(content{msg: ignore}) # 5. 立即回复“正在处理”避免钉钉超时钉钉要求5秒内响应 # 这里先快速返回一个成功响应 response JSONResponse(content{msg: ok}) # 6. 将实际处理任务放入后台异步执行 background_tasks.add_task(process_agent_task, text_content, sender_id) return response except json.JSONDecodeError: raise HTTPException(status_code400, detail无效的JSON数据) app.on_event(shutdown) async def shutdown_event(): 应用关闭时清理资源 await close_agent() if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)这个main.py是服务的核心入口verify_dingtalk_signature严格按照钉钉文档实现签名验证这是安全的第一道关卡。send_dingtalk_message用于主动向钉钉群发送消息。注意发送时也需要对Webhook URL进行实时签名。process_agent_task真正的业务逻辑处理函数。它从后台任务中调用获取Agent实例运行用户消息并将结果发回钉钉。使用BackgroundTasks是关键因为Agent处理尤其是调用大模型和网络搜索可能需要几秒到十几秒远超钉钉回调要求的5秒超时限制。我们必须先快速返回一个成功响应再在后台异步处理。/dingtalk/callback钉钉消息推送的接收端点。它验证签名、解析消息然后触发后台任务。4.3 部署与启动服务现在我们可以启动服务了。在生产环境建议使用systemd或supervisor来管理进程。这里先以开发模式启动# 在项目根目录下 uvicorn main:app --host 0.0.0.0 --port 8000 --reload服务启动后你会看到类似Uvicorn running on http://0.0.0.0:8000的输出。由于钉钉需要回调你的公网IP你需要在路由器或云服务器安全组中开放8000端口并将http://你的公网IP:8000/dingtalk/callback这个地址配置到钉钉机器人的“消息接收地址”中在机器人设置页面。配置钉钉机器人回到之前创建机器人的设置页面找到“消息接收地址”。填入http://你的公网IP或域名:8000/dingtalk/callback。点击保存。钉钉会向这个地址发送一条验证请求如果你的服务签名验证通过并返回了正确的响应配置就成功了。5. 实战测试与效果优化一切就绪是时候在钉钉群里你的机器人试试了。5.1 基础功能测试在钉钉群里输入AI小助理 你好介绍一下你自己。稍等片刻你应该会收到回复内容大致是通义千问生成的自我介绍表明它是一个AI助手。再测试一下工具调用AI小助理 搜索一下今天科技圈有什么重要新闻。如果配置正确nanobot会指示通义千问调用web_search工具工具会调用Serper API进行搜索并将结果返回给通义千问进行总结最终你将收到一个包含几条新闻摘要和链接的回复。5.2 效果优化与高级技巧基础的跑通只是第一步要让这个“小助理”真正好用还需要不少优化。5.1 提升工具描述的准确性工具的description是模型决定是否调用的关键。模糊的描述会导致误调用或不调用。例如对于搜索工具我最初的描述是“搜索信息”结果模型经常在不需要实时信息时也调用它。后来我优化为“当用户询问需要实时数据、新闻、最新事件或你不知道的信息时使用此工具”准确率大幅提升。给你的工具写描述就像给新同事写岗位说明书越具体越好。5.2 实现对话历史管理目前的agent.run(content)是单轮对话Agent没有上下文记忆。在实际使用中用户可能会说“上一条新闻的详细内容呢”。我们需要让Agent记住之前的对话。nanobot支持在run方法中传入message_history。我们需要修改process_agent_task为每个用户或每个会话维护一个历史消息列表。一个简单的实现是使用字典在内存中缓存生产环境建议用Redis# 在 agent_core.py 或单独模块中 from collections import defaultdict from nanobot.schema import Message conversation_history defaultdict(list) # key: conversation_id, value: list of Messages async def process_agent_task(content: str, sender_id: str, conversation_id: str): agent await get_agent() history conversation_history.get(conversation_id, []) # 将用户新消息加入历史 history.append(Message(roleuser, contentcontent)) # 调用Agent传入历史 response await agent.run(history) # 将AI回复加入历史 history.append(Message(roleassistant, contentresponse)) # 限制历史长度防止token超限 if len(history) 10: # 保留最近5轮对话10条消息 history history[-10:] conversation_history[conversation_id] history await send_dingtalk_message(response, at_user_ids[sender_id])5.3 处理复杂任务与规划能力简单的问答和工具调用够了但用户可能想完成多步骤任务比如“帮我查一下杭州明天的天气如果是晴天就提醒我洗车如果是雨天就提醒我带伞”。这需要Agent具备规划能力。nanobot本身提供了基础的规划能力通过LLM的思维链Chain-of-Thought来实现。我们可以通过设计更精细的System Prompt系统指令来引导模型。在创建Agent时可以传入初始系统消息def create_agent(): llm QwenLLM(...) system_prompt 你是一个高效的AI助手集成在钉钉中。你的能力包括 1. 通用对话和问答。 2. 使用web_search工具获取实时信息。 3. 对于复杂任务请一步步思考并可以多次调用工具。 请用中文回复保持友好和专业。 initial_message Message(rolesystem, contentsystem_prompt) agent Nanobot(llmllm, initial_messages[initial_message]) agent.register_tool(web_search_tool) return agent一个强大的System Prompt能极大地塑造Agent的行为。你可以在这里定义它的角色、能力边界、回复格式等。5.4 增加更多实用工具一个搜索工具远远不够。我们可以根据团队需求开发更多工具内部知识库查询工具连接团队的Confluence、Wiki或数据库让AI能回答内部流程、项目信息。数据查询工具连接内部BI系统或数据库让AI能回答“上个月销售额是多少”这类问题。任务创建工具连接Jira、Teambition等实现“小助理 创建一个关于登录页面优化的任务指派给张三”。代码片段生成/解释工具针对技术团队可以集成代码模型的能力。每个工具的开发和注册模式都与web_search_tool类似关键是设计好输入输出和给模型的描述。5.5 错误处理与用户体验网络可能波动API可能超限模型可能抽风。良好的错误处理至关重要。在工具run方法中做好异常捕获返回友好的错误信息给模型而不是抛出异常导致整个流程中断。设置合理的超时时间在QwenLLM和工具中配置timeout避免一个请求卡住整个服务。提供降级方案如果搜索工具失败可以尝试让模型基于已有知识回答并提示“以下信息可能不是最新的”。在钉钉回复中加入状态提示对于耗时较长的任务可以在处理中、处理完成、处理失败时发送不同的状态消息。6. 部署、监控与成本考量6.1 生产环境部署开发模式的uvicorn不适合生产。建议使用Gunicorn with Uvicorn Workers提高并发能力。反向代理Nginx处理SSL、负载均衡和静态文件。进程管理systemd保证服务开机自启和自动重启。一个简单的systemd服务文件示例 (/etc/systemd/system/dingtalk-agent.service)[Unit] DescriptionDingTalk AI Agent Service Afternetwork.target [Service] Useryour_username Groupyour_groupname WorkingDirectory/home/your_username/projects/dingtalk-agent EnvironmentPATH/home/your_username/projects/dingtalk-agent/venv/bin ExecStart/home/your_username/projects/dingtalk-agent/venv/bin/gunicorn \ -w 4 \ -k uvicorn.workers.UvicornWorker \ --bind 0.0.0.0:8000 \ main:app Restartalways RestartSec10 [Install] WantedBymulti-user.target然后使用sudo systemctl start dingtalk-agent启动服务。6.2 日志与监控日志是排查问题的生命线。使用Python的logging模块在关键节点收到请求、调用模型、调用工具、发送回复记录信息。可以将日志输出到文件并使用logrotate管理。监控方面需要关注服务健康HTTP端点的可用性可用crontab定时curl。API调用情况通义千问API的调用次数、Token消耗、错误率。阿里云控制台有详细监控。工具调用延迟搜索、数据库查询等工具的响应时间。钉钉消息成功率消息发送失败的情况。6.3 成本分析与优化成本主要来自两块通义千问API费用按Token消耗计费。qwen-max比qwen-plus能力强但也更贵。可以根据任务复杂度选择模型。对于简单的分类、总结任务可以用便宜模型对于需要复杂推理的再用强模型。也可以在Agent逻辑里实现这种路由。第三方工具API费用如Serper搜索API、数据库查询可能产生的费用。优化建议缓存对常见、结果变化不频繁的查询如“公司请假流程是什么”进行缓存可以缓存模型回复也可以缓存工具结果。设置用量限额在代码层面或云平台设置每日/每月API调用上限防止意外消耗。精简上下文合理管理对话历史长度过长的历史会消耗大量Token。考虑私有化部署如果使用量非常大可以考虑在本地GPU服务器上部署通义千问的开源模型如Qwen2.5-7B/14B一次性硬件投入后调用成本几乎为零。7. 避坑指南我踩过的那些“坑”这个项目从构思到稳定运行我遇到了不少问题这里总结几个典型的希望能帮你绕过去。7.1 钉钉签名验证失败这是第一个拦路虎。现象是钉钉后台配置回调地址时一直报“验证失败”。坑1时间戳问题。钉钉要求服务器时间与网络时间同步。我最初服务器时间慢了几分钟导致签名验证永远失败。务必使用ntpdate或chronyd同步时间。坑2签名计算错误。钉钉的签名字符串是timestamp \n secret这个\n是换行符必须包含。我一开始用timestamp secret拼接怎么都对不上。仔细对照官方文档的示例代码才发现这个细节。坑3URL编码。计算出的签名sign需要经过URL编码后拼接到Webhook URL上。urllib.parse.quote_plus用在这里正合适它能正确处理加号等特殊字符。7.2 通义千问API调用返回空或错误坑模型名称错误或区域不对。通义千问在不同区域杭州、上海等的端点可能略有不同模型名如qwen-max,qwen-plus也要确保正确。最好先在阿里云控制台的“模型体验”或通过简单的curl命令测试API密钥和模型是否可用。坑Token超限。通义千问API有每分钟/每天的调用频率限制和Token数量限制。如果突然大量调用会被限流。需要在代码中加入简单的限流和重试机制并监控调用量。7.3 Agent“乱用”或“不用”工具坑工具描述太笼统。如前所述这是最常见的问题。把工具描述当成给一个“死板但聪明”的新员工写操作手册场景、输入、输出格式都要说清楚。坑System Prompt冲突。如果你的System Prompt里说“你是一个聊天机器人不要进行网络操作”那模型很可能就不会调用搜索工具。System Prompt和工具描述要协同工作明确告诉模型“在需要时你可以使用以下工具”。坑模型温度temperature过高。温度参数控制输出的随机性。太高如0.9可能导致模型行为不稳定有时调用工具有时不调用。对于任务型Agent建议设置较低的温度如0.1或0.2使其输出更确定、更可预测。7.4 异步处理与超时坑同步处理导致钉钉超时。这是架构上的关键点。钉钉回调要求5秒内响应而LLM调用工具调用很容易超过5秒。必须使用BackgroundTasks或消息队列如Celery进行异步处理主线程立即返回“成功接收”。坑后台任务异常导致用户无反馈。如果后台任务process_agent_task崩溃了用户就再也收不到回复。一定要在这个函数内部做好全面的异常捕获至少记录日志并考虑在失败时通过钉钉给用户发送一个错误提示。7.5 安全性问题坑暴露内部信息。Agent能调用搜索工具如果用户诱导它搜索“公司内网地址”、“内部系统密码”怎么办虽然工具本身可能搜不到但需要在System Prompt里明确禁止并考虑对用户输入进行基础的关键词过滤。更重要的不要给Agent连接具有高权限的内部系统工具。遵循最小权限原则为Agent创建专用的、只有只读权限或受限权限的账户。坑Webhook端点被恶意调用。虽然有了签名验证但也要防止重放攻击签名被截获后重复使用。我们的代码已经通过校验时间戳1小时窗口来缓解这一问题。在生产环境还可以考虑增加IP白名单如果钉钉出口IP固定或额外的自定义Token验证。经过以上这些步骤一个能够“随时派活”的钉钉AI智能体就真正搭建完成了。它不再是一个玩具而是一个能融入团队日常工作流、切实提升效率的生产力工具。你可以根据自己团队的需求不断为它添加新的“技能”让它变得越来越能干。
1. 项目缘起当“派活”遇上AI一个真实的生产力痛点最近在团队里我经常遇到一个挺烦人的场景手头正忙着写代码或者开会突然钉钉上就弹出一条消息是同事或者老板发来的一个链接、一个文档或者干脆就是一句“这个需求你看看”。这种“随时派活”的模式在快节奏的协作里很常见但打断深度工作流不说更麻烦的是很多“活”本身并不复杂比如“把这份会议纪要总结成三个要点”、“查一下上周项目A的进度数据”、“给这个英文技术名词找个准确的中文翻译”。为了这些小事我得停下手里的事打开浏览器、登录系统、复制粘贴、整理格式……一套流程下来十几分钟就没了思路也断了。我一直琢磨着能不能让钉钉这个“消息中心”变得更智能一点让它不仅能接收任务还能自己“消化”一部分简单任务直接把结果返回来。这不就是AI Agent智能体最擅长的事吗一个能理解自然语言指令、调用工具、并返回结果的自动化程序。市面上成熟的AI Agent框架不少但要么部署复杂要么对国内生态支持不够友好特别是和钉钉、飞书这类国内主流办公软件深度集成的方案公开的、能直接拿来用的并不多。直到我遇到了nanobot和通义千问这个组合我把它戏称为“国产小龙虾方案”——就像夏夜吃小龙虾食材模型和烹饪方式框架都是本土的搭配起来格外对味。nanobot是一个轻量级、可扩展的Python AI Agent框架而通义千问是阿里云出品的大语言模型对中文场景的理解和生成能力非常出色。更重要的是它们都能很好地跑在我自己的服务器上数据安全可控并且通过钉钉的机器人接口可以无缝嵌入到日常的聊天流中。这个项目的核心目标很简单在钉钉群里一个机器人用自然语言给它“派活”它能在后台自动调用通义千问进行分析、思考、执行比如搜索、总结、计算然后把最终结果直接回复到群里。下面我就把从零搭建这个“智能小助理”的完整过程、核心原理以及我踩过的坑毫无保留地分享出来。2. 技术栈选型为什么是 nanobot 通义千问在开始动手之前我们先拆解一下这个方案里的几个关键角色以及我为什么做出这样的选择。这有助于你理解整个系统的运作逻辑未来替换其中某个组件时也能心中有数。2.1 nanobot轻量灵活的AI Agent“骨架”nanobot 不是一个功能大而全的“重型”框架它的设计哲学是轻量和模块化。你可以把它理解为一个AI Agent的“骨架”或者“路由器”。它的核心职责是接收输入比如从钉钉机器人回调接口收到的用户消息。编排流程决定如何处理这条消息。是直接调用大模型还是先调用某个工具Tool管理上下文维护与用户对话的历史让AI能理解上下文语境。返回输出将大模型或工具产生的结果格式化后返回给调用方钉钉。我选择nanobot的主要原因有三点Python原生易于上手和调试Agent的核心逻辑用Python编写与我团队的技术栈一致调试和扩展非常方便。它的代码结构清晰没有过于复杂的抽象层。对“工具”调用支持友好nanobot内置了对OpenAI格式函数调用Function Calling的良好支持这使得为Agent定义各种技能Skills——比如调用搜索引擎API、查询数据库、执行命令行——变得非常直观。虽然我们用的是通义千问但其API也兼容OpenAI的格式这让整合工作简化了不少。社区活跃文档尚可作为一个较新的项目nanobot的社区在不断成长遇到问题时有地方可以讨论。其文档虽然不算极其详尽但核心概念和示例足以让人快速入门。注意nanobot本身不提供大模型能力它只是一个调度框架。你需要为它配备一个“大脑”这就是大语言模型LLM。2.2 通义千问本土化场景下的“最强大脑”为什么不用ChatGPT或Claude原因很现实网络稳定性、数据合规性、成本以及对中文语境的理解深度。通义千问作为国内头部模型在这几点上优势明显出色的中文能力在中文理解、生成、特别是涉及国内文化、网络用语、专业术语的场景下表现更加自然和准确。让它处理会议纪要、总结中文文档比用英文模型翻译后再处理要靠谱得多。稳定的API服务通过阿里云平台调用网络延迟低服务稳定有SLA保障不用担心随时可能中断。可控的成本与数据安全API调用按量计费清晰可控。更重要的是数据通过阿里云国内节点传输符合很多企业对数据不出境的要求。你也可以选择部署通义千问的开源模型如Qwen2.5到自己的GPU服务器上实现完全私有化。兼容OpenAI API通义千问的API设计很大程度上兼容了OpenAI的格式这意味着大量为ChatGPT设计的工具链和框架包括nanobot可以几乎无缝迁移极大地降低了开发门槛。在这个方案里通义千问扮演着“思考者”和“执行者”的角色。nanobot将用户的问题和可用的工具列表交给它它来思考“用户想干什么我需要用到哪个工具工具返回结果后我又该如何组织语言回复用户”2.3 钉钉机器人无缝嵌入工作流的“交互界面”钉钉机器人提供了标准的Webhook接入方式。我们创建一个自定义机器人后会得到一个带有签名的Webhook URL。当用户在群里这个机器人并发送消息时钉钉服务器会将消息内容以HTTP POST请求的形式推送到我们配置的服务器地址即我们的Agent服务。我们的nanobot服务就需要提供一个HTTP接口来接收这个回调解析出消息内容交给nanobot核心处理最后再将处理结果通过钉钉机器人提供的API发送回群里。这个过程是异步的为了不超时我们通常需要先快速回复一个“收到正在处理”然后再异步执行耗时任务并推送最终结果。这个组合形成了一个清晰的管道钉钉输入/输出 - nanobot流程编排 - 通义千问思考/规划 - 各种工具执行 - 通义千问总结 - nanobot格式化 - 钉钉输出。3. 环境搭建与核心组件部署理论讲完了我们开始动手。这一部分会非常详细包括环境准备、服务部署、配置细节确保你能跟着一步步复现。3.1 基础环境准备你需要准备一台具有公网IP的服务器或使用内网穿透工具用于部署我们的Agent服务。这里以Ubuntu 22.04 LTS为例。# 1. 更新系统并安装基础依赖 sudo apt update sudo apt upgrade -y sudo apt install -y python3-pip python3-venv git curl # 2. 创建项目目录并进入 mkdir -p ~/projects/dingtalk-agent cd ~/projects/dingtalk-agent # 3. 创建并激活Python虚拟环境强烈推荐避免包冲突 python3 -m venv venv source venv/bin/activate # 看到命令行前缀变成 (venv) 即表示激活成功3.2 安装与配置 nanobotnanobot可以通过pip直接安装。同时我们还需要安装用于构建HTTP服务接收钉钉回调的框架这里选择轻量级的FastAPI。# 在虚拟环境中安装核心包 pip install nanobot-sdk fastapi uvicorn httpx python-multipart # httpx用于异步HTTP客户端python-multipart用于解析钉钉POST请求接下来创建项目的基本结构dingtalk-agent/ ├── main.py # FastAPI应用主入口处理钉钉回调 ├── agent_core.py # nanobot Agent核心逻辑定义 ├── tools/ # 存放自定义工具Skills │ └── __init__.py │ └── web_search_tool.py ├── config.py # 配置文件存放API密钥等 ├── requirements.txt └── (其他日志等目录)首先创建config.py来管理敏感信息# config.py import os from dotenv import load_dotenv load_dotenv() # 从 .env 文件加载环境变量 # 通义千问 API 配置 (以阿里云灵积平台为例) DASHSCOPE_API_KEY os.getenv(DASHSCOPE_API_KEY) # 从环境变量读取 DASHSCOPE_API_BASE https://dashscope.aliyuncs.com/compatible-mode/v1 # 兼容OpenAI的端点 # 钉钉机器人配置 DINGTALK_WEBHOOK os.getenv(DINGTALK_WEBHOOK) # 钉钉机器人的Webhook地址 DINGTALK_SECRET os.getenv(DINGTALK_SECRET) # 钉钉机器人的加签密钥 # 其他工具API密钥如搜索引擎 SERPER_API_KEY os.getenv(SERPER_API_KEY) # 用于谷歌搜索的替代品然后在项目根目录创建.env文件切记不要提交到GitDASHSCOPE_API_KEYyour_dashscope_api_key_here DINGTALK_WEBHOOKhttps://oapi.dingtalk.com/robot/send?access_tokenxxx DINGTALK_SECRETyour_secret_here SERPER_API_KEYyour_serper_api_key_here3.3 创建 nanobot Agent 核心这是大脑的连接处。在agent_core.py中我们将初始化通义千问作为LLM并定义Agent。# agent_core.py import asyncio from typing import List, Optional from nanobot import Nanobot, LLM from nanobot.schema import Message import httpx from config import DASHSCOPE_API_KEY, DASHSCOPE_API_BASE class QwenLLM(LLM): 自定义通义千问LLM类适配nanobot框架 def __init__(self, api_key: str, base_url: str, model: str qwen-max): self.api_key api_key self.base_url base_url self.model model self.client httpx.AsyncClient( base_urlbase_url, headers{Authorization: fBearer {api_key}}, timeout30.0 ) async def generate(self, messages: List[Message], **kwargs) - str: 调用通义千问兼容API生成回复 # 将nanobot的Message格式转换为OpenAI API格式 openai_messages [{role: msg.role, content: msg.content} for msg in messages] payload { model: self.model, messages: openai_messages, stream: False, **kwargs # 可以传递 temperature, max_tokens 等参数 } try: response await self.client.post(/chat/completions, jsonpayload) response.raise_for_status() data response.json() return data[choices][0][message][content] except Exception as e: return f调用模型时出错: {str(e)} async def close(self): await self.client.aclose() def create_agent(): 创建并返回一个配置好的nanobot Agent实例 # 1. 初始化通义千问LLM llm QwenLLM( api_keyDASHSCOPE_API_KEY, base_urlDASHSCOPE_API_BASE, modelqwen-max # 可根据需要选择 qwen-plus, qwen-max 等 ) # 2. 创建Agent并传入LLM agent Nanobot(llmllm) # 3. (后续步骤) 在这里注册工具Skills # 例如agent.register_tool(web_search_tool) return agent # 创建一个全局Agent实例避免每次请求都创建 global_agent None async def get_agent(): 获取或创建全局Agent单例模式 global global_agent if global_agent is None: global_agent create_agent() return global_agent async def close_agent(): 关闭Agent释放资源 global global_agent if global_agent: await global_agent.llm.close() global_agent None这段代码的关键点在于QwenLLM类它实现了nanobot框架期望的LLM接口内部使用httpx异步客户端调用通义千问的兼容端点。这样nanobot就可以像使用OpenAI一样使用通义千问了。3.4 为Agent添加“技能”自定义工具Tools一个只会聊天的Agent用处有限关键是它能“做事”。我们需要为它定义工具。这里以一个“网络搜索”工具为例它能让Agent回答实时性问题。首先安装搜索工具依赖这里用Serper API一个性价比高的谷歌搜索API替代品pip install aiohttp创建tools/web_search_tool.py# tools/web_search_tool.py import aiohttp import json from typing import Dict, Any from nanobot import Tool from config import SERPER_API_KEY class WebSearchTool(Tool): 一个简单的网络搜索工具 name web_search description 使用此工具在互联网上搜索最新信息。当用户询问需要实时数据、新闻、最新事件或你不知道的信息时使用此工具。输入应为搜索查询字符串。 async def run(self, query: str) - str: 执行搜索并返回格式化结果 if not SERPER_API_KEY: return 错误未配置搜索API密钥。 url https://google.serper.dev/search headers { X-API-KEY: SERPER_API_KEY, Content-Type: application/json } payload json.dumps({q: query, gl: cn}) # gl参数可指定国家/地区 async with aiohttp.ClientSession() as session: try: async with session.post(url, headersheaders, datapayload) as resp: if resp.status 200: data await resp.json() # 简化处理提取有机搜索结果 results data.get(organic, []) if not results: return f未找到关于 {query} 的相关结果。 # 拼接前3个结果 summary f关于 {query} 的搜索结果\n\n for i, item in enumerate(results[:3], 1): title item.get(title, 无标题) link item.get(link, #) snippet item.get(snippet, 无摘要) summary f{i}. **{title}**\n {snippet}\n 链接{link}\n\n return summary else: return f搜索API请求失败状态码{resp.status} except Exception as e: return f搜索过程中发生异常{str(e)} # 工具实例 web_search_tool WebSearchTool()这个工具类继承自nanobot.Tool必须定义name、description和run方法。description非常重要它是给大模型看的“说明书”模型通过阅读它来决定是否以及如何调用这个工具。描述要清晰准确。现在我们需要在agent_core.py的create_agent函数中注册这个工具# agent_core.py (更新部分) from tools.web_search_tool import web_search_tool def create_agent(): llm QwenLLM(...) # 同上 agent Nanobot(llmllm) # 注册工具 agent.register_tool(web_search_tool) return agent这样当用户问“今天北京天气怎么样”时nanobot会将对话历史和工具描述发给通义千问通义千问可能会判断需要调用web_search工具并生成类似{tool_name: web_search, arguments: {query: 北京今天天气}}的调用请求。nanobot收到后执行工具将结果返回给通义千问最后由通义千问组织成自然语言回复。4. 钉钉机器人集成与消息处理Agent准备好了现在要让它能听得到、回得了钉钉的消息。4.1 创建钉钉群自定义机器人打开任意钉钉群 - 点击右上角设置图标 - 「智能群助手」。点击「添加机器人」- 「自定义机器人」。设置机器人名字如“AI小助理”、头像并选择添加到哪个群。安全设置至关重要务必选择“加签”。复制生成的secret即DINGTALK_SECRET。同时你也会获得一个Webhook地址其中包含access_token即DINGTALK_WEBHOOK。将secret和完整的Webhook地址填入你的.env文件。4.2 实现钉钉回调接口与签名验证钉钉为了安全要求对机器人收到的回调进行签名验证。我们需要在main.py中实现这个逻辑。# main.py import hashlib import hmac import base64 import urllib.parse import time import json from fastapi import FastAPI, Request, HTTPException, BackgroundTasks from fastapi.responses import JSONResponse import httpx from agent_core import get_agent, close_agent from config import DINGTALK_SECRET, DINGTALK_WEBHOOK app FastAPI(titleDingTalk AI Agent) def verify_dingtalk_signature(timestamp: str, sign: str, secret: str, body: bytes) - bool: 验证钉钉机器人回调签名 if not timestamp or not sign: return False # 钉钉要求如果时间戳与当前时间相差超过1小时则拒绝 if abs(int(time.time() * 1000) - int(timestamp)) 3600 * 1000: return False string_to_sign f{timestamp}\n{secret} hmac_code hmac.new( secret.encode(utf-8), string_to_sign.encode(utf-8), digestmodhashlib.sha256 ).digest() my_sign base64.b64encode(hmac_code).decode(utf-8) return hmac.compare_digest(my_sign, sign) async def send_dingtalk_message(text: str, at_all: bool False, at_user_ids: list None): 通过钉钉机器人Webhook发送消息 headers {Content-Type: application/json} payload { msgtype: text, text: { content: text }, at: { isAtAll: at_all, atUserIds: at_user_ids or [] } } async with httpx.AsyncClient() as client: try: # 注意钉钉要求对Webhook URL进行签名 timestamp str(int(time.time() * 1000)) secret DINGTALK_SECRET string_to_sign f{timestamp}\n{secret} hmac_code hmac.new(secret.encode(utf-8), string_to_sign.encode(utf-8), hashlib.sha256).digest() sign urllib.parse.quote_plus(base64.b64encode(hmac_code).decode(utf-8)) webhook_url f{DINGTALK_WEBHOOK}timestamp{timestamp}sign{sign} resp await client.post(webhook_url, jsonpayload, headersheaders) resp.raise_for_status() except Exception as e: print(f发送钉钉消息失败: {e}) async def process_agent_task(content: str, sender_id: str): 后台任务调用Agent处理消息并回复 try: agent await get_agent() # 这里简化处理将用户消息直接传给Agent # 在实际中你可能需要构建更复杂的对话历史管理 response await agent.run(content) # 将Agent的回复发送回钉钉并发送者 await send_dingtalk_message(f{response}\n\n—— 来自AI小助理, at_user_ids[sender_id]) except Exception as e: error_msg f处理您的请求时出现错误{str(e)} await send_dingtalk_message(error_msg, at_user_ids[sender_id]) app.post(/dingtalk/callback) async def dingtalk_callback(request: Request, background_tasks: BackgroundTasks): 钉钉机器人消息回调入口 # 1. 获取签名和参数 timestamp request.headers.get(timestamp) sign request.headers.get(sign) body_bytes await request.body() # 2. 验证签名 if not verify_dingtalk_signature(timestamp, sign, DINGTALK_SECRET, body_bytes): raise HTTPException(status_code403, detail签名验证失败) # 3. 解析消息体 try: body json.loads(body_bytes.decode(utf-8)) msg_type body.get(msgtype) text_content body.get(text, {}).get(content, ).strip() sender_id body.get(senderId) # 发送者钉钉ID conversation_id body.get(conversationId) # 4. 只处理文本消息并过滤掉机器人自己的消息防止循环 if msg_type ! text or [AI小助理] in text_content: return JSONResponse(content{msg: ignore}) # 5. 立即回复“正在处理”避免钉钉超时钉钉要求5秒内响应 # 这里先快速返回一个成功响应 response JSONResponse(content{msg: ok}) # 6. 将实际处理任务放入后台异步执行 background_tasks.add_task(process_agent_task, text_content, sender_id) return response except json.JSONDecodeError: raise HTTPException(status_code400, detail无效的JSON数据) app.on_event(shutdown) async def shutdown_event(): 应用关闭时清理资源 await close_agent() if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)这个main.py是服务的核心入口verify_dingtalk_signature严格按照钉钉文档实现签名验证这是安全的第一道关卡。send_dingtalk_message用于主动向钉钉群发送消息。注意发送时也需要对Webhook URL进行实时签名。process_agent_task真正的业务逻辑处理函数。它从后台任务中调用获取Agent实例运行用户消息并将结果发回钉钉。使用BackgroundTasks是关键因为Agent处理尤其是调用大模型和网络搜索可能需要几秒到十几秒远超钉钉回调要求的5秒超时限制。我们必须先快速返回一个成功响应再在后台异步处理。/dingtalk/callback钉钉消息推送的接收端点。它验证签名、解析消息然后触发后台任务。4.3 部署与启动服务现在我们可以启动服务了。在生产环境建议使用systemd或supervisor来管理进程。这里先以开发模式启动# 在项目根目录下 uvicorn main:app --host 0.0.0.0 --port 8000 --reload服务启动后你会看到类似Uvicorn running on http://0.0.0.0:8000的输出。由于钉钉需要回调你的公网IP你需要在路由器或云服务器安全组中开放8000端口并将http://你的公网IP:8000/dingtalk/callback这个地址配置到钉钉机器人的“消息接收地址”中在机器人设置页面。配置钉钉机器人回到之前创建机器人的设置页面找到“消息接收地址”。填入http://你的公网IP或域名:8000/dingtalk/callback。点击保存。钉钉会向这个地址发送一条验证请求如果你的服务签名验证通过并返回了正确的响应配置就成功了。5. 实战测试与效果优化一切就绪是时候在钉钉群里你的机器人试试了。5.1 基础功能测试在钉钉群里输入AI小助理 你好介绍一下你自己。稍等片刻你应该会收到回复内容大致是通义千问生成的自我介绍表明它是一个AI助手。再测试一下工具调用AI小助理 搜索一下今天科技圈有什么重要新闻。如果配置正确nanobot会指示通义千问调用web_search工具工具会调用Serper API进行搜索并将结果返回给通义千问进行总结最终你将收到一个包含几条新闻摘要和链接的回复。5.2 效果优化与高级技巧基础的跑通只是第一步要让这个“小助理”真正好用还需要不少优化。5.1 提升工具描述的准确性工具的description是模型决定是否调用的关键。模糊的描述会导致误调用或不调用。例如对于搜索工具我最初的描述是“搜索信息”结果模型经常在不需要实时信息时也调用它。后来我优化为“当用户询问需要实时数据、新闻、最新事件或你不知道的信息时使用此工具”准确率大幅提升。给你的工具写描述就像给新同事写岗位说明书越具体越好。5.2 实现对话历史管理目前的agent.run(content)是单轮对话Agent没有上下文记忆。在实际使用中用户可能会说“上一条新闻的详细内容呢”。我们需要让Agent记住之前的对话。nanobot支持在run方法中传入message_history。我们需要修改process_agent_task为每个用户或每个会话维护一个历史消息列表。一个简单的实现是使用字典在内存中缓存生产环境建议用Redis# 在 agent_core.py 或单独模块中 from collections import defaultdict from nanobot.schema import Message conversation_history defaultdict(list) # key: conversation_id, value: list of Messages async def process_agent_task(content: str, sender_id: str, conversation_id: str): agent await get_agent() history conversation_history.get(conversation_id, []) # 将用户新消息加入历史 history.append(Message(roleuser, contentcontent)) # 调用Agent传入历史 response await agent.run(history) # 将AI回复加入历史 history.append(Message(roleassistant, contentresponse)) # 限制历史长度防止token超限 if len(history) 10: # 保留最近5轮对话10条消息 history history[-10:] conversation_history[conversation_id] history await send_dingtalk_message(response, at_user_ids[sender_id])5.3 处理复杂任务与规划能力简单的问答和工具调用够了但用户可能想完成多步骤任务比如“帮我查一下杭州明天的天气如果是晴天就提醒我洗车如果是雨天就提醒我带伞”。这需要Agent具备规划能力。nanobot本身提供了基础的规划能力通过LLM的思维链Chain-of-Thought来实现。我们可以通过设计更精细的System Prompt系统指令来引导模型。在创建Agent时可以传入初始系统消息def create_agent(): llm QwenLLM(...) system_prompt 你是一个高效的AI助手集成在钉钉中。你的能力包括 1. 通用对话和问答。 2. 使用web_search工具获取实时信息。 3. 对于复杂任务请一步步思考并可以多次调用工具。 请用中文回复保持友好和专业。 initial_message Message(rolesystem, contentsystem_prompt) agent Nanobot(llmllm, initial_messages[initial_message]) agent.register_tool(web_search_tool) return agent一个强大的System Prompt能极大地塑造Agent的行为。你可以在这里定义它的角色、能力边界、回复格式等。5.4 增加更多实用工具一个搜索工具远远不够。我们可以根据团队需求开发更多工具内部知识库查询工具连接团队的Confluence、Wiki或数据库让AI能回答内部流程、项目信息。数据查询工具连接内部BI系统或数据库让AI能回答“上个月销售额是多少”这类问题。任务创建工具连接Jira、Teambition等实现“小助理 创建一个关于登录页面优化的任务指派给张三”。代码片段生成/解释工具针对技术团队可以集成代码模型的能力。每个工具的开发和注册模式都与web_search_tool类似关键是设计好输入输出和给模型的描述。5.5 错误处理与用户体验网络可能波动API可能超限模型可能抽风。良好的错误处理至关重要。在工具run方法中做好异常捕获返回友好的错误信息给模型而不是抛出异常导致整个流程中断。设置合理的超时时间在QwenLLM和工具中配置timeout避免一个请求卡住整个服务。提供降级方案如果搜索工具失败可以尝试让模型基于已有知识回答并提示“以下信息可能不是最新的”。在钉钉回复中加入状态提示对于耗时较长的任务可以在处理中、处理完成、处理失败时发送不同的状态消息。6. 部署、监控与成本考量6.1 生产环境部署开发模式的uvicorn不适合生产。建议使用Gunicorn with Uvicorn Workers提高并发能力。反向代理Nginx处理SSL、负载均衡和静态文件。进程管理systemd保证服务开机自启和自动重启。一个简单的systemd服务文件示例 (/etc/systemd/system/dingtalk-agent.service)[Unit] DescriptionDingTalk AI Agent Service Afternetwork.target [Service] Useryour_username Groupyour_groupname WorkingDirectory/home/your_username/projects/dingtalk-agent EnvironmentPATH/home/your_username/projects/dingtalk-agent/venv/bin ExecStart/home/your_username/projects/dingtalk-agent/venv/bin/gunicorn \ -w 4 \ -k uvicorn.workers.UvicornWorker \ --bind 0.0.0.0:8000 \ main:app Restartalways RestartSec10 [Install] WantedBymulti-user.target然后使用sudo systemctl start dingtalk-agent启动服务。6.2 日志与监控日志是排查问题的生命线。使用Python的logging模块在关键节点收到请求、调用模型、调用工具、发送回复记录信息。可以将日志输出到文件并使用logrotate管理。监控方面需要关注服务健康HTTP端点的可用性可用crontab定时curl。API调用情况通义千问API的调用次数、Token消耗、错误率。阿里云控制台有详细监控。工具调用延迟搜索、数据库查询等工具的响应时间。钉钉消息成功率消息发送失败的情况。6.3 成本分析与优化成本主要来自两块通义千问API费用按Token消耗计费。qwen-max比qwen-plus能力强但也更贵。可以根据任务复杂度选择模型。对于简单的分类、总结任务可以用便宜模型对于需要复杂推理的再用强模型。也可以在Agent逻辑里实现这种路由。第三方工具API费用如Serper搜索API、数据库查询可能产生的费用。优化建议缓存对常见、结果变化不频繁的查询如“公司请假流程是什么”进行缓存可以缓存模型回复也可以缓存工具结果。设置用量限额在代码层面或云平台设置每日/每月API调用上限防止意外消耗。精简上下文合理管理对话历史长度过长的历史会消耗大量Token。考虑私有化部署如果使用量非常大可以考虑在本地GPU服务器上部署通义千问的开源模型如Qwen2.5-7B/14B一次性硬件投入后调用成本几乎为零。7. 避坑指南我踩过的那些“坑”这个项目从构思到稳定运行我遇到了不少问题这里总结几个典型的希望能帮你绕过去。7.1 钉钉签名验证失败这是第一个拦路虎。现象是钉钉后台配置回调地址时一直报“验证失败”。坑1时间戳问题。钉钉要求服务器时间与网络时间同步。我最初服务器时间慢了几分钟导致签名验证永远失败。务必使用ntpdate或chronyd同步时间。坑2签名计算错误。钉钉的签名字符串是timestamp \n secret这个\n是换行符必须包含。我一开始用timestamp secret拼接怎么都对不上。仔细对照官方文档的示例代码才发现这个细节。坑3URL编码。计算出的签名sign需要经过URL编码后拼接到Webhook URL上。urllib.parse.quote_plus用在这里正合适它能正确处理加号等特殊字符。7.2 通义千问API调用返回空或错误坑模型名称错误或区域不对。通义千问在不同区域杭州、上海等的端点可能略有不同模型名如qwen-max,qwen-plus也要确保正确。最好先在阿里云控制台的“模型体验”或通过简单的curl命令测试API密钥和模型是否可用。坑Token超限。通义千问API有每分钟/每天的调用频率限制和Token数量限制。如果突然大量调用会被限流。需要在代码中加入简单的限流和重试机制并监控调用量。7.3 Agent“乱用”或“不用”工具坑工具描述太笼统。如前所述这是最常见的问题。把工具描述当成给一个“死板但聪明”的新员工写操作手册场景、输入、输出格式都要说清楚。坑System Prompt冲突。如果你的System Prompt里说“你是一个聊天机器人不要进行网络操作”那模型很可能就不会调用搜索工具。System Prompt和工具描述要协同工作明确告诉模型“在需要时你可以使用以下工具”。坑模型温度temperature过高。温度参数控制输出的随机性。太高如0.9可能导致模型行为不稳定有时调用工具有时不调用。对于任务型Agent建议设置较低的温度如0.1或0.2使其输出更确定、更可预测。7.4 异步处理与超时坑同步处理导致钉钉超时。这是架构上的关键点。钉钉回调要求5秒内响应而LLM调用工具调用很容易超过5秒。必须使用BackgroundTasks或消息队列如Celery进行异步处理主线程立即返回“成功接收”。坑后台任务异常导致用户无反馈。如果后台任务process_agent_task崩溃了用户就再也收不到回复。一定要在这个函数内部做好全面的异常捕获至少记录日志并考虑在失败时通过钉钉给用户发送一个错误提示。7.5 安全性问题坑暴露内部信息。Agent能调用搜索工具如果用户诱导它搜索“公司内网地址”、“内部系统密码”怎么办虽然工具本身可能搜不到但需要在System Prompt里明确禁止并考虑对用户输入进行基础的关键词过滤。更重要的不要给Agent连接具有高权限的内部系统工具。遵循最小权限原则为Agent创建专用的、只有只读权限或受限权限的账户。坑Webhook端点被恶意调用。虽然有了签名验证但也要防止重放攻击签名被截获后重复使用。我们的代码已经通过校验时间戳1小时窗口来缓解这一问题。在生产环境还可以考虑增加IP白名单如果钉钉出口IP固定或额外的自定义Token验证。经过以上这些步骤一个能够“随时派活”的钉钉AI智能体就真正搭建完成了。它不再是一个玩具而是一个能融入团队日常工作流、切实提升效率的生产力工具。你可以根据自己团队的需求不断为它添加新的“技能”让它变得越来越能干。
