1. 項目緣起當(dāng)“派活”遇上AI一個真實的生產(chǎn)力痛點在團(tuán)隊協(xié)作里最讓人頭疼的往往不是技術(shù)難題本身而是那些瑣碎、重復(fù)但又必須有人去處理的“小活”。比如每天要手動匯總幾個渠道的數(shù)據(jù)生成日報或者根據(jù)客戶反饋的關(guān)鍵詞去內(nèi)部知識庫搜一圈資料再或者新同事入職時需要有人一步步告訴他項目環(huán)境怎么搭、代碼怎么拉。這些任務(wù)本身不復(fù)雜但極其消耗人的注意力和時間打斷深度工作流。過去我們可能會寫個腳本或者依賴某個固定的流程文檔但腳本需要觸發(fā)文檔需要人去看都不夠“絲滑”。直到AI Agent的概念火起來尤其是像nanobot這樣輕量、可編程的框架出現(xiàn)事情開始變得有趣。我們能不能做一個“智能小工”把它放到釘釘這種大家每天必用的協(xié)作工具里讓它7x24小時待命用自然語言就能給它派活它還能調(diào)用各種能力把活干完甚至把結(jié)果主動推回來這個想法就是我們這個“國產(chǎn)小龍蝦方案”的起點——像吃小龍蝦一樣把那些瑣碎的“殼”重復(fù)勞動剝掉只享受核心的“肉”價值創(chuàng)造。核心三件套nanobot作為機(jī)器人的“大腦”和“手腳”通義千問Qwen提供強(qiáng)大的自然語言理解與生成能力作為“思維”釘釘機(jī)器人則是它融入我們工作流的“身體”和“接口”。2. 方案核心組件選型與架構(gòu)解析為什么是這三個組合這背后是一套經(jīng)過權(quán)衡的“性價比”架構(gòu)。2.1 nanobot輕量級AI Agent的“骨架”nanobot不是一個大眾熟知的名字但在特定的開發(fā)者圈子里它正成為快速構(gòu)建AI Agent的熱門選擇。你可以把它理解為一個高度模塊化、事件驅(qū)動的機(jī)器人框架。它核心解決了兩個問題“如何響應(yīng)”和“如何執(zhí)行”。事件驅(qū)動模型nanobot的核心是監(jiān)聽各種事件比如釘釘?shù)南ⅰTTP接口調(diào)用、定時任務(wù)觸發(fā)然后根據(jù)預(yù)定義的邏輯進(jìn)行處理。這非常適合IM機(jī)器人場景你的消息就是觸發(fā)它的事件。技能Skill插件化它的能力不是固化的而是通過加載不同的“技能”插件來擴(kuò)展。一個技能就是一個獨立的Python模塊負(fù)責(zé)處理一類具體的任務(wù)比如“查詢天氣”、“執(zhí)行Shell命令”、“調(diào)用某個API”。這種設(shè)計讓功能的增刪改變得非常清晰和簡單。與LLM的松耦合nanobot本身不綁定任何特定的大語言模型LLM。它通過一個標(biāo)準(zhǔn)的接口與LLM對話你可以在配置文件中輕松切換不同的LLM服務(wù)提供商比如今天用通義千問明天想試試GPT改個配置就行代碼幾乎不用動。選型理由相比于一些重型的、面向復(fù)雜規(guī)劃的Agent框架如LangChain、AutoGennanobot更輕、更直接學(xué)習(xí)曲線平緩對于“接收指令-執(zhí)行任務(wù)-返回結(jié)果”這類明確的工作流它的開銷和復(fù)雜度都更低更像一個“機(jī)器人操作系統(tǒng)”。2.2 通義千問Qwen本土化強(qiáng)大的“思維引擎”通義千問是阿里云推出的開源大語言模型系列。選擇它而非常見的OpenAI接口主要基于以下幾點考慮合規(guī)與可控性所有數(shù)據(jù)在國內(nèi)處理無需擔(dān)憂跨境數(shù)據(jù)傳輸?shù)恼吲c延遲風(fēng)險這對于企業(yè)級應(yīng)用是首要考量。成本與性能Qwen系列提供了從7B到72B不同規(guī)模的模型并且有專門優(yōu)化的API服務(wù)。對于機(jī)器人場景通常不需要動用千億參數(shù)模型一個效果不錯的7B或14B模型通過其API調(diào)用在響應(yīng)速度和成本間能取得很好的平衡。實測下來Qwen在中文指令理解、上下文關(guān)聯(lián)和代碼生成方面表現(xiàn)相當(dāng)可靠。生態(tài)集成作為阿里系產(chǎn)品與釘釘、阿里云函數(shù)計算等服務(wù)的集成理論上會更順暢文檔和案例也更豐富。在我們的方案里Qwen扮演著“意圖理解”和“內(nèi)容生成”的核心角色。nanobot將用戶模糊的自然語言指令如“幫我查一下上周項目A的日志錯誤”傳遞給QwenQwen需要解析出用戶的真實意圖“查詢?nèi)罩尽辈⑻崛￡P(guān)鍵參數(shù)“項目A”、“上周”、“錯誤”。有時它還需要根據(jù)對話歷史進(jìn)行多輪澄清。2.3 釘釘機(jī)器人無縫嵌入工作流的“交互界面”釘釘機(jī)器人是釘釘開放平臺提供的能力它允許開發(fā)者創(chuàng)建一個自定義的機(jī)器人并將其添加到群聊或單聊中。它成為了我們AI Agent最自然的入口。開箱即用的通道無需自己搭建WebSocket或長輪詢釘釘負(fù)責(zé)消息的接收和推送我們只需要提供一個HTTPS的回調(diào)地址Webhook供釘釘在收到消息時調(diào)用。豐富的消息格式支持文本、鏈接、Markdown、ActionCard交互卡片等多種消息格式這讓我們的機(jī)器人回復(fù)可以非常美觀和交互性強(qiáng)。例如返回一個數(shù)據(jù)匯總時可以用Markdown表格提供一個操作選擇時可以用帶按鈕的卡片。安全的鑒權(quán)機(jī)制支持簽名驗證確保發(fā)送到我們服務(wù)端的請求確實來自釘釘防止惡意調(diào)用。架構(gòu)全景圖用戶機(jī)器人并發(fā)送指令 - 釘釘服務(wù)器將消息封裝成HTTP POST請求發(fā)送到我們部署的nanobot服務(wù) - nanobot接收到事件觸發(fā)對應(yīng)的消息處理流程 - nanobot調(diào)用配置的Qwen API進(jìn)行意圖識別和參數(shù)提取 - 根據(jù)識別結(jié)果nanobot調(diào)用對應(yīng)的技能Skill來執(zhí)行具體任務(wù)如查詢數(shù)據(jù)庫、調(diào)用外部API、執(zhí)行腳本- 技能執(zhí)行完畢將結(jié)果返回給nanobot - nanobot將結(jié)果格式化為釘釘消息通過釘釘提供的API發(fā)送回對應(yīng)的群聊或會話。整個流程形成了一個閉環(huán)。3. 從零到一環(huán)境搭建與核心配置實戰(zhàn)理論講完我們動手搭一個。假設(shè)你已經(jīng)有一個Linux服務(wù)器或本地開發(fā)環(huán)境并且擁有釘釘開發(fā)者賬號和阿里云API密鑰。3.1 基礎(chǔ)環(huán)境準(zhǔn)備首先為我們的“小龍蝦”準(zhǔn)備一個干凈的“廚房”。# 1. 創(chuàng)建并進(jìn)入項目目錄 mkdir nanobot_dingtalk_agent cd nanobot_dingtalk_agent # 2. 創(chuàng)建虛擬環(huán)境強(qiáng)烈推薦避免依賴沖突 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 3. 安裝nanobot核心包 pip install nanobot-core # 根據(jù)你需要安裝額外的社區(qū)技能包例如一個用于HTTP請求的通用技能 pip install nanobot-skill-http # 4. 安裝通義千問的SDK # 這里假設(shè)使用DashScope阿里云靈積的API它是調(diào)用通義千問的官方途徑 pip install dashscope3.2 釘釘機(jī)器人創(chuàng)建與配置這是連接現(xiàn)實世界的第一步。登錄釘釘開發(fā)者后臺找到你的組織進(jìn)入“應(yīng)用開發(fā)” - “企業(yè)內(nèi)部開發(fā)” - “機(jī)器人”。創(chuàng)建機(jī)器人點擊“創(chuàng)建應(yīng)用”選擇“機(jī)器人”填寫名稱和描述。創(chuàng)建成功后記錄下AppKey和AppSecret這是機(jī)器人身份的憑證。配置消息接收在機(jī)器人詳情頁找到“消息接收”配置。這里需要填寫一個公網(wǎng)可訪問的URL即我們即將部署的nanobot服務(wù)的Webhook地址。例如https://your-server.com/dingtalk/callback。在開發(fā)階段你可以使用內(nèi)網(wǎng)穿透工具如ngrok、frp將本地服務(wù)暴露為一個臨時公網(wǎng)地址進(jìn)行測試。獲取Webhook地址在“機(jī)器人詳情”頁你還能找到一個“Webhook”地址格式如https://oapi.dingtalk.com/robot/send?access_tokenXXX。這個地址用于主動發(fā)送消息到釘釘。我們nanobot在完成任務(wù)后會調(diào)用這個地址把結(jié)果推回去。請妥善保管這個access_token。發(fā)布與安裝將機(jī)器人發(fā)布到你的組織并把它添加到需要使用的釘釘群中。3.3 nanobot項目初始化與核心邏輯編寫現(xiàn)在我們來編寫機(jī)器人的“大腦”和“反射弧”。# 在項目根目錄初始化一個nanobot項目結(jié)構(gòu) nanobot init my_dingtalk_bot cd my_dingtalk_bot你會看到一個基礎(chǔ)的項目結(jié)構(gòu)包含configs,skills,main.py等。我們主要關(guān)注三個文件1. 配置文件 (configs/config.yaml): 機(jī)器人的“基因”name: 小龍蝦助手 description: 一個幫你處理瑣事的AI小工 # LLM配置連接通義千問 llm: provider: dashscope # 指定提供商 model: qwen-plus # 使用qwen-plus模型可根據(jù)需要換為 qwen-turbo, qwen-max等 api_key: ${DASHSCOPE_API_KEY} # 從環(huán)境變量讀取API Key更安全 parameters: temperature: 0.1 # 降低隨機(jī)性讓回答更穩(wěn)定 top_p: 0.8 # 技能配置加載我們自定義的技能 skills: - skills.my_custom_skill # 一個自定義的示例技能 - nanobot_skill_http # 引入的第三方HTTP技能 # 事件監(jiān)聽器配置處理釘釘消息 listeners: - name: dingtalk_webhook type: webhook # 監(jiān)聽Webhook請求 path: /dingtalk/callback # 對應(yīng)釘釘后臺配置的路徑 method: POST2. 自定義技能 (skills/my_custom_skill.py): 機(jī)器人的“手藝”技能是能力的載體。這里我們創(chuàng)建一個簡單的技能它能夠理解用戶關(guān)于“時間”的詢問。import logging from datetime import datetime from nanobot.skills import Skill, skill from nanobot.messages import TextMessage logger logging.getLogger(__name__) skill(nametime_query, description查詢當(dāng)前時間或日期) class TimeQuerySkill(Skill): 一個簡單的示例技能當(dāng)用戶詢問時間時返回當(dāng)前時間。 async def execute(self, context, **kwargs): 技能執(zhí)行入口。 context: 包含當(dāng)前會話、用戶消息等上下文信息。 user_message context.current_message.text # 這里可以做得更智能比如用LLM判斷意圖。 # 但作為示例我們簡單匹配關(guān)鍵詞。 if 時間 in user_message or 幾點 in user_message: now datetime.now().strftime(%Y-%m-%d %H:%M:%S) reply_text f現(xiàn)在是北京時間{now} # 返回一個文本消息對象nanobot會將其轉(zhuǎn)換為釘釘消息格式 return TextMessage(contentreply_text) # 如果不匹配返回Nonenanobot會嘗試其他技能或使用LLM直接回復(fù) return None3. 主程序適配釘釘 (main.py): 消息的“翻譯官”nanobot默認(rèn)的Webhook監(jiān)聽器可能不直接兼容釘釘?shù)南⒏袷健ａ斸敯l(fā)送過來的是一套特定的JSON結(jié)構(gòu)。我們需要一個適配器來“翻譯”。import hashlib import hmac import base64 import time import json from urllib.parse import quote_plus from nanobot import Nanobot from nanobot.listeners import WebhookListener from fastapi import FastAPI, Request, Header, HTTPException from fastapi.responses import JSONResponse # 從環(huán)境變量或配置中讀取釘釘?shù)暮灻荑€ import os DINGTALK_SECRET os.getenv(DINGTALK_SECRET, ) app FastAPI() bot Nanobot.from_config() # 自定義一個釘釘消息解析函數(shù) def parse_dingtalk_message(request_data: dict): 將釘釘?shù)腤ebhook JSON格式轉(zhuǎn)換為nanobot能理解的內(nèi)部消息格式。 msg_type request_data.get(msgtype, text) if msg_type text: content request_data.get(text, {}).get(content, ).strip() # 移除可能存在的機(jī)器人名字 # 釘釘消息格式可能是“小龍蝦助手 現(xiàn)在幾點” # 這里簡單處理實際可能需要更精確的解析 if content.startswith(): # 簡單移除第一個和其后的空格或直到下一個空格 parts content.split( , 1) if len(parts) 1: content parts[1] else: content sender_id request_data.get(senderStaffId, ) # 發(fā)送者ID conversation_id request_data.get(conversationId, ) # 會話ID # 構(gòu)造nanobot需要的消息字典 nanobot_msg { text: content, user_id: sender_id, conversation_id: conversation_id, platform: dingtalk, raw_data: request_data # 保留原始數(shù)據(jù)以備后用 } return nanobot_msg # 可以繼續(xù)處理其他消息類型如圖片、鏈接等 return None # 釘釘簽名驗證函數(shù)重要用于安全驗證 def verify_dingtalk_signature(timestamp: str, sign: str, secret: str, body: str): 驗證釘釘回調(diào)請求的簽名。 參考釘釘官方文檔https://open.dingtalk.com/document/robots/customize-robot-security-settings if not secret: return True # 如果未設(shè)置SECRET跳過驗證不推薦生產(chǎn)環(huán)境 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) app.post(/dingtalk/callback) async def dingtalk_callback(request: Request, timestamp: str Header(None, aliastimestamp), sign: str Header(None, aliassign)): 釘釘Webhook回調(diào)入口。 try: body_bytes await request.body() body_str body_bytes.decode(utf-8) request_data json.loads(body_str) # 1. 簽名驗證 if not verify_dingtalk_signature(timestamp, sign, DINGTALK_SECRET, body_str): raise HTTPException(status_code403, detailInvalid signature) # 2. 解析釘釘消息為nanobot格式 nanobot_msg parse_dingtalk_message(request_data) if not nanobot_msg: return JSONResponse(content{msg: Unsupported message type}) # 3. 將消息交給nanobot核心處理 # 這里我們同步調(diào)用對于簡單場景夠用。復(fù)雜場景可考慮異步隊列。 responses await bot.process_message(nanobot_msg) # 4. 將nanobot的回復(fù)消息通過釘釘?shù)腤ebhook發(fā)送回去 # 注意這里需要你機(jī)器人的access_token dingtalk_webhook_url fhttps://oapi.dingtalk.com/robot/send?access_tokenYOUR_ACCESS_TOKEN for resp in responses: if hasattr(resp, to_dingtalk): # 如果消息對象有轉(zhuǎn)換為釘釘格式的方法 dingtalk_msg resp.to_dingtalk() else: # 默認(rèn)處理文本消息 dingtalk_msg { msgtype: text, text: {content: resp.content} } # 使用requests或aiohttp發(fā)送消息到釘釘此處省略發(fā)送代碼 # await send_to_dingtalk(dingtalk_webhook_url, dingtalk_msg) print(f[待發(fā)送] 釘釘消息: {dingtalk_msg}) # 開發(fā)階段先打印 return JSONResponse(content{msg: ok}) except json.JSONDecodeError: raise HTTPException(status_code400, detailInvalid JSON) except Exception as e: logging.error(f處理釘釘回調(diào)失敗: {e}) raise HTTPException(status_code500, detailInternal server error) if __name__ __main__: import uvicorn # 啟動FastAPI服務(wù)監(jiān)聽在8080端口 uvicorn.run(app, host0.0.0.0, port8080)這個main.py做了幾件關(guān)鍵事驗證釘釘請求的真實性、將釘釘格式的消息“翻譯”成nanobot能懂的結(jié)構(gòu)、將nanobot處理后的結(jié)果再“翻譯”回釘釘格式并發(fā)送。你需要將YOUR_ACCESS_TOKEN替換為你的機(jī)器人Webhook地址中的token并實現(xiàn)send_to_dingtalk函數(shù)可以使用aiohttp庫。3.4 連接通義千問讓機(jī)器人“聽懂人話”上面的自定義技能只是簡單匹配關(guān)鍵詞。要真正理解復(fù)雜、模糊的指令必須引入Qwen。我們需要修改技能讓它學(xué)會“問”Qwen。首先確保在config.yaml中正確配置了llm部分并且環(huán)境變量DASHSCOPE_API_KEY已設(shè)置。然后我們創(chuàng)建一個更高級的技能例如一個“智能問答”技能它會把所有無法被其他技能處理的指令都交給Qwen來處理并可以聯(lián)網(wǎng)搜索。# skills/smart_qa_skill.py import logging from nanobot.skills import Skill, skill from nanobot.messages import TextMessage from nanobot.llm import LLMClient # 引入LLM客戶端 logger logging.getLogger(__name__) skill(namesmart_qa, description通用智能問答與指令理解, priority50) # priority較低讓具體技能先匹配 class SmartQASkill(Skill): 一個利用Qwen進(jìn)行通用問答和意圖理解的技能。 它可以作為兜底技能處理其他技能無法處理的模糊指令。 def __init__(self, config): super().__init__(config) # 從nanobot的配置中獲取LLM客戶端實例 self.llm_client LLMClient.from_config(config) async def execute(self, context, **kwargs): user_message context.current_message.text if not user_message: return None # 構(gòu)建一個系統(tǒng)提示詞引導(dǎo)Qwen更好地理解機(jī)器人場景 system_prompt 你是一個部署在釘釘上的AI助手名叫“小龍蝦助手”。你的主要職責(zé)是幫助用戶處理工作瑣事。 用戶可能會給你一些模糊的指令你需要嘗試?yán)斫馄浔澈蟮囊鈭D。 常見的意圖包括但不限于查詢信息天氣、時間、股票、公司內(nèi)部數(shù)據(jù)、執(zhí)行簡單計算、翻譯、總結(jié)文本、生成代碼片段、回答問題等。 如果你判斷用戶的指令需要調(diào)用某個具體工具或技能比如查天氣需要調(diào)用API你可以回復(fù)一個結(jié)構(gòu)化的指令例如 [ACTION:weather_query location北京]。 否則請直接給出友好、有幫助的文本回答。 請用中文回復(fù)。 # 準(zhǔn)備對話歷史如果有的話 messages [ {role: system, content: system_prompt}, {role: user, content: user_message} ] try: # 調(diào)用通義千問API response await self.llm_client.chat_completion( messagesmessages, temperature0.2, # 較低的隨機(jī)性保證回答穩(wěn)定 max_tokens1024 ) ai_reply response.choices[0].message.content.strip() # 這里可以添加后處理解析AI回復(fù)中是否包含結(jié)構(gòu)化指令 [ACTION:...] # 如果包含可以觸發(fā)另一個對應(yīng)的技能如weather_query # 這是一個簡單的實現(xiàn)示例 if ai_reply.startswith([ACTION:) and ai_reply.endswith(]): # 解析動作和參數(shù)這里簡化處理 action_str ai_reply[8:-1] # 去掉[ACTION:和] # 理論上這里應(yīng)該調(diào)度到其他技能本例中我們只做演示 logger.info(f識別到結(jié)構(gòu)化指令: {action_str}) return TextMessage(contentf“我已理解您想執(zhí)行 {action_str}相關(guān)功能正在開發(fā)中。”) else: # 直接返回AI的文本回答 return TextMessage(contentai_reply) except Exception as e: logger.error(f調(diào)用Qwen API失敗: {e}) return TextMessage(content抱歉我的大腦Qwen服務(wù)暫時開小差了請稍后再試。)別忘了在config.yaml的skills列表里加上這個新技能- “skills.smart_qa_skill”。現(xiàn)在你的機(jī)器人就接上了通義千問的“大腦”能夠理解更復(fù)雜的指令并進(jìn)行智能對話了。4. 部署、調(diào)試與進(jìn)階實戰(zhàn)技巧讓代碼跑起來只是第一步讓它穩(wěn)定、好用才是關(guān)鍵。4.1 本地調(diào)試與內(nèi)網(wǎng)穿透在開發(fā)階段你的服務(wù)運行在本地localhost釘釘?shù)姆?wù)器無法直接訪問。你需要一個“橋梁”。使用 ngrok (最簡便)# 下載ngrok并注冊獲取authtoken ngrok config add-authtoken YOUR_AUTH_TOKEN # 將本地的8080端口暴露到公網(wǎng) ngrok http 8080運行后ngrok會給你一個隨機(jī)的https://xxx.ngrok.io地址。將這個地址后面加上/dingtalk/callback填到釘釘機(jī)器人的“消息接收”URL中。使用 frp (更穩(wěn)定可控)如果你有自己的云服務(wù)器可以搭建frp服務(wù)獲得一個固定的二級域名更適合長期測試。調(diào)試技巧在main.py中多使用logging打印關(guān)鍵信息如接收到的原始數(shù)據(jù)、解析后的消息、調(diào)用Qwen的請求和響應(yīng)。釘釘機(jī)器人平臺也有“消息推送記錄”可以查看發(fā)送是否成功。4.2 服務(wù)器部署與進(jìn)程守護(hù)開發(fā)完成后需要部署到正式的服務(wù)器。使用 systemd (Linux)創(chuàng)建一個服務(wù)文件/etc/systemd/system/nanobot-dingtalk.service。[Unit] DescriptionNanobot DingTalk AI Agent Afternetwork.target [Service] Typesimple Userwww-data # 根據(jù)你的情況修改 WorkingDirectory/path/to/your/nanobot_dingtalk_agent EnvironmentPATH/path/to/your/venv/bin EnvironmentDASHSCOPE_API_KEYyour_key_here EnvironmentDINGTALK_SECRETyour_secret_here ExecStart/path/to/your/venv/bin/python main.py Restartalways RestartSec10 [Install] WantedBymulti-user.target然后啟用并啟動服務(wù)sudo systemctl daemon-reload sudo systemctl enable nanobot-dingtalk sudo systemctl start nanobot-dingtalk sudo systemctl status nanobot-dingtalk # 查看狀態(tài)使用 Docker (推薦便于隔離和遷移)編寫Dockerfile和docker-compose.yml將應(yīng)用、Python環(huán)境、配置文件打包。這樣可以在任何有Docker的環(huán)境一鍵啟動。4.3 設(shè)計更強(qiáng)大的技能以“數(shù)據(jù)查詢”為例一個真正的“派活”機(jī)器人必須能操作真實的數(shù)據(jù)或系統(tǒng)。我們設(shè)計一個技能讓機(jī)器人能查詢數(shù)據(jù)庫以MySQL為例并返回結(jié)果。# skills/data_query_skill.py import aiomysql import logging from nanobot.skills import Skill, skill from nanobot.messages import TextMessage, MarkdownMessage logger logging.getLogger(__name__) skill(namedata_query, description查詢指定數(shù)據(jù)庫表的數(shù)據(jù), patterns[“查一下”, “數(shù)據(jù)”, “報表”]) # 可以加一些觸發(fā)關(guān)鍵詞 class DataQuerySkill(Skill): def __init__(self, config): super().__init__(config) self.db_config { host: config.get(db_host, localhost), port: config.get(db_port, 3306), user: config.get(db_user, root), password: config.get(db_password, ), db: config.get(db_name, test), charset: utf8mb4 } self.pool None async def connect_db(self): 創(chuàng)建數(shù)據(jù)庫連接池 if self.pool is None: self.pool await aiomysql.create_pool(**self.db_config) return self.pool async def execute(self, context, **kwargs): user_message context.current_message.text # 這里應(yīng)該集成LLM進(jìn)行意圖識別和SQL生成。 # 為了安全我們不做全自然語言轉(zhuǎn)SQL而是預(yù)定義幾個查詢模板。 # 例如用戶說“查一下上周的訂單總數(shù)” # 我們可以用LLM提取出“上周”和“訂單總數(shù)”然后映射到預(yù)定義的SQL模板。 # 假設(shè)我們通過LLM或簡單規(guī)則已經(jīng)確定了查詢參數(shù) query_type “l(fā)ast_week_order_count” # 這個應(yīng)由更高級的意圖識別模塊提供 params {} if query_type “l(fā)ast_week_order_count”: sql “SELECT COUNT(*) as total_orders FROM orders WHERE order_date DATE_SUB(CURDATE(), INTERVAL 7 DAY)” result_key “total_orders” result_desc “上周訂單總數(shù)” else: return None # 不處理其他類型 try: pool await self.connect_db() async with pool.acquire() as conn: async with conn.cursor(aiomysql.DictCursor) as cur: await cur.execute(sql, params) result await cur.fetchone() if result: # 將結(jié)果格式化為Markdown表格或文本 reply_md f“**{result_desc}**\n\n” reply_md f“{result_key}: **{result[result_key]}**\n” # 可以構(gòu)造更復(fù)雜的表格 # reply_md “| 指標(biāo) | 值 |\n| :--- | :--- |\n” # for k, v in result.items(): # reply_md f“| {k} | {v} |\n” return MarkdownMessage(contentreply_md) else: return TextMessage(content“未查詢到相關(guān)數(shù)據(jù)。”) except Exception as e: logger.exception(f“數(shù)據(jù)庫查詢失敗: {e}”) return TextMessage(content“查詢數(shù)據(jù)時出現(xiàn)錯誤請檢查數(shù)據(jù)庫連接或查詢語句。”) async def cleanup(self): 技能卸載時關(guān)閉連接池 if self.pool: self.pool.close() await self.pool.wait_closed()這個技能展示了如何連接外部資源。關(guān)鍵點在于永遠(yuǎn)不要將用戶輸入直接拼接成SQLSQL注入風(fēng)險應(yīng)該通過LLM或固定模板將自然語言轉(zhuǎn)換為安全的查詢參數(shù)。4.4 安全與權(quán)限管控把機(jī)器人放進(jìn)工作群安全是重中之重。請求簽名驗證如前文代碼所示必須驗證釘釘回調(diào)的簽名(timestampsign)這是防止偽造請求的第一道防線。技能執(zhí)行權(quán)限不是所有用戶都能調(diào)用所有技能。可以在Skill的execute方法開頭檢查context.current_message.user_id對照一個權(quán)限列表可以配置在文件或數(shù)據(jù)庫里。例如只有管理員才能執(zhí)行“重啟服務(wù)”這樣的高危技能。敏感信息脫敏從數(shù)據(jù)庫或API查詢到的結(jié)果在發(fā)送到釘釘前要過濾掉手機(jī)號、身份證號、密鑰等敏感信息。限流與防刷對用戶的請求頻率做限制防止惡意調(diào)用消耗你的LLM API額度或數(shù)據(jù)庫資源。可以在main.py的入口處添加簡單的計數(shù)器或使用Redis實現(xiàn)更復(fù)雜的限流。網(wǎng)絡(luò)隔離部署機(jī)器人的服務(wù)器應(yīng)處于內(nèi)網(wǎng)或具有嚴(yán)格安全組策略的網(wǎng)絡(luò)中僅開放必要的端口如80/443給釘釘回調(diào)。4.5 監(jiān)控與日志一個健壯的服務(wù)離不開監(jiān)控。結(jié)構(gòu)化日志使用Python的logging模塊配置JSON格式輸出方便接入ELKElasticsearch, Logstash, Kibana或類似日志系統(tǒng)。記錄每一次請求、LLM調(diào)用、技能執(zhí)行結(jié)果和耗時。健康檢查為你的FastAPI服務(wù)添加一個/health端點返回服務(wù)狀態(tài)、數(shù)據(jù)庫連接狀態(tài)等。方便運維監(jiān)控。關(guān)鍵指標(biāo)統(tǒng)計技能調(diào)用次數(shù)、Qwen API調(diào)用耗時與Token消耗、錯誤率等。這些數(shù)據(jù)可以幫助你優(yōu)化技能設(shè)計和成本控制。5. 避坑指南與效能提升心法在實際開發(fā)和運維中我踩過不少坑也總結(jié)了一些讓“小龍蝦”更好用的經(jīng)驗。5.1 意圖識別的“最后一公里”難題最初我試圖讓Qwen直接解析所有指令并返回可執(zhí)行的命令。但發(fā)現(xiàn)對于邊界模糊的指令效果不穩(wěn)定。比如“看看張三上周的業(yè)績”Qwen可能理解成“查詢數(shù)據(jù)庫”也可能直接生成一段文本描述。我的解決方案是分層處理第一層精確技能匹配。先定義一批高頻、明確的技能并用關(guān)鍵詞或正則表達(dá)式觸發(fā)。如“/天氣 北京”、“打卡”。第二層LLM意圖分類。對于未匹配的指令交給Qwen但任務(wù)不是直接執(zhí)行而是分類。我訓(xùn)練或通過提示詞引導(dǎo)Qwen將指令分類到預(yù)定義的幾個“技能槽”如[QUERY_DATA]、[CALCULATE]、[GENERATE_TEXT]等并提取結(jié)構(gòu)化參數(shù)。第三層技能執(zhí)行與LLM潤色。根據(jù)分類結(jié)果調(diào)用對應(yīng)的技能獲取原始數(shù)據(jù)或結(jié)果再將這個結(jié)果和原始指令一起交給Qwen讓它生成一段通順、友好的最終回復(fù)。這樣既保證了關(guān)鍵動作的準(zhǔn)確執(zhí)行又利用了LLM的泛化理解和語言生成能力。5.2 釘釘消息格式的“坑”釘釘?shù)南⒏袷奖容^獨特稍不注意就會發(fā)送失敗或顯示異常。Markdown表格對齊釘釘?shù)腗arkdown對表格支持有限復(fù)雜的表格可能渲染錯亂。建議簡單表格用|復(fù)雜數(shù)據(jù)可以考慮用“字段值”的列表形式或者直接生成圖片通過技能調(diào)用圖表生成API發(fā)送。特定用戶在機(jī)器人回復(fù)中想某人需要在文本中使用用戶手機(jī)號并且同時傳遞at字段。但獲取群成員的手機(jī)號需要額外的釘釘API權(quán)限需申請。通常我們只在回復(fù)中觸發(fā)指令的用戶context.current_message.sender_id對應(yīng)的手機(jī)號。消息長度限制釘釘單條文本消息有長度限制約5000字符。如果回復(fù)內(nèi)容很長需要主動切割成多條發(fā)送或者先生成一個文件如txt然后發(fā)送文件消息。ActionCard按鈕回調(diào)如果你使用了交互卡片按鈕點擊后的回調(diào)地址也必須是公網(wǎng)可訪問的且同樣需要處理簽名驗證。這相當(dāng)于為你的機(jī)器人增加了“事件”處理能力可以實現(xiàn)更復(fù)雜的交互流程。5.3 成本控制與性能優(yōu)化Qwen API是按Token收費的無節(jié)制地使用會讓賬單飛漲。設(shè)計精簡的提示詞Prompt系統(tǒng)提示詞不要過于冗長。在滿足指令清晰的前提下盡量簡短。將固定的上下文如公司背景、常用指令說明做得精煉。緩存LLM響應(yīng)對于一些常見、結(jié)果相對固定的查詢?nèi)纭肮竞喗椤薄ⅰ爱a(chǎn)品列表”可以將Qwen的回復(fù)緩存起來用Redis或內(nèi)存緩存設(shè)定一個合理的過期時間。下次遇到相同或相似問題時直接返回緩存結(jié)果大幅節(jié)省Token和延遲。設(shè)置使用限額為每個用戶或每個群設(shè)置每日/每周的LLM調(diào)用次數(shù)上限。可以在技能執(zhí)行前檢查計數(shù)。異步與非阻塞在main.py中處理釘釘回調(diào)并調(diào)用LLM和技能的過程如果耗時較長5秒釘釘可能會超時重試。建議將核心處理邏輯放入異步任務(wù)隊列如Celery Redis或直接用asyncio.create_task在收到回調(diào)后立即返回“ok”然后異步處理任務(wù)處理完再主動推送結(jié)果。這需要更復(fù)雜的架構(gòu)但能提供更好的用戶體驗和可靠性。5.4 從“玩具”到“工具”的演進(jìn)一個能回答問題的機(jī)器人和一個能真正“派活”的Agent差距在于狀態(tài)管理和工作流。會話狀態(tài)nanobot本身支持簡單的會話上下文。但對于一個需要多步交互的復(fù)雜任務(wù)比如“幫我訂會議室要下午2點10個人的”你需要自己維護(hù)一個更強(qiáng)大的狀態(tài)機(jī)。可以將對話狀態(tài)當(dāng)前任務(wù)、已收集的參數(shù)、下一步動作存儲到數(shù)據(jù)庫或Redis中以conversation_iduser_id為鍵。工作流引擎對于極其復(fù)雜的任務(wù)可以考慮集成一個輕量級的工作流引擎如Prefect的核心邏輯或自己實現(xiàn)一個簡單的狀態(tài)機(jī)。將任務(wù)分解為多個步驟每個步驟可能調(diào)用一個技能或詢問用戶根據(jù)上一步的結(jié)果決定下一步的走向。這樣你的機(jī)器人就能處理“需求收集-方案制定-執(zhí)行-反饋”的完整閉環(huán)了。技能市場與動態(tài)加載當(dāng)技能越來越多可以設(shè)計一個技能注冊中心。甚至允許用戶在釘釘上通過特定指令“安裝”或“啟用”某個技能實現(xiàn)功能的動態(tài)擴(kuò)展。nanobot的插件化架構(gòu)為這提供了可能。這個“國產(chǎn)小龍蝦方案”從構(gòu)思到上線最深的體會是技術(shù)選型的“輕”與“巧”比“大”與“全”更重要。nanobot的輕量化讓我們能快速聚焦業(yè)務(wù)邏輯通義千問的本地化能力消除了合規(guī)顧慮釘釘則提供了現(xiàn)成的、高粘性的用戶入口。三者結(jié)合確實像吃小龍蝦一樣用最小的代價剝殼獲取了最大的滿足感吃肉。它可能不是功能最強(qiáng)大的Agent但絕對是能最快在你團(tuán)隊里跑起來、并真切解決一些痛點的那一個。接下來我打算為它增加一個“技能學(xué)習(xí)”功能讓團(tuán)隊成員能通過自然語言描述共同教會它處理新的瑣事讓這個“小工”越來越能干。