1. 項(xiàng)目緣起當(dāng)“派活”遇上AI一個(gè)真實(shí)的生產(chǎn)力痛點(diǎn)在團(tuán)隊(duì)協(xié)作里最讓人頭疼的往往不是技術(shù)難題本身而是那些瑣碎、重復(fù)但又必須有人去處理的“小活”。比如每天要手動(dòng)匯總幾個(gè)渠道的數(shù)據(jù)生成日?qǐng)?bào)或者根據(jù)客戶反饋的關(guān)鍵詞去內(nèi)部知識(shí)庫搜一圈資料再或者新同事入職時(shí)需要有人一步步告訴他項(xiàng)目環(huán)境怎么搭、代碼怎么拉。這些任務(wù)本身不復(fù)雜但極其消耗人的注意力和時(shí)間打斷深度工作流。過去我們可能會(huì)寫個(gè)腳本或者依賴某個(gè)固定的流程文檔但腳本需要觸發(fā)文檔需要人去看都不夠“絲滑”。直到AI Agent的概念火起來尤其是像nanobot這樣輕量、可編程的框架出現(xiàn)事情開始變得有趣。我們能不能做一個(gè)“智能小工”把它放到釘釘這種大家每天必用的協(xié)作工具里讓它7x24小時(shí)待命用自然語言就能給它派活它還能調(diào)用各種能力把活干完甚至把結(jié)果主動(dòng)推回來這個(gè)想法就是我們這個(gè)“國產(chǎn)小龍蝦方案”的起點(diǎn)——像吃小龍蝦一樣把那些瑣碎的“殼”重復(fù)勞動(dòng)剝掉只享受核心的“肉”價(jià)值創(chuàng)造。核心三件套nanobot作為機(jī)器人的“大腦”和“手腳”通義千問Qwen提供強(qiáng)大的自然語言理解與生成能力作為“思維”釘釘機(jī)器人則是它融入我們工作流的“身體”和“接口”。2. 方案核心組件選型與架構(gòu)解析為什么是這三個(gè)組合這背后是一套經(jīng)過權(quán)衡的“性價(jià)比”架構(gòu)。2.1 nanobot輕量級(jí)AI Agent的“骨架”nanobot不是一個(gè)大眾熟知的名字但在特定的開發(fā)者圈子里它正成為快速構(gòu)建AI Agent的熱門選擇。你可以把它理解為一個(gè)高度模塊化、事件驅(qū)動(dòng)的機(jī)器人框架。它核心解決了兩個(gè)問題“如何響應(yīng)”和“如何執(zhí)行”。事件驅(qū)動(dòng)模型nanobot的核心是監(jiān)聽各種事件比如釘釘?shù)南ⅰTTP接口調(diào)用、定時(shí)任務(wù)觸發(fā)然后根據(jù)預(yù)定義的邏輯進(jìn)行處理。這非常適合IM機(jī)器人場(chǎng)景你的消息就是觸發(fā)它的事件。技能Skill插件化它的能力不是固化的而是通過加載不同的“技能”插件來擴(kuò)展。一個(gè)技能就是一個(gè)獨(dú)立的Python模塊負(fù)責(zé)處理一類具體的任務(wù)比如“查詢天氣”、“執(zhí)行Shell命令”、“調(diào)用某個(gè)API”。這種設(shè)計(jì)讓功能的增刪改變得非常清晰和簡單。與LLM的松耦合nanobot本身不綁定任何特定的大語言模型LLM。它通過一個(gè)標(biāo)準(zhǔn)的接口與LLM對(duì)話你可以在配置文件中輕松切換不同的LLM服務(wù)提供商比如今天用通義千問明天想試試GPT改個(gè)配置就行代碼幾乎不用動(dòng)。選型理由相比于一些重型的、面向復(fù)雜規(guī)劃的Agent框架如LangChain、AutoGennanobot更輕、更直接學(xué)習(xí)曲線平緩對(duì)于“接收指令-執(zhí)行任務(wù)-返回結(jié)果”這類明確的工作流它的開銷和復(fù)雜度都更低更像一個(gè)“機(jī)器人操作系統(tǒng)”。2.2 通義千問Qwen本土化強(qiáng)大的“思維引擎”通義千問是阿里云推出的開源大語言模型系列。選擇它而非常見的OpenAI接口主要基于以下幾點(diǎn)考慮合規(guī)與可控性所有數(shù)據(jù)在國內(nèi)處理無需擔(dān)憂跨境數(shù)據(jù)傳輸?shù)恼吲c延遲風(fēng)險(xiǎn)這對(duì)于企業(yè)級(jí)應(yīng)用是首要考量。成本與性能Qwen系列提供了從7B到72B不同規(guī)模的模型并且有專門優(yōu)化的API服務(wù)。對(duì)于機(jī)器人場(chǎng)景通常不需要?jiǎng)佑们|參數(shù)模型一個(gè)效果不錯(cuò)的7B或14B模型通過其API調(diào)用在響應(yīng)速度和成本間能取得很好的平衡。實(shí)測(cè)下來Qwen在中文指令理解、上下文關(guān)聯(lián)和代碼生成方面表現(xiàn)相當(dāng)可靠。生態(tài)集成作為阿里系產(chǎn)品與釘釘、阿里云函數(shù)計(jì)算等服務(wù)的集成理論上會(huì)更順暢文檔和案例也更豐富。在我們的方案里Qwen扮演著“意圖理解”和“內(nèi)容生成”的核心角色。nanobot將用戶模糊的自然語言指令如“幫我查一下上周項(xiàng)目A的日志錯(cuò)誤”傳遞給QwenQwen需要解析出用戶的真實(shí)意圖“查詢?nèi)罩尽辈⑻崛￡P(guān)鍵參數(shù)“項(xiàng)目A”、“上周”、“錯(cuò)誤”。有時(shí)它還需要根據(jù)對(duì)話歷史進(jìn)行多輪澄清。2.3 釘釘機(jī)器人無縫嵌入工作流的“交互界面”釘釘機(jī)器人是釘釘開放平臺(tái)提供的能力它允許開發(fā)者創(chuàng)建一個(gè)自定義的機(jī)器人并將其添加到群聊或單聊中。它成為了我們AI Agent最自然的入口。開箱即用的通道無需自己搭建WebSocket或長輪詢釘釘負(fù)責(zé)消息的接收和推送我們只需要提供一個(gè)HTTPS的回調(diào)地址Webhook供釘釘在收到消息時(shí)調(diào)用。豐富的消息格式支持文本、鏈接、Markdown、ActionCard交互卡片等多種消息格式這讓我們的機(jī)器人回復(fù)可以非常美觀和交互性強(qiáng)。例如返回一個(gè)數(shù)據(jù)匯總時(shí)可以用Markdown表格提供一個(gè)操作選擇時(shí)可以用帶按鈕的卡片。安全的鑒權(quán)機(jī)制支持簽名驗(yàn)證確保發(fā)送到我們服務(wù)端的請(qǐng)求確實(shí)來自釘釘防止惡意調(diào)用。架構(gòu)全景圖用戶機(jī)器人并發(fā)送指令 - 釘釘服務(wù)器將消息封裝成HTTP POST請(qǐng)求發(fā)送到我們部署的nanobot服務(wù) - nanobot接收到事件觸發(fā)對(duì)應(yīng)的消息處理流程 - nanobot調(diào)用配置的Qwen API進(jìn)行意圖識(shí)別和參數(shù)提取 - 根據(jù)識(shí)別結(jié)果nanobot調(diào)用對(duì)應(yīng)的技能Skill來執(zhí)行具體任務(wù)如查詢數(shù)據(jù)庫、調(diào)用外部API、執(zhí)行腳本- 技能執(zhí)行完畢將結(jié)果返回給nanobot - nanobot將結(jié)果格式化為釘釘消息通過釘釘提供的API發(fā)送回對(duì)應(yīng)的群聊或會(huì)話。整個(gè)流程形成了一個(gè)閉環(huán)。3. 從零到一環(huán)境搭建與核心配置實(shí)戰(zhàn)理論講完我們動(dòng)手搭一個(gè)。假設(shè)你已經(jīng)有一個(gè)Linux服務(wù)器或本地開發(fā)環(huán)境并且擁有釘釘開發(fā)者賬號(hào)和阿里云API密鑰。3.1 基礎(chǔ)環(huán)境準(zhǔn)備首先為我們的“小龍蝦”準(zhǔn)備一個(gè)干凈的“廚房”。# 1. 創(chuàng)建并進(jìn)入項(xiàng)目目錄 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ū)技能包例如一個(gè)用于HTTP請(qǐng)求的通用技能 pip install nanobot-skill-http # 4. 安裝通義千問的SDK # 這里假設(shè)使用DashScope阿里云靈積的API它是調(diào)用通義千問的官方途徑 pip install dashscope3.2 釘釘機(jī)器人創(chuàng)建與配置這是連接現(xiàn)實(shí)世界的第一步。登錄釘釘開發(fā)者后臺(tái)找到你的組織進(jìn)入“應(yīng)用開發(fā)” - “企業(yè)內(nèi)部開發(fā)” - “機(jī)器人”。創(chuàng)建機(jī)器人點(diǎn)擊“創(chuàng)建應(yīng)用”選擇“機(jī)器人”填寫名稱和描述。創(chuàng)建成功后記錄下AppKey和AppSecret這是機(jī)器人身份的憑證。配置消息接收在機(jī)器人詳情頁找到“消息接收”配置。這里需要填寫一個(gè)公網(wǎng)可訪問的URL即我們即將部署的nanobot服務(wù)的Webhook地址。例如https://your-server.com/dingtalk/callback。在開發(fā)階段你可以使用內(nèi)網(wǎng)穿透工具如ngrok、frp將本地服務(wù)暴露為一個(gè)臨時(shí)公網(wǎng)地址進(jìn)行測(cè)試。獲取Webhook地址在“機(jī)器人詳情”頁你還能找到一個(gè)“Webhook”地址格式如https://oapi.dingtalk.com/robot/send?access_tokenXXX。這個(gè)地址用于主動(dòng)發(fā)送消息到釘釘。我們nanobot在完成任務(wù)后會(huì)調(diào)用這個(gè)地址把結(jié)果推回去。請(qǐng)妥善保管這個(gè)access_token。發(fā)布與安裝將機(jī)器人發(fā)布到你的組織并把它添加到需要使用的釘釘群中。3.3 nanobot項(xiàng)目初始化與核心邏輯編寫現(xiàn)在我們來編寫機(jī)器人的“大腦”和“反射弧”。# 在項(xiàng)目根目錄初始化一個(gè)nanobot項(xiàng)目結(jié)構(gòu) nanobot init my_dingtalk_bot cd my_dingtalk_bot你會(huì)看到一個(gè)基礎(chǔ)的項(xiàng)目結(jié)構(gòu)包含configs,skills,main.py等。我們主要關(guān)注三個(gè)文件1. 配置文件 (configs/config.yaml): 機(jī)器人的“基因”name: 小龍蝦助手 description: 一個(gè)幫你處理瑣事的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 # 一個(gè)自定義的示例技能 - nanobot_skill_http # 引入的第三方HTTP技能 # 事件監(jiān)聽器配置處理釘釘消息 listeners: - name: dingtalk_webhook type: webhook # 監(jiān)聽Webhook請(qǐng)求 path: /dingtalk/callback # 對(duì)應(yīng)釘釘后臺(tái)配置的路徑 method: POST2. 自定義技能 (skills/my_custom_skill.py): 機(jī)器人的“手藝”技能是能力的載體。這里我們創(chuàng)建一個(gè)簡單的技能它能夠理解用戶關(guān)于“時(shí)間”的詢問。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)前時(shí)間或日期) class TimeQuerySkill(Skill): 一個(gè)簡單的示例技能當(dāng)用戶詢問時(shí)間時(shí)返回當(dāng)前時(shí)間。 async def execute(self, context, **kwargs): 技能執(zhí)行入口。 context: 包含當(dāng)前會(huì)話、用戶消息等上下文信息。 user_message context.current_message.text # 這里可以做得更智能比如用LLM判斷意圖。 # 但作為示例我們簡單匹配關(guān)鍵詞。 if 時(shí)間 in user_message or 幾點(diǎn) in user_message: now datetime.now().strftime(%Y-%m-%d %H:%M:%S) reply_text f現(xiàn)在是北京時(shí)間{now} # 返回一個(gè)文本消息對(duì)象nanobot會(huì)將其轉(zhuǎn)換為釘釘消息格式 return TextMessage(contentreply_text) # 如果不匹配返回Nonenanobot會(huì)嘗試其他技能或使用LLM直接回復(fù) return None3. 主程序適配釘釘 (main.py): 消息的“翻譯官”nanobot默認(rèn)的Webhook監(jiān)聽器可能不直接兼容釘釘?shù)南⒏袷?。釘釘發(fā)送過來的是一套特定的JSON結(jié)構(gòu)。我們需要一個(gè)適配器來“翻譯”。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() # 自定義一個(gè)釘釘消息解析函數(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)在幾點(diǎn)” # 這里簡單處理實(shí)際可能需要更精確的解析 if content.startswith(): # 簡單移除第一個(gè)和其后的空格或直到下一個(gè)空格 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, ) # 會(huì)話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 # 釘釘簽名驗(yàn)證函數(shù)重要用于安全驗(yàn)證 def verify_dingtalk_signature(timestamp: str, sign: str, secret: str, body: str): 驗(yàn)證釘釘回調(diào)請(qǐng)求的簽名。 參考釘釘官方文檔https://open.dingtalk.com/document/robots/customize-robot-security-settings if not secret: return True # 如果未設(shè)置SECRET跳過驗(yàn)證不推薦生產(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. 簽名驗(yàn)證 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)用對(duì)于簡單場(chǎng)景夠用。復(fù)雜場(chǎng)景可考慮異步隊(duì)列。 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): # 如果消息對(duì)象有轉(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 # 啟動(dòng)FastAPI服務(wù)監(jiān)聽在8080端口 uvicorn.run(app, host0.0.0.0, port8080)這個(gè)main.py做了幾件關(guān)鍵事驗(yàn)證釘釘請(qǐng)求的真實(shí)性、將釘釘格式的消息“翻譯”成nanobot能懂的結(jié)構(gòu)、將nanobot處理后的結(jié)果再“翻譯”回釘釘格式并發(fā)送。你需要將YOUR_ACCESS_TOKEN替換為你的機(jī)器人Webhook地址中的token并實(shí)現(xiàn)send_to_dingtalk函數(shù)可以使用aiohttp庫。3.4 連接通義千問讓機(jī)器人“聽懂人話”上面的自定義技能只是簡單匹配關(guān)鍵詞。要真正理解復(fù)雜、模糊的指令必須引入Qwen。我們需要修改技能讓它學(xué)會(huì)“問”Qwen。首先確保在config.yaml中正確配置了llm部分并且環(huán)境變量DASHSCOPE_API_KEY已設(shè)置。然后我們創(chuàng)建一個(gè)更高級(jí)的技能例如一個(gè)“智能問答”技能它會(huì)把所有無法被其他技能處理的指令都交給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): 一個(gè)利用Qwen進(jìn)行通用問答和意圖理解的技能。 它可以作為兜底技能處理其他技能無法處理的模糊指令。 def __init__(self, config): super().__init__(config) # 從nanobot的配置中獲取LLM客戶端實(shí)例 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)建一個(gè)系統(tǒng)提示詞引導(dǎo)Qwen更好地理解機(jī)器人場(chǎng)景 system_prompt 你是一個(gè)部署在釘釘上的AI助手名叫“小龍蝦助手”。你的主要職責(zé)是幫助用戶處理工作瑣事。 用戶可能會(huì)給你一些模糊的指令你需要嘗試?yán)斫馄浔澈蟮囊鈭D。 常見的意圖包括但不限于查詢信息天氣、時(shí)間、股票、公司內(nèi)部數(shù)據(jù)、執(zhí)行簡單計(jì)算、翻譯、總結(jié)文本、生成代碼片段、回答問題等。 如果你判斷用戶的指令需要調(diào)用某個(gè)具體工具或技能比如查天氣需要調(diào)用API你可以回復(fù)一個(gè)結(jié)構(gòu)化的指令例如 [ACTION:weather_query location北京]。 否則請(qǐng)直接給出友好、有幫助的文本回答。 請(qǐng)用中文回復(fù)。 # 準(zhǔn)備對(duì)話歷史如果有的話 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ā)另一個(gè)對(duì)應(yīng)的技能如weather_query # 這是一個(gè)簡單的實(shí)現(xiàn)示例 if ai_reply.startswith([ACTION:) and ai_reply.endswith(]): # 解析動(dòng)作和參數(shù)這里簡化處理 action_str ai_reply[8:-1] # 去掉[ACTION:和] # 理論上這里應(yīng)該調(diào)度到其他技能本例中我們只做演示 logger.info(f識(shí)別到結(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ù)暫時(shí)開小差了請(qǐng)稍后再試。)別忘了在config.yaml的skills列表里加上這個(gè)新技能- “skills.smart_qa_skill”?，F(xiàn)在你的機(jī)器人就接上了通義千問的“大腦”能夠理解更復(fù)雜的指令并進(jìn)行智能對(duì)話了。4. 部署、調(diào)試與進(jìn)階實(shí)戰(zhàn)技巧讓代碼跑起來只是第一步讓它穩(wěn)定、好用才是關(guān)鍵。4.1 本地調(diào)試與內(nèi)網(wǎng)穿透在開發(fā)階段你的服務(wù)運(yùn)行在本地localhost釘釘?shù)姆?wù)器無法直接訪問。你需要一個(gè)“橋梁”。使用 ngrok (最簡便)# 下載ngrok并注冊(cè)獲取authtoken ngrok config add-authtoken YOUR_AUTH_TOKEN # 將本地的8080端口暴露到公網(wǎng) ngrok http 8080運(yùn)行后ngrok會(huì)給你一個(gè)隨機(jī)的https://xxx.ngrok.io地址。將這個(gè)地址后面加上/dingtalk/callback填到釘釘機(jī)器人的“消息接收”URL中。使用 frp (更穩(wěn)定可控)如果你有自己的云服務(wù)器可以搭建frp服務(wù)獲得一個(gè)固定的二級(jí)域名更適合長期測(cè)試。調(diào)試技巧在main.py中多使用logging打印關(guān)鍵信息如接收到的原始數(shù)據(jù)、解析后的消息、調(diào)用Qwen的請(qǐng)求和響應(yīng)。釘釘機(jī)器人平臺(tái)也有“消息推送記錄”可以查看發(fā)送是否成功。4.2 服務(wù)器部署與進(jìn)程守護(hù)開發(fā)完成后需要部署到正式的服務(wù)器。使用 systemd (Linux)創(chuàng)建一個(gè)服務(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然后啟用并啟動(dòng)服務(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)境一鍵啟動(dòng)。4.3 設(shè)計(jì)更強(qiáng)大的技能以“數(shù)據(jù)查詢”為例一個(gè)真正的“派活”機(jī)器人必須能操作真實(shí)的數(shù)據(jù)或系統(tǒng)。我們?cè)O(shè)計(jì)一個(gè)技能讓機(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ù)”, “報(bào)表”]) # 可以加一些觸發(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)行意圖識(shí)別和SQL生成。 # 為了安全我們不做全自然語言轉(zhuǎn)SQL而是預(yù)定義幾個(gè)查詢模板。 # 例如用戶說“查一下上周的訂單總數(shù)” # 我們可以用LLM提取出“上周”和“訂單總數(shù)”然后映射到預(yù)定義的SQL模板。 # 假設(shè)我們通過LLM或簡單規(guī)則已經(jīng)確定了查詢參數(shù) query_type “l(fā)ast_week_order_count” # 這個(gè)應(yīng)由更高級(jí)的意圖識(shí)別模塊提供 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ù)時(shí)出現(xiàn)錯(cuò)誤請(qǐng)檢查數(shù)據(jù)庫連接或查詢語句?！? async def cleanup(self): 技能卸載時(shí)關(guān)閉連接池 if self.pool: self.pool.close() await self.pool.wait_closed()這個(gè)技能展示了如何連接外部資源。關(guān)鍵點(diǎn)在于永遠(yuǎn)不要將用戶輸入直接拼接成SQLSQL注入風(fēng)險(xiǎn)應(yīng)該通過LLM或固定模板將自然語言轉(zhuǎn)換為安全的查詢參數(shù)。4.4 安全與權(quán)限管控把機(jī)器人放進(jìn)工作群安全是重中之重。請(qǐng)求簽名驗(yàn)證如前文代碼所示必須驗(yàn)證釘釘回調(diào)的簽名(timestampsign)這是防止偽造請(qǐng)求的第一道防線。技能執(zhí)行權(quán)限不是所有用戶都能調(diào)用所有技能?？梢栽赟kill的execute方法開頭檢查context.current_message.user_id對(duì)照一個(gè)權(quán)限列表可以配置在文件或數(shù)據(jù)庫里。例如只有管理員才能執(zhí)行“重啟服務(wù)”這樣的高危技能。敏感信息脫敏從數(shù)據(jù)庫或API查詢到的結(jié)果在發(fā)送到釘釘前要過濾掉手機(jī)號(hào)、身份證號(hào)、密鑰等敏感信息。限流與防刷對(duì)用戶的請(qǐng)求頻率做限制防止惡意調(diào)用消耗你的LLM API額度或數(shù)據(jù)庫資源。可以在main.py的入口處添加簡單的計(jì)數(shù)器或使用Redis實(shí)現(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)控與日志一個(gè)健壯的服務(wù)離不開監(jiān)控。結(jié)構(gòu)化日志使用Python的logging模塊配置JSON格式輸出方便接入ELKElasticsearch, Logstash, Kibana或類似日志系統(tǒng)。記錄每一次請(qǐng)求、LLM調(diào)用、技能執(zhí)行結(jié)果和耗時(shí)。健康檢查為你的FastAPI服務(wù)添加一個(gè)/health端點(diǎn)返回服務(wù)狀態(tài)、數(shù)據(jù)庫連接狀態(tài)等。方便運(yùn)維監(jiān)控。關(guān)鍵指標(biāo)統(tǒng)計(jì)技能調(diào)用次數(shù)、Qwen API調(diào)用耗時(shí)與Token消耗、錯(cuò)誤率等。這些數(shù)據(jù)可以幫助你優(yōu)化技能設(shè)計(jì)和成本控制。5. 避坑指南與效能提升心法在實(shí)際開發(fā)和運(yùn)維中我踩過不少坑也總結(jié)了一些讓“小龍蝦”更好用的經(jīng)驗(yàn)。5.1 意圖識(shí)別的“最后一公里”難題最初我試圖讓Qwen直接解析所有指令并返回可執(zhí)行的命令。但發(fā)現(xiàn)對(duì)于邊界模糊的指令效果不穩(wěn)定。比如“看看張三上周的業(yè)績”Qwen可能理解成“查詢數(shù)據(jù)庫”也可能直接生成一段文本描述。我的解決方案是分層處理第一層精確技能匹配。先定義一批高頻、明確的技能并用關(guān)鍵詞或正則表達(dá)式觸發(fā)。如“/天氣 北京”、“打卡”。第二層LLM意圖分類。對(duì)于未匹配的指令交給Qwen但任務(wù)不是直接執(zhí)行而是分類。我訓(xùn)練或通過提示詞引導(dǎo)Qwen將指令分類到預(yù)定義的幾個(gè)“技能槽”如[QUERY_DATA]、[CALCULATE]、[GENERATE_TEXT]等并提取結(jié)構(gòu)化參數(shù)。第三層技能執(zhí)行與LLM潤色。根據(jù)分類結(jié)果調(diào)用對(duì)應(yīng)的技能獲取原始數(shù)據(jù)或結(jié)果再將這個(gè)結(jié)果和原始指令一起交給Qwen讓它生成一段通順、友好的最終回復(fù)。這樣既保證了關(guān)鍵動(dòng)作的準(zhǔn)確執(zhí)行又利用了LLM的泛化理解和語言生成能力。5.2 釘釘消息格式的“坑”釘釘?shù)南⒏袷奖容^獨(dú)特稍不注意就會(huì)發(fā)送失敗或顯示異常。Markdown表格對(duì)齊釘釘?shù)腗arkdown對(duì)表格支持有限復(fù)雜的表格可能渲染錯(cuò)亂。建議簡單表格用|復(fù)雜數(shù)據(jù)可以考慮用“字段值”的列表形式或者直接生成圖片通過技能調(diào)用圖表生成API發(fā)送。特定用戶在機(jī)器人回復(fù)中想某人需要在文本中使用用戶手機(jī)號(hào)并且同時(shí)傳遞at字段。但獲取群成員的手機(jī)號(hào)需要額外的釘釘API權(quán)限需申請(qǐng)。通常我們只在回復(fù)中觸發(fā)指令的用戶context.current_message.sender_id對(duì)應(yīng)的手機(jī)號(hào)。消息長度限制釘釘單條文本消息有長度限制約5000字符。如果回復(fù)內(nèi)容很長需要主動(dòng)切割成多條發(fā)送或者先生成一個(gè)文件如txt然后發(fā)送文件消息。ActionCard按鈕回調(diào)如果你使用了交互卡片按鈕點(diǎn)擊后的回調(diào)地址也必須是公網(wǎng)可訪問的且同樣需要處理簽名驗(yàn)證。這相當(dāng)于為你的機(jī)器人增加了“事件”處理能力可以實(shí)現(xiàn)更復(fù)雜的交互流程。5.3 成本控制與性能優(yōu)化Qwen API是按Token收費(fèi)的無節(jié)制地使用會(huì)讓賬單飛漲。設(shè)計(jì)精簡的提示詞Prompt系統(tǒng)提示詞不要過于冗長。在滿足指令清晰的前提下盡量簡短。將固定的上下文如公司背景、常用指令說明做得精煉。緩存LLM響應(yīng)對(duì)于一些常見、結(jié)果相對(duì)固定的查詢?nèi)纭肮竞喗椤?、“產(chǎn)品列表”可以將Qwen的回復(fù)緩存起來用Redis或內(nèi)存緩存設(shè)定一個(gè)合理的過期時(shí)間。下次遇到相同或相似問題時(shí)直接返回緩存結(jié)果大幅節(jié)省Token和延遲。設(shè)置使用限額為每個(gè)用戶或每個(gè)群設(shè)置每日/每周的LLM調(diào)用次數(shù)上限?？梢栽诩寄軋?zhí)行前檢查計(jì)數(shù)。異步與非阻塞在main.py中處理釘釘回調(diào)并調(diào)用LLM和技能的過程如果耗時(shí)較長5秒釘釘可能會(huì)超時(shí)重試。建議將核心處理邏輯放入異步任務(wù)隊(duì)列如Celery Redis或直接用asyncio.create_task在收到回調(diào)后立即返回“ok”然后異步處理任務(wù)處理完再主動(dòng)推送結(jié)果。這需要更復(fù)雜的架構(gòu)但能提供更好的用戶體驗(yàn)和可靠性。5.4 從“玩具”到“工具”的演進(jìn)一個(gè)能回答問題的機(jī)器人和一個(gè)能真正“派活”的Agent差距在于狀態(tài)管理和工作流。會(huì)話狀態(tài)nanobot本身支持簡單的會(huì)話上下文。但對(duì)于一個(gè)需要多步交互的復(fù)雜任務(wù)比如“幫我訂會(huì)議室要下午2點(diǎn)10個(gè)人的”你需要自己維護(hù)一個(gè)更強(qiáng)大的狀態(tài)機(jī)。可以將對(duì)話狀態(tài)當(dāng)前任務(wù)、已收集的參數(shù)、下一步動(dòng)作存儲(chǔ)到數(shù)據(jù)庫或Redis中以conversation_iduser_id為鍵。工作流引擎對(duì)于極其復(fù)雜的任務(wù)可以考慮集成一個(gè)輕量級(jí)的工作流引擎如Prefect的核心邏輯或自己實(shí)現(xiàn)一個(gè)簡單的狀態(tài)機(jī)。將任務(wù)分解為多個(gè)步驟每個(gè)步驟可能調(diào)用一個(gè)技能或詢問用戶根據(jù)上一步的結(jié)果決定下一步的走向。這樣你的機(jī)器人就能處理“需求收集-方案制定-執(zhí)行-反饋”的完整閉環(huán)了。技能市場(chǎng)與動(dòng)態(tài)加載當(dāng)技能越來越多可以設(shè)計(jì)一個(gè)技能注冊(cè)中心。甚至允許用戶在釘釘上通過特定指令“安裝”或“啟用”某個(gè)技能實(shí)現(xiàn)功能的動(dòng)態(tài)擴(kuò)展。nanobot的插件化架構(gòu)為這提供了可能。這個(gè)“國產(chǎn)小龍蝦方案”從構(gòu)思到上線最深的體會(huì)是技術(shù)選型的“輕”與“巧”比“大”與“全”更重要。nanobot的輕量化讓我們能快速聚焦業(yè)務(wù)邏輯通義千問的本地化能力消除了合規(guī)顧慮釘釘則提供了現(xiàn)成的、高粘性的用戶入口。三者結(jié)合確實(shí)像吃小龍蝦一樣用最小的代價(jià)剝殼獲取了最大的滿足感吃肉。它可能不是功能最強(qiáng)大的Agent但絕對(duì)是能最快在你團(tuán)隊(duì)里跑起來、并真切解決一些痛點(diǎn)的那一個(gè)。接下來我打算為它增加一個(gè)“技能學(xué)習(xí)”功能讓團(tuán)隊(duì)成員能通過自然語言描述共同教會(huì)它處理新的瑣事讓這個(gè)“小工”越來越能干。