事情要从一个周五的下午说起。我待的技术群那天格外热闹。有人贴了一屏报错日志有人问中间件到底怎么配才对还有个人直接让我帮他写一条抓取特定格式的正则。我一边开会一边瞟群消息回着回着实在顾不过来脑子里冒出个念头要不弄个机器人顶上这个念头后来变成了CloddsBot。名字没什么高深含义——Clodds 是从 clouds 随手改出来的加了个 d意思是这机器人跑在云上、用着云端的模型干些云一样轻但覆盖面特别广的杂活。它本质是一个基于大模型 API 的智能聊天机器人服务通过 Webhook 接收消息做意图识别再决定是直接回答、查天气、记待办还是设提醒。这篇文章我把 CloddsBot 从需求拆解到部署运维完完整整讲一遍。适合谁看一个是想自己搞个聊天机器人但不知道从哪下手的人另一个是已经在用大模型 API 写应用、却被上下文管理和重试逻辑反复折磨过的人。我会把每个关键决策背后的为什么都讲清楚而不是只丢一段能跑的代码。1. 从群聊没人回到 CloddsBot这个机器人到底在解决什么问题1.1 需求是怎么长出来的项目启动之前我花了大概一个晚上整理需求。不整理不行聊天机器人这个方向看着简单实际一展开就是无底洞什么多轮对话、情感识别、语音合成、知识库问答都能往里塞。我给自己定的原则很朴素只解决那些每周都会发生的、重复度高的麻烦。我当时列了这么几类高频需求群里有人问这个报错啥意思其实把错误信息丢给大模型就有答案。隔三差五有人让帮忙查某个城市的天气一次两次无所谓次数多了真烦。我自己经常忘事需要有一个随手说一句提醒我下午三点开会就能记住的地方。群聊里经常插入一些和主题无关的闲聊如果我逐个回复一天的时间全没了。这四类需求对应到机器人能力上就是自由对话、天气查询、提醒备忘、自动处理重复性的文字劳动。注意这里没有语音多模态自动进群踢人这些花活不是它们不好而是我第一个版本不需要。1.2 现成方案为什么不够用决定自造轮子之前我其实是先看了一圈现成方案。老牌的 IRC 机器人和各种开源群管机器人有个通病逻辑靠关键词匹配规则写死了。你说天气 北京它能命中你说北京今天冷吗它就傻了。后来也试过接大模型套壳的开源项目但那些项目普遍为了演示效果把功能堆得特别满光配置文件就有几百行部署完我根本不知道它哪些逻辑是可用的、哪些是作者一拍脑袋写的。更关键的一点是现成方案解决不了这个机器人是我自己的延伸这件事。我希望它能按照我的语气说话记住和我相关的待办事项而不是每次对话都从零开始。这些东西不是靠配置能配出来的得自己改代码。1.3 边界比功能更重要CloddsBot 第一个版本的需求边界我是这么定的要做接收文本消息、多轮对话、天气查询、待办提醒、基础信息查询、按会话隔离上下文。不做不做用户画像、不做情感分析、不主动推送消息、不开放任意系统命令执行、不接语音和图片理解。这个边界在后来帮了大忙。比如有朋友问我为什么不做成 plug-in 架构我说第一版不需要——我需要的是把整条链路跑通而不是先把插件系统设计出花来。等项目跑了一段时间你会发现哪些功能是真的高频、哪些只是你以为的高频这时候再抽象插件系统才能抽象对地方。这是我做这个项目学到最实在的一课机器人的价值判断标准不是它能做多少事而是它在日常里被真正用起了几次。2. 技术栈与整体设计先把地基夯实再谈智能2.1 为什么是 Python asyncio aiohttpCloddsBot 的消息入口是一个 Webhook 服务也就是说 IM 平台收到消息后会通过 HTTP 请求往我这边推一条消息。这里有个天然的并发场景用户 A 的消息还在等大模型回复用户 B 的消息又进来了。所以我需要的是一个异步非阻塞的运行时而不是传统的阻塞式多线程。我选的是 Python 3.11 asyncio aiohttp。原因很直接我已经在用 Python 生态做大模型相关开发SDK 支持最成熟。aiohttp 做 Webhook 服务足够轻不需要为了一个消息入口引一个 FastAPI 全家桶当然你用 FastAPI 也没问题纯粹是取舍。异步编程在处理 HTTP 长连接时优势明显一个请求在等响应的时候事件循环可以继续处理别的请求。这里有一个常见的误判很多人以为并发高才需要异步。其实聊天机器人场景里你的瓶颈不是 QPS而是连接数和等待时间。就算一分钟只有 20 条消息每条消息要等大模型 5 到 30 秒如果同步写消息就得排长队体验极差。2.2 模块划分先画好边界再写代码CloddsBot 的目录结构长这样cloddsbot/ ├── main.py # 入口加载配置、装配模块、启动服务 ├── config.yaml # 配置文件 ├── requirements.txt ├── clodds/ │ ├── __init__.py │ ├── config.py # 配置加载与校验 │ ├── server.py # Webhook 服务 │ ├── adapter/ # 消息适配层 │ │ ├── base.py # 统一消息抽象 │ │ └── webhook.py # 具体平台的推送解析 │ ├── llm/ │ │ ├── client.py # 大模型 API 封装 │ │ └── prompts.py # 系统提示词 │ ├── memory/ │ │ └── session.py # 会话状态管理 │ ├── tools/ │ │ ├── registry.py # 工具注册表 │ │ ├── weather.py # 天气查询 │ │ └── reminder.py # 提醒备忘 │ └── scheduler/ │ └── runner.py # 定时任务调度每个模块只做一件事依赖方向是从上往下的server 调 adapter 和 llmllm 调 toolsmemory 被 llm 层调用谁都不许反向依赖。这个设计不是一上来就有的是我写到第三天发现 tools 和 adapter 开始互相 import 才抽出来的。2.3 配置管理的细节配置这块我踩过一次坑早期为了省事把 API Key 直接写在代码里结果有一次差点把仓库推到公开仓库去吓出一身冷汗。后来老老实实做了三件事配置全部集中在config.yaml用 yaml 文件管理。敏感信息API Key、Token一律从环境变量读取配置文件里只写变量名。准备config.example.yaml提交到仓库真实配置放在本地 gitignore。配置文件示例bot: name: CloddsBot version: 0.3.0 webhook: host: 0.0.0.0 port: 8080 path: /webhook/msg llm: provider: anthropic model: claude-sonnet-4-5 api_key_env: CLODDS_API_KEY temperature: 0.4 max_tokens: 1024 memory: backend: redis session_ttl_hours: 72 max_context_rounds: 12 scheduler: timezone: Asia/Shanghai notify_url: http://localhost:9090/notify模型名那里我特意留了注释按你账户实际开通的版本来填不同的版本能力差异、价格差异都很大。配置文件中一个容易被忽略的点是session_ttl_hours这个决定了一段对话能记多久。我设的是 72 小时超过之后会话自动清空避免 Redis 里的历史越攒越多。3. 消息链路的完整实现从 Webhook 到工具调用3.1 统一消息抽象层CloddsBot 的完整消息链路是IM 平台推送 → Webhook → 适配层解析 → 会话管理 → 大模型调用 → 工具调用如果需要→ 回复发送。这条链路里最重要的一环是抽象层因为它决定了未来你想接更多消息源时要不要重写整个项目。我定义了一个非常薄的消息模型# adapter/base.py from dataclasses import dataclass, field from datetime import datetime dataclass class IncomingMessage: message_id: str # 平台消息 ID用于幂等去重 chat_id: str # 会话 ID可以是群 ID 或用户 ID sender_id: str # 发送者 ID sender_name: str text: str raw: dict field(default_factorydict) property def is_group(self) - bool: return self.chat_id ! self.sender_id为什么要单独抽一个 IncomingMessage因为不同平台的消息结构五花八门有的字段叫from有的叫user_id有的消息里还带各种扩展字段。如果业务逻辑里到处直接操作原始消息体将来每接一个新平台就得到处改判断逻辑。统一成这个模型之后后面所有环节都只认 IncomingMessage。is_group这个属性是后来加的。群聊和单聊的处理策略很不一样群聊里机器人被 了才回复单聊里所有消息都要回。这个逻辑放在抽象层是因为各个平台判断是否被 到的方式不同但抽象后对所有上层模块透明。3.2 LLM 调用的封装LLM 调用是整个机器人最核心的环节。我用的 SDK 是anthropic异步版本# llm/client.py import os from anthropic import AsyncAnthropic class LLMClient: def __init__(self, config): self.api_key os.environ[config[llm][api_key_env]] self.model config[llm][model] self.temperature config[llm][temperature] self.max_tokens config[llm][max_tokens] self.client AsyncAnthropic(api_keyself.api_key) async def chat(self, system: str, messages: list, tools: list): resp await self.client.messages.create( modelself.model, systemsystem, messagesmessages, toolstools or [], temperatureself.temperature, max_tokensself.max_tokens, ) return resp这里有两个容易忽视的点。第一个是max_tokens。我一开始设的比较大2048后来看统计发现绝大多数回复根本用不到那么多 token还白白增加延迟和费用改成 1024 后日常场景完全够用。这个值应该根据你的实际场景去测而不是抄别人的配置。第二个是异常处理。API 调用在任何不可靠的网络上都有可能超时或返回 5xx这段代码的 try-except 我故意省略了但实际项目里是必须有的——后面有一节专门讲这个。3.3 工具注册表让机器人学会动手光会聊天的大模型只是个聊天框CloddsBot 真正有用的地方是它能调用工具。这里用到了 Function Calling在 Anthropic 的术语里叫 Tool Use。我先定义了一个工具注册表所有工具函数都通过装饰器注册# tools/registry.py import inspect from typing import Callable TOOL_SCHEMAS [] TOOL_FUNCTIONS {} def register_tool(schema: dict): def decorator(func: Callable): name schema[name] TOOL_SCHEMAS.append(schema) TOOL_FUNCTIONS[name] func return func return decorator def get_tools(): return TOOL_SCHEMAS async def execute_tool(name: str, arguments: dict): func TOOL_FUNCTIONS.get(name) if func is None: return {error: ftool {name} not found} return await func(**arguments)然后写具体的工具函数。比如天气查询# tools/weather.py import aiohttp from .registry import register_tool register_tool({ name: query_weather, description: 查询指定城市的当前天气、温度和风力情况, input_schema: { type: object, properties: { city: {type: string, description: 城市名称如 北京、上海、深圳} }, required: [city] } }) async def query_weather(city: str): # 这里对接一个公开的天气 API做参数校验和超时处理 async with aiohttp.ClientSession() as session: async with session.get( fhttps://example.com/api/weather, params{city: city}, timeoutaiohttp.ClientTimeout(total5) ) as resp: data await resp.json() return {city: city, weather: data.get(condition, 未知), temp: data.get(temp)}注册表的好处是新增工具时不需要改动调用链路的代码。写完一个函数加一个装饰器它就自动出现在传给大模型的 tools 列表里了。这个轻量方案比引入一个重量级框架更适合我现在这个阶段的项目。3.4 Tool Use 的闭环一次完整对话的旅程工具调用不是大模型自己偷偷调而是一个多回合的闭环。我画一下这个流程用户说北京今天天气怎么样Adapter 把消息转成 IncomingMessage。会话管理器取出该会话的历史消息加上系统提示词和工具列表一起发给大模型。大模型返回的不是最终文本而是一个tool_use块内容是{name: query_weather, input: {city: 北京}}。机器人执行query_weather(北京)拿到天气结果。把工具执行结果作为一条user角色消息回传给大模型。大模型看到工具结果生成最终回答北京现在是晴12 度风力三级。这个流程用代码写出来大概是# 伪代码简化版 tool_use 循环 from anthropic.types import Message async def process_message(session, user_text: str): messages session.get_history() [{role: user, content: user_text}] for _ in range(5): # 最大工具调用轮数防止死循环 resp: Message await llm.chat( systemsession.system_prompt, messagesmessages, toolsget_tools(), ) content_blocks resp.content tool_uses [b for b in content_blocks if b.type tool_use] if not tool_uses: # 没有工具调用直接返回文本给用户 text .join(b.text for b in content_blocks if b.type text) session.append(assistant, text) return text # 有工具调用先把 assistant 消息加入历史 messages.append({role: assistant, content: content_blocks}) # 执行每个工具把结果作为 user 消息接回去 for tool_use in tool_uses: result await execute_tool(tool_use.name, tool_use.input) messages.append({ role: user, content: [ { type: tool_result, tool_use_id: tool_use.id, content: str(result), } ], }) return 工具调用超限请稍后再试注意那个max 5的循环边界。没有这个边界如果大模型连续调用工具多次比如某个工具返回的数据又触发另一个工具理论上会无限循环。我第一次跑的时候就是没设边界结果机器人自己连续调用了一串工具虽然最后还是回到了正轨但白白浪费了几万个 token。3.5 响应如何回到用户手里最后一步是把生成的回复发送出去。我这里的做法是先把回复写入一个发送队列由专门的发送任务负责推送。为什么不直接发因为 Webhook 处理函数应该尽快返回告诉平台消息收到了否则平台会认为推送失败并反复重试。大模型生成可能要 10 秒Webhook 不可能等 10 秒再返回 200。所以正确的顺序是异步入队立即返回 200后台发送。# server.py简化 from aiohttp import web async def handle_message(request): raw await request.json() msg adapter.parse(raw) # 幂等去重同一个 message_id 只处理一次 if not await dedup_pipeline.check_and_add(msg.message_id): return web.Response(status200) # 入队异步处理 asyncio.create_task(processing_pipeline.run(msg)) return web.Response(status200)这个设计在后面排障那节还会再提到它引出了两个新问题一是消息去重二是后台任务的可靠性。4. 上下文管理CloddsBot 不失忆、不跑题的工程方案4.1 会话隔离与状态存储聊天机器人最大的坑之一就是上下文管理。刚开始我用一个全局列表存历史消息测试的时候没感觉一进群发现所有人都共享同一个记忆——A 说了自己的名字B 再问我叫什么机器人回的是 A 的名字。这就是典型的会话串线。解决办法是按chat_id做隔离。每个会话独立保存自己的历史消息列表。存储后端我选的是 Redis原因有两个一是天然支持 TTL我可以给每个会话设置 72 小时过期时间二是重启不丢数据不像内存 dict服务一崩所有历史全没了。Redis 里的 key 结构大概是session:{chat_id}:messages # 历史消息列表JSON 编码 session:{chat_id}:meta # 会话元信息比如上次活跃时间每次有新消息就把这条消息追加到列表尾部同时用 LPUSH/RPUSH 配合 LTRIM 控制列表长度。4.2 窗口滑动与 token 预算大模型的上下文窗口是有限的就算最新模型的窗口很大也不可能无限堆历史。原因不只是窗口大小还有两个实际约束成本每多一个 token 都要付钱和延迟上下文越长首字延迟越高这是实打实的用户体验问题。我采用的策略是给每个会话设一个 token 预算超了就从头裁剪。分成几部分上下文组成部分预算占比说明系统提示词15%角色设定、功能说明每次请求都带工具定义25%传给大模型的 tools schema对话历史50%最近 N 轮消息本轮输入10%用户当前这条消息怎么算 token最简单的方式是调用 API 前用len(text) // 2粗估因为中文在大多数 tokenizer 里大概是 1 到 1.5 个字节对应一个 token。更准确的做法是直接用 SDK 提供的 tokenizer 接口把消息内容逐个编码再求和。项目初期我用的是粗估法够用等要精细控制成本时才换成精确 tokenizer。4.3 摘要压缩旧对话的去留光靠裁剪有一个问题用户问我上回让你记的那件事呢如果那件事发生在很久之前已经被裁掉了机器人就失忆了。我需要一个折中方案——把旧消息压缩成摘要。当历史消息超过一定轮数或 token 数时触发摘要流程把最早的那批消息比如前 30 条取出来。用大模型生成一段话摘要指令是把以下对话压缩成简洁摘要保留所有关键事实、用户偏好、待办事项、已讨论过的结论。不要遗漏具体信息。把摘要作为一条特殊历史消息放在消息列表最前面。删除被压缩的原始消息。这样做的效果是机器人既不会丢失关键记忆又不会让上下文无限膨胀。摘要本身占的 token 远小于原文。4.4 上下文管理最容易犯的三个错第一个错是把所有历史不分大小一股脑塞进去。我刚开始就是这么干的跑了一周看账单发现 token 消耗是预期的三倍。后来加了 token 预算和裁剪成本立刻下来了。第二个错是只存用户消息、不存机器人回复。这会导致多轮对话完全错乱——大模型看到的是一连串用户提问没有自己的前序回答对话一致性无从谈起。我在 logger 里加了一个 debug 字段专门记录每次请求发出的消息条数和 token 数才发现这个问题。第三个错是忽略系统提示词和工具的 token 开销。我见过有人配置 max_tokens 只有 1024但系统提示词本身就写了 3000 token这样模型每轮回复都会被截断。系统提示词不是越详细越好够用就行能用 500 字讲清楚的事不要写 2000 字。5. 生产环境踩坑实录重试、并发、回环一个都不能少5.1 限流与 429指数退避背后的数学大模型 API 不是无限量的。每个账户有每分钟请求数限制和每分钟 token 数限制超出就返回 429。我第一个版本完全没有处理这个结果有次群聊突然活跃一秒钟进来十几条消息API 直接开始批量报错机器人处于半瘫状态。正确的做法是重试但不是立即重试而是指数退避加抖动。我之前对指数退避有个误解以为就是每次等待时间翻倍。实际上如果所有重试请求都在同一时刻发出会让服务器雪上加霜。所以要加一个随机抖动让每个请求的重试时间错开。import asyncio import random async def retry_with_backoff(coro_func, max_retries5, base_delay1.0): for attempt in range(max_retries): try: return await coro_func() except Exception as e: if attempt max_retries - 1: raise delay base_delay * (2 ** attempt) random.uniform(0, 0.5) await asyncio.sleep(delay) # unreachable return None注意重试只能针对幂等操作消息处理本身天然幂等因为我们已经按 message_id 去重了否则重试会导致重复执行。这也是为什么 5.2 的去重逻辑那么重要——两个机制是配套的。5.2 重复消息与幂等处理IM 平台的 Webhook 推送是带有重试机制的。如果你的服务返回了非 200 状态码或者响应超时平台会隔一段时间再推一次同一条消息。我当时遇到的现象是用户发一条消息机器人偶尔回两条。排查了很久最后发现是平台在那次请求时收到了一个超时响应自动重推了一模一样的消息而我的服务没有去重把同一条消息处理了两遍。解决方案是在入口处加一层消息 ID 去重class DedupPipeline: def __init__(self, redis): self.redis redis async def check_and_add(self, message_id: str) - bool: # 返回 False 表示这条消息已经处理过 result await self.redis.set( fdedup:{message_id}, 1, ex3600, nxTrue ) return bool(result)利用 Redis 的NX参数实现原子性的如果不存在才写入。如果写入成功说明是首次处理返回 True如果 Key 已存在说明是重复推送直接忽略。过期时间设一小时足够覆盖平台的重试窗口了。5.3 消息回环机器人自言自语一晚上这是我踩过最诡异的坑。某个周末晚上群里没什么人说话但 CloddsBot 自己一条接一条地发消息内容完全不知所云像和一个看不见的人吵架。我翻日志才发现原因Webhook 把机器人自己发出的消息也当成新消息推了回来。链路是这样的机器人回复了一条消息 → 平台的 Webhook 把这条回复当作新的入站消息推给机器人 → 机器人又回复它 → 又触发 Webhook……无限循环。如果一个人都没有平台那边可能因为某些原因把机器人的消息触发回调于是形成一个没有终点的消息回环。解决方案很简单在适配层过滤掉自己发送者的消息# adapter/base.py async def parse(self, raw: dict) - IncomingMessage | None: msg IncomingMessage.from_raw(raw) if msg.sender_id self.bot_user_id: return None # 自己的消息丢弃 return msg这个 filter 要在进入处理管道之前做越早越好。另外还要给回复也做一层标记确保即使平台把它回调回来也能被识别。5.4 长耗时请求与任务队列前面提到 Webhook 要尽快返回 200后台任务得异步跑。但异步不等于可靠。用asyncio.create_task跑后台任务有一个风险如果进程在任务执行过程中崩溃这个任务就丢了。我当时就遇到过用户发了一条消息机器人还没回复服务因为 OOM 重启了重试又因为消息已处理没法再次触发用户的这条消息就永远石沉大海。第一版我用的是内存队列加 create_task后来改成在 Redis 里维护一个待处理任务列表任务处理完成后删除启动时重新加载未完成任务。代码量不大但可靠性提升了一个档次。这里给个结论消息处理管道最好具备至少一次处理的语义配合去重机制保证最多一次的可见效果。两者组合才能在消息不丢的同时不重复。6. 提示词工程让角色设定稳定住、任务执行不跑偏6.1 系统提示词的结构化写法CloddsBot 的角色行为完全由系统提示词控制。早期我的提示词就一句话你是一个智能助手请帮助用户回答问题。结果机器人回复的格式千奇百怪有时候说作为一个人工智能……有时候又突然用列表把所有可能答案都列一遍。后来我把提示词改成了结构化写法分成固定几个段落你是 CloddsBot一个部署在聊天群里的智能助手。 # 性格与语气 - 简短、直接、不啰嗦日常回复不超过 5 句话 - 偶尔可以幽默但不要刻意讲冷笑话 - 不知道的事直接说不知道不要编造 # 能力范围 - 可以回答技术问题、常识问题、帮助梳理思路 - 可以调用工具查询天气、设置提醒 - 不要讨论你没有把握的主观评价 # 工具使用原则 - 用户问天气时必须调用 query_weather 工具 - 工具返回的结果要转成自然语言回复不要直接贴原始 JSON - 一次只执行一个必要的工具不要为了展示能力而多余调用 # 安全与边界 - 拒绝执行任何系统级命令 - 拒绝提供违法有害内容结构化提示词的好处是每次改某个维度的行为时不需要重写全部内容。比如我觉得机器人太啰嗦只需要调整性格与语气段落。而且分段模板方便做成配置文件不同场景群聊、私聊、客服可以加载不同的提示词模板。6.2 用输出格式约束替代自由发挥大模型自由发挥的空间越少行为越可控。我遇到过一个典型问题机器人回复内容正确但经常自己加一段如果你有其他问题随时问我哦群聊里刷屏感特别强。解决方案是在系统提示词里明确加了一条输出格式约束# 输出格式 - 回复不使用 Markdown 标题最多使用简单的加粗 - 不在回复末尾添加任何客套话 - 如果不需要工具调用直接输出最终文本不要输出中间推理过程这一条看似简单但它带来的改变是实实在在的机器人不再在每条回复后面加一句有什么我可以帮助你的吗。对于群聊场景来说这种克制比任何技术优化都重要。6.3 工具调用的意图识别调优Function Calling 的核心是将用户意图映射到工具参数。这个映射偶尔会出错。最典型的例子是用户问上海明天穿什么衣服模型应该先查天气但它可能因为穿衣服这个词直接开始给穿搭建议而不是调用query_weather(上海)。调优方法有两个方向。第一是在工具描述里写清楚触发条件描述越具体越好。query_weather的描述从查询天气改成当用户提到任何城市或地点的天气、温度、体感、穿衣建议时调用参数 city 是用户提到或暗示的城市名。第二是在系统提示词的工具使用原则里重复这些触发条件。两个地方都写了之后这类错误大幅减少。我还有一个习惯每次意图识别失败就把对话样本记录下来整理成测试集。改完提示词后跑一遍测试集看回归情况。这个习惯看起来笨但防止修复一个 bug 引入两个新 bug特别有效。7. 部署与运维systemd 托管、日志、无感重启7.1 进程托管与自动重启CloddsBot 是个常驻服务不能靠nohup python main.py 这种野路子跑。我用 systemd 做了一个 Unit 文件让它在崩溃后能自动拉起# /etc/systemd/system/cloddsbot.service [Unit] DescriptionCloddsBot Service Afternetwork-online.target redis.service [Service] Userdeploy WorkingDirectory/opt/cloddsbot EnvironmentFile/opt/cloddsbot/.env ExecStart/opt/cloddsbot/venv/bin/python main.py Restartalways RestartSec5 TimeoutStopSec30 [Install] WantedBymulti-user.targetRestartalways的意思是无论什么原因退出都自动重启。要小心一种情况代码启动时抛异常进程一直重复启动即崩systemd 会一直快速重启把系统日志刷爆。所以我加了RestartSec5每次重启至少间隔五秒并且给 main.py 加了启动日志方便排查。7.2 日志不要什么都 print要轮转要分级项目的日志策略是从一次事故中逼出来的。有一次机器人半夜掉线我想查日志结果发现所有输出都混在一个 nohup.out 里翻了几千行才找到报错原因。后来我把日志接入了 Python 标准库 logging按天轮转保留七天import logging from logging.handlers import TimedRotatingFileHandler handler TimedRotatingFileHandler( logs/cloddsbot.log, whenmidnight, backupCount7, encodingutf-8, ) logging.basicConfig( handlers[handler, logging.StreamHandler()], levellogging.INFO, format%(asctime)s %(levelname)s [%(name)s] %(message)s, )日志里除了普通运行信息我会在每次 LLM 调用后记录耗时和 token 数logger.info( llm_call session%s duration%.2fs prompt_tokens%d completion_tokens%d, chat_id, duration, resp.usage.input_tokens, resp.usage.output_tokens, )这样每天看日志就能掌握机器人的健康度。token 数如果异常增长一般意味着某个会话的上下文管理出了问题。7.3 健康检查与平滑重启我给 CloddsBot 加了一个极简的/healthz接口返回 JSON{ok: true, version: 0.3.0}。部署配置里用它做健康检查每 30 秒探一次连续几次失败就触发容器或进程重启。平滑重启这件事我用的最土但最可靠的方案systemd 的ExecReload重新加载进程靠前端负载均衡切换流量。因为机器人的量级远没到需要滚动更新的程度重启过程即使有三五秒的空白也没有用户感知得到。如果你要追求真正的零停机可以在 Webhook 入口加一层队列缓冲收到消息先落 Redis处理进程可以随时重启启动后从 Redis 拉未处理的消息继续跑。这个架构其实很简单但效果很好消息入口和处理进程解耦让重启从事故变成日常操作。7.4 部署流程一键更新脚本部署流程我固定成了一套脚本动作备份当前版本。拉取代码到新目录。安装依赖。跑一遍冒烟测试发一条ping消息看是否正常回复。切换 systemd 指向新版本。旧版本保留三天后删除。这套流程看起来笨拙但胜在稳定。第三方的 API SDK 偶尔会有破坏性升级代码拉下来直接跑可能因为某个依赖版本变化导致启动失败。冒烟测试这一步不能省我在部署上踩过的坑基本都是因为偷懒跳过了它。CloddsBot 从最初的想法到稳定运行前后花了大概两周的业余时间。回过头看最花时间的不是写代码而是反复调整上下文管理策略和排查那些只有在真实流量下才会暴露的问题。我个人的体会是聊天机器人这个方向单一功能都不复杂复杂的是把它们组合成一个在真实环境里稳定不掉链子的系统。每一个看似小的环节——去重、限流、上下文裁剪、消息回环——都值得单独设计。如果你也准备动手造一个自己的机器人我的建议是一个版本只加一个核心能力跑稳了再加下一个。别一上来就想着做成全平台通用的万能助手先让它在自己的群里被真正用起来比什么都重要。 从零构建智能聊天机器人：基于大模型API的CloddsBot实战解析
事情要从一个周五的下午说起。我待的技术群那天格外热闹。有人贴了一屏报错日志有人问中间件到底怎么配才对还有个人直接让我帮他写一条抓取特定格式的正则。我一边开会一边瞟群消息回着回着实在顾不过来脑子里冒出个念头要不弄个机器人顶上这个念头后来变成了CloddsBot。名字没什么高深含义——Clodds 是从 clouds 随手改出来的加了个 d意思是这机器人跑在云上、用着云端的模型干些云一样轻但覆盖面特别广的杂活。它本质是一个基于大模型 API 的智能聊天机器人服务通过 Webhook 接收消息做意图识别再决定是直接回答、查天气、记待办还是设提醒。这篇文章我把 CloddsBot 从需求拆解到部署运维完完整整讲一遍。适合谁看一个是想自己搞个聊天机器人但不知道从哪下手的人另一个是已经在用大模型 API 写应用、却被上下文管理和重试逻辑反复折磨过的人。我会把每个关键决策背后的为什么都讲清楚而不是只丢一段能跑的代码。1. 从群聊没人回到 CloddsBot这个机器人到底在解决什么问题1.1 需求是怎么长出来的项目启动之前我花了大概一个晚上整理需求。不整理不行聊天机器人这个方向看着简单实际一展开就是无底洞什么多轮对话、情感识别、语音合成、知识库问答都能往里塞。我给自己定的原则很朴素只解决那些每周都会发生的、重复度高的麻烦。我当时列了这么几类高频需求群里有人问这个报错啥意思其实把错误信息丢给大模型就有答案。隔三差五有人让帮忙查某个城市的天气一次两次无所谓次数多了真烦。我自己经常忘事需要有一个随手说一句提醒我下午三点开会就能记住的地方。群聊里经常插入一些和主题无关的闲聊如果我逐个回复一天的时间全没了。这四类需求对应到机器人能力上就是自由对话、天气查询、提醒备忘、自动处理重复性的文字劳动。注意这里没有语音多模态自动进群踢人这些花活不是它们不好而是我第一个版本不需要。1.2 现成方案为什么不够用决定自造轮子之前我其实是先看了一圈现成方案。老牌的 IRC 机器人和各种开源群管机器人有个通病逻辑靠关键词匹配规则写死了。你说天气 北京它能命中你说北京今天冷吗它就傻了。后来也试过接大模型套壳的开源项目但那些项目普遍为了演示效果把功能堆得特别满光配置文件就有几百行部署完我根本不知道它哪些逻辑是可用的、哪些是作者一拍脑袋写的。更关键的一点是现成方案解决不了这个机器人是我自己的延伸这件事。我希望它能按照我的语气说话记住和我相关的待办事项而不是每次对话都从零开始。这些东西不是靠配置能配出来的得自己改代码。1.3 边界比功能更重要CloddsBot 第一个版本的需求边界我是这么定的要做接收文本消息、多轮对话、天气查询、待办提醒、基础信息查询、按会话隔离上下文。不做不做用户画像、不做情感分析、不主动推送消息、不开放任意系统命令执行、不接语音和图片理解。这个边界在后来帮了大忙。比如有朋友问我为什么不做成 plug-in 架构我说第一版不需要——我需要的是把整条链路跑通而不是先把插件系统设计出花来。等项目跑了一段时间你会发现哪些功能是真的高频、哪些只是你以为的高频这时候再抽象插件系统才能抽象对地方。这是我做这个项目学到最实在的一课机器人的价值判断标准不是它能做多少事而是它在日常里被真正用起了几次。2. 技术栈与整体设计先把地基夯实再谈智能2.1 为什么是 Python asyncio aiohttpCloddsBot 的消息入口是一个 Webhook 服务也就是说 IM 平台收到消息后会通过 HTTP 请求往我这边推一条消息。这里有个天然的并发场景用户 A 的消息还在等大模型回复用户 B 的消息又进来了。所以我需要的是一个异步非阻塞的运行时而不是传统的阻塞式多线程。我选的是 Python 3.11 asyncio aiohttp。原因很直接我已经在用 Python 生态做大模型相关开发SDK 支持最成熟。aiohttp 做 Webhook 服务足够轻不需要为了一个消息入口引一个 FastAPI 全家桶当然你用 FastAPI 也没问题纯粹是取舍。异步编程在处理 HTTP 长连接时优势明显一个请求在等响应的时候事件循环可以继续处理别的请求。这里有一个常见的误判很多人以为并发高才需要异步。其实聊天机器人场景里你的瓶颈不是 QPS而是连接数和等待时间。就算一分钟只有 20 条消息每条消息要等大模型 5 到 30 秒如果同步写消息就得排长队体验极差。2.2 模块划分先画好边界再写代码CloddsBot 的目录结构长这样cloddsbot/ ├── main.py # 入口加载配置、装配模块、启动服务 ├── config.yaml # 配置文件 ├── requirements.txt ├── clodds/ │ ├── __init__.py │ ├── config.py # 配置加载与校验 │ ├── server.py # Webhook 服务 │ ├── adapter/ # 消息适配层 │ │ ├── base.py # 统一消息抽象 │ │ └── webhook.py # 具体平台的推送解析 │ ├── llm/ │ │ ├── client.py # 大模型 API 封装 │ │ └── prompts.py # 系统提示词 │ ├── memory/ │ │ └── session.py # 会话状态管理 │ ├── tools/ │ │ ├── registry.py # 工具注册表 │ │ ├── weather.py # 天气查询 │ │ └── reminder.py # 提醒备忘 │ └── scheduler/ │ └── runner.py # 定时任务调度每个模块只做一件事依赖方向是从上往下的server 调 adapter 和 llmllm 调 toolsmemory 被 llm 层调用谁都不许反向依赖。这个设计不是一上来就有的是我写到第三天发现 tools 和 adapter 开始互相 import 才抽出来的。2.3 配置管理的细节配置这块我踩过一次坑早期为了省事把 API Key 直接写在代码里结果有一次差点把仓库推到公开仓库去吓出一身冷汗。后来老老实实做了三件事配置全部集中在config.yaml用 yaml 文件管理。敏感信息API Key、Token一律从环境变量读取配置文件里只写变量名。准备config.example.yaml提交到仓库真实配置放在本地 gitignore。配置文件示例bot: name: CloddsBot version: 0.3.0 webhook: host: 0.0.0.0 port: 8080 path: /webhook/msg llm: provider: anthropic model: claude-sonnet-4-5 api_key_env: CLODDS_API_KEY temperature: 0.4 max_tokens: 1024 memory: backend: redis session_ttl_hours: 72 max_context_rounds: 12 scheduler: timezone: Asia/Shanghai notify_url: http://localhost:9090/notify模型名那里我特意留了注释按你账户实际开通的版本来填不同的版本能力差异、价格差异都很大。配置文件中一个容易被忽略的点是session_ttl_hours这个决定了一段对话能记多久。我设的是 72 小时超过之后会话自动清空避免 Redis 里的历史越攒越多。3. 消息链路的完整实现从 Webhook 到工具调用3.1 统一消息抽象层CloddsBot 的完整消息链路是IM 平台推送 → Webhook → 适配层解析 → 会话管理 → 大模型调用 → 工具调用如果需要→ 回复发送。这条链路里最重要的一环是抽象层因为它决定了未来你想接更多消息源时要不要重写整个项目。我定义了一个非常薄的消息模型# adapter/base.py from dataclasses import dataclass, field from datetime import datetime dataclass class IncomingMessage: message_id: str # 平台消息 ID用于幂等去重 chat_id: str # 会话 ID可以是群 ID 或用户 ID sender_id: str # 发送者 ID sender_name: str text: str raw: dict field(default_factorydict) property def is_group(self) - bool: return self.chat_id ! self.sender_id为什么要单独抽一个 IncomingMessage因为不同平台的消息结构五花八门有的字段叫from有的叫user_id有的消息里还带各种扩展字段。如果业务逻辑里到处直接操作原始消息体将来每接一个新平台就得到处改判断逻辑。统一成这个模型之后后面所有环节都只认 IncomingMessage。is_group这个属性是后来加的。群聊和单聊的处理策略很不一样群聊里机器人被 了才回复单聊里所有消息都要回。这个逻辑放在抽象层是因为各个平台判断是否被 到的方式不同但抽象后对所有上层模块透明。3.2 LLM 调用的封装LLM 调用是整个机器人最核心的环节。我用的 SDK 是anthropic异步版本# llm/client.py import os from anthropic import AsyncAnthropic class LLMClient: def __init__(self, config): self.api_key os.environ[config[llm][api_key_env]] self.model config[llm][model] self.temperature config[llm][temperature] self.max_tokens config[llm][max_tokens] self.client AsyncAnthropic(api_keyself.api_key) async def chat(self, system: str, messages: list, tools: list): resp await self.client.messages.create( modelself.model, systemsystem, messagesmessages, toolstools or [], temperatureself.temperature, max_tokensself.max_tokens, ) return resp这里有两个容易忽视的点。第一个是max_tokens。我一开始设的比较大2048后来看统计发现绝大多数回复根本用不到那么多 token还白白增加延迟和费用改成 1024 后日常场景完全够用。这个值应该根据你的实际场景去测而不是抄别人的配置。第二个是异常处理。API 调用在任何不可靠的网络上都有可能超时或返回 5xx这段代码的 try-except 我故意省略了但实际项目里是必须有的——后面有一节专门讲这个。3.3 工具注册表让机器人学会动手光会聊天的大模型只是个聊天框CloddsBot 真正有用的地方是它能调用工具。这里用到了 Function Calling在 Anthropic 的术语里叫 Tool Use。我先定义了一个工具注册表所有工具函数都通过装饰器注册# tools/registry.py import inspect from typing import Callable TOOL_SCHEMAS [] TOOL_FUNCTIONS {} def register_tool(schema: dict): def decorator(func: Callable): name schema[name] TOOL_SCHEMAS.append(schema) TOOL_FUNCTIONS[name] func return func return decorator def get_tools(): return TOOL_SCHEMAS async def execute_tool(name: str, arguments: dict): func TOOL_FUNCTIONS.get(name) if func is None: return {error: ftool {name} not found} return await func(**arguments)然后写具体的工具函数。比如天气查询# tools/weather.py import aiohttp from .registry import register_tool register_tool({ name: query_weather, description: 查询指定城市的当前天气、温度和风力情况, input_schema: { type: object, properties: { city: {type: string, description: 城市名称如 北京、上海、深圳} }, required: [city] } }) async def query_weather(city: str): # 这里对接一个公开的天气 API做参数校验和超时处理 async with aiohttp.ClientSession() as session: async with session.get( fhttps://example.com/api/weather, params{city: city}, timeoutaiohttp.ClientTimeout(total5) ) as resp: data await resp.json() return {city: city, weather: data.get(condition, 未知), temp: data.get(temp)}注册表的好处是新增工具时不需要改动调用链路的代码。写完一个函数加一个装饰器它就自动出现在传给大模型的 tools 列表里了。这个轻量方案比引入一个重量级框架更适合我现在这个阶段的项目。3.4 Tool Use 的闭环一次完整对话的旅程工具调用不是大模型自己偷偷调而是一个多回合的闭环。我画一下这个流程用户说北京今天天气怎么样Adapter 把消息转成 IncomingMessage。会话管理器取出该会话的历史消息加上系统提示词和工具列表一起发给大模型。大模型返回的不是最终文本而是一个tool_use块内容是{name: query_weather, input: {city: 北京}}。机器人执行query_weather(北京)拿到天气结果。把工具执行结果作为一条user角色消息回传给大模型。大模型看到工具结果生成最终回答北京现在是晴12 度风力三级。这个流程用代码写出来大概是# 伪代码简化版 tool_use 循环 from anthropic.types import Message async def process_message(session, user_text: str): messages session.get_history() [{role: user, content: user_text}] for _ in range(5): # 最大工具调用轮数防止死循环 resp: Message await llm.chat( systemsession.system_prompt, messagesmessages, toolsget_tools(), ) content_blocks resp.content tool_uses [b for b in content_blocks if b.type tool_use] if not tool_uses: # 没有工具调用直接返回文本给用户 text .join(b.text for b in content_blocks if b.type text) session.append(assistant, text) return text # 有工具调用先把 assistant 消息加入历史 messages.append({role: assistant, content: content_blocks}) # 执行每个工具把结果作为 user 消息接回去 for tool_use in tool_uses: result await execute_tool(tool_use.name, tool_use.input) messages.append({ role: user, content: [ { type: tool_result, tool_use_id: tool_use.id, content: str(result), } ], }) return 工具调用超限请稍后再试注意那个max 5的循环边界。没有这个边界如果大模型连续调用工具多次比如某个工具返回的数据又触发另一个工具理论上会无限循环。我第一次跑的时候就是没设边界结果机器人自己连续调用了一串工具虽然最后还是回到了正轨但白白浪费了几万个 token。3.5 响应如何回到用户手里最后一步是把生成的回复发送出去。我这里的做法是先把回复写入一个发送队列由专门的发送任务负责推送。为什么不直接发因为 Webhook 处理函数应该尽快返回告诉平台消息收到了否则平台会认为推送失败并反复重试。大模型生成可能要 10 秒Webhook 不可能等 10 秒再返回 200。所以正确的顺序是异步入队立即返回 200后台发送。# server.py简化 from aiohttp import web async def handle_message(request): raw await request.json() msg adapter.parse(raw) # 幂等去重同一个 message_id 只处理一次 if not await dedup_pipeline.check_and_add(msg.message_id): return web.Response(status200) # 入队异步处理 asyncio.create_task(processing_pipeline.run(msg)) return web.Response(status200)这个设计在后面排障那节还会再提到它引出了两个新问题一是消息去重二是后台任务的可靠性。4. 上下文管理CloddsBot 不失忆、不跑题的工程方案4.1 会话隔离与状态存储聊天机器人最大的坑之一就是上下文管理。刚开始我用一个全局列表存历史消息测试的时候没感觉一进群发现所有人都共享同一个记忆——A 说了自己的名字B 再问我叫什么机器人回的是 A 的名字。这就是典型的会话串线。解决办法是按chat_id做隔离。每个会话独立保存自己的历史消息列表。存储后端我选的是 Redis原因有两个一是天然支持 TTL我可以给每个会话设置 72 小时过期时间二是重启不丢数据不像内存 dict服务一崩所有历史全没了。Redis 里的 key 结构大概是session:{chat_id}:messages # 历史消息列表JSON 编码 session:{chat_id}:meta # 会话元信息比如上次活跃时间每次有新消息就把这条消息追加到列表尾部同时用 LPUSH/RPUSH 配合 LTRIM 控制列表长度。4.2 窗口滑动与 token 预算大模型的上下文窗口是有限的就算最新模型的窗口很大也不可能无限堆历史。原因不只是窗口大小还有两个实际约束成本每多一个 token 都要付钱和延迟上下文越长首字延迟越高这是实打实的用户体验问题。我采用的策略是给每个会话设一个 token 预算超了就从头裁剪。分成几部分上下文组成部分预算占比说明系统提示词15%角色设定、功能说明每次请求都带工具定义25%传给大模型的 tools schema对话历史50%最近 N 轮消息本轮输入10%用户当前这条消息怎么算 token最简单的方式是调用 API 前用len(text) // 2粗估因为中文在大多数 tokenizer 里大概是 1 到 1.5 个字节对应一个 token。更准确的做法是直接用 SDK 提供的 tokenizer 接口把消息内容逐个编码再求和。项目初期我用的是粗估法够用等要精细控制成本时才换成精确 tokenizer。4.3 摘要压缩旧对话的去留光靠裁剪有一个问题用户问我上回让你记的那件事呢如果那件事发生在很久之前已经被裁掉了机器人就失忆了。我需要一个折中方案——把旧消息压缩成摘要。当历史消息超过一定轮数或 token 数时触发摘要流程把最早的那批消息比如前 30 条取出来。用大模型生成一段话摘要指令是把以下对话压缩成简洁摘要保留所有关键事实、用户偏好、待办事项、已讨论过的结论。不要遗漏具体信息。把摘要作为一条特殊历史消息放在消息列表最前面。删除被压缩的原始消息。这样做的效果是机器人既不会丢失关键记忆又不会让上下文无限膨胀。摘要本身占的 token 远小于原文。4.4 上下文管理最容易犯的三个错第一个错是把所有历史不分大小一股脑塞进去。我刚开始就是这么干的跑了一周看账单发现 token 消耗是预期的三倍。后来加了 token 预算和裁剪成本立刻下来了。第二个错是只存用户消息、不存机器人回复。这会导致多轮对话完全错乱——大模型看到的是一连串用户提问没有自己的前序回答对话一致性无从谈起。我在 logger 里加了一个 debug 字段专门记录每次请求发出的消息条数和 token 数才发现这个问题。第三个错是忽略系统提示词和工具的 token 开销。我见过有人配置 max_tokens 只有 1024但系统提示词本身就写了 3000 token这样模型每轮回复都会被截断。系统提示词不是越详细越好够用就行能用 500 字讲清楚的事不要写 2000 字。5. 生产环境踩坑实录重试、并发、回环一个都不能少5.1 限流与 429指数退避背后的数学大模型 API 不是无限量的。每个账户有每分钟请求数限制和每分钟 token 数限制超出就返回 429。我第一个版本完全没有处理这个结果有次群聊突然活跃一秒钟进来十几条消息API 直接开始批量报错机器人处于半瘫状态。正确的做法是重试但不是立即重试而是指数退避加抖动。我之前对指数退避有个误解以为就是每次等待时间翻倍。实际上如果所有重试请求都在同一时刻发出会让服务器雪上加霜。所以要加一个随机抖动让每个请求的重试时间错开。import asyncio import random async def retry_with_backoff(coro_func, max_retries5, base_delay1.0): for attempt in range(max_retries): try: return await coro_func() except Exception as e: if attempt max_retries - 1: raise delay base_delay * (2 ** attempt) random.uniform(0, 0.5) await asyncio.sleep(delay) # unreachable return None注意重试只能针对幂等操作消息处理本身天然幂等因为我们已经按 message_id 去重了否则重试会导致重复执行。这也是为什么 5.2 的去重逻辑那么重要——两个机制是配套的。5.2 重复消息与幂等处理IM 平台的 Webhook 推送是带有重试机制的。如果你的服务返回了非 200 状态码或者响应超时平台会隔一段时间再推一次同一条消息。我当时遇到的现象是用户发一条消息机器人偶尔回两条。排查了很久最后发现是平台在那次请求时收到了一个超时响应自动重推了一模一样的消息而我的服务没有去重把同一条消息处理了两遍。解决方案是在入口处加一层消息 ID 去重class DedupPipeline: def __init__(self, redis): self.redis redis async def check_and_add(self, message_id: str) - bool: # 返回 False 表示这条消息已经处理过 result await self.redis.set( fdedup:{message_id}, 1, ex3600, nxTrue ) return bool(result)利用 Redis 的NX参数实现原子性的如果不存在才写入。如果写入成功说明是首次处理返回 True如果 Key 已存在说明是重复推送直接忽略。过期时间设一小时足够覆盖平台的重试窗口了。5.3 消息回环机器人自言自语一晚上这是我踩过最诡异的坑。某个周末晚上群里没什么人说话但 CloddsBot 自己一条接一条地发消息内容完全不知所云像和一个看不见的人吵架。我翻日志才发现原因Webhook 把机器人自己发出的消息也当成新消息推了回来。链路是这样的机器人回复了一条消息 → 平台的 Webhook 把这条回复当作新的入站消息推给机器人 → 机器人又回复它 → 又触发 Webhook……无限循环。如果一个人都没有平台那边可能因为某些原因把机器人的消息触发回调于是形成一个没有终点的消息回环。解决方案很简单在适配层过滤掉自己发送者的消息# adapter/base.py async def parse(self, raw: dict) - IncomingMessage | None: msg IncomingMessage.from_raw(raw) if msg.sender_id self.bot_user_id: return None # 自己的消息丢弃 return msg这个 filter 要在进入处理管道之前做越早越好。另外还要给回复也做一层标记确保即使平台把它回调回来也能被识别。5.4 长耗时请求与任务队列前面提到 Webhook 要尽快返回 200后台任务得异步跑。但异步不等于可靠。用asyncio.create_task跑后台任务有一个风险如果进程在任务执行过程中崩溃这个任务就丢了。我当时就遇到过用户发了一条消息机器人还没回复服务因为 OOM 重启了重试又因为消息已处理没法再次触发用户的这条消息就永远石沉大海。第一版我用的是内存队列加 create_task后来改成在 Redis 里维护一个待处理任务列表任务处理完成后删除启动时重新加载未完成任务。代码量不大但可靠性提升了一个档次。这里给个结论消息处理管道最好具备至少一次处理的语义配合去重机制保证最多一次的可见效果。两者组合才能在消息不丢的同时不重复。6. 提示词工程让角色设定稳定住、任务执行不跑偏6.1 系统提示词的结构化写法CloddsBot 的角色行为完全由系统提示词控制。早期我的提示词就一句话你是一个智能助手请帮助用户回答问题。结果机器人回复的格式千奇百怪有时候说作为一个人工智能……有时候又突然用列表把所有可能答案都列一遍。后来我把提示词改成了结构化写法分成固定几个段落你是 CloddsBot一个部署在聊天群里的智能助手。 # 性格与语气 - 简短、直接、不啰嗦日常回复不超过 5 句话 - 偶尔可以幽默但不要刻意讲冷笑话 - 不知道的事直接说不知道不要编造 # 能力范围 - 可以回答技术问题、常识问题、帮助梳理思路 - 可以调用工具查询天气、设置提醒 - 不要讨论你没有把握的主观评价 # 工具使用原则 - 用户问天气时必须调用 query_weather 工具 - 工具返回的结果要转成自然语言回复不要直接贴原始 JSON - 一次只执行一个必要的工具不要为了展示能力而多余调用 # 安全与边界 - 拒绝执行任何系统级命令 - 拒绝提供违法有害内容结构化提示词的好处是每次改某个维度的行为时不需要重写全部内容。比如我觉得机器人太啰嗦只需要调整性格与语气段落。而且分段模板方便做成配置文件不同场景群聊、私聊、客服可以加载不同的提示词模板。6.2 用输出格式约束替代自由发挥大模型自由发挥的空间越少行为越可控。我遇到过一个典型问题机器人回复内容正确但经常自己加一段如果你有其他问题随时问我哦群聊里刷屏感特别强。解决方案是在系统提示词里明确加了一条输出格式约束# 输出格式 - 回复不使用 Markdown 标题最多使用简单的加粗 - 不在回复末尾添加任何客套话 - 如果不需要工具调用直接输出最终文本不要输出中间推理过程这一条看似简单但它带来的改变是实实在在的机器人不再在每条回复后面加一句有什么我可以帮助你的吗。对于群聊场景来说这种克制比任何技术优化都重要。6.3 工具调用的意图识别调优Function Calling 的核心是将用户意图映射到工具参数。这个映射偶尔会出错。最典型的例子是用户问上海明天穿什么衣服模型应该先查天气但它可能因为穿衣服这个词直接开始给穿搭建议而不是调用query_weather(上海)。调优方法有两个方向。第一是在工具描述里写清楚触发条件描述越具体越好。query_weather的描述从查询天气改成当用户提到任何城市或地点的天气、温度、体感、穿衣建议时调用参数 city 是用户提到或暗示的城市名。第二是在系统提示词的工具使用原则里重复这些触发条件。两个地方都写了之后这类错误大幅减少。我还有一个习惯每次意图识别失败就把对话样本记录下来整理成测试集。改完提示词后跑一遍测试集看回归情况。这个习惯看起来笨但防止修复一个 bug 引入两个新 bug特别有效。7. 部署与运维systemd 托管、日志、无感重启7.1 进程托管与自动重启CloddsBot 是个常驻服务不能靠nohup python main.py 这种野路子跑。我用 systemd 做了一个 Unit 文件让它在崩溃后能自动拉起# /etc/systemd/system/cloddsbot.service [Unit] DescriptionCloddsBot Service Afternetwork-online.target redis.service [Service] Userdeploy WorkingDirectory/opt/cloddsbot EnvironmentFile/opt/cloddsbot/.env ExecStart/opt/cloddsbot/venv/bin/python main.py Restartalways RestartSec5 TimeoutStopSec30 [Install] WantedBymulti-user.targetRestartalways的意思是无论什么原因退出都自动重启。要小心一种情况代码启动时抛异常进程一直重复启动即崩systemd 会一直快速重启把系统日志刷爆。所以我加了RestartSec5每次重启至少间隔五秒并且给 main.py 加了启动日志方便排查。7.2 日志不要什么都 print要轮转要分级项目的日志策略是从一次事故中逼出来的。有一次机器人半夜掉线我想查日志结果发现所有输出都混在一个 nohup.out 里翻了几千行才找到报错原因。后来我把日志接入了 Python 标准库 logging按天轮转保留七天import logging from logging.handlers import TimedRotatingFileHandler handler TimedRotatingFileHandler( logs/cloddsbot.log, whenmidnight, backupCount7, encodingutf-8, ) logging.basicConfig( handlers[handler, logging.StreamHandler()], levellogging.INFO, format%(asctime)s %(levelname)s [%(name)s] %(message)s, )日志里除了普通运行信息我会在每次 LLM 调用后记录耗时和 token 数logger.info( llm_call session%s duration%.2fs prompt_tokens%d completion_tokens%d, chat_id, duration, resp.usage.input_tokens, resp.usage.output_tokens, )这样每天看日志就能掌握机器人的健康度。token 数如果异常增长一般意味着某个会话的上下文管理出了问题。7.3 健康检查与平滑重启我给 CloddsBot 加了一个极简的/healthz接口返回 JSON{ok: true, version: 0.3.0}。部署配置里用它做健康检查每 30 秒探一次连续几次失败就触发容器或进程重启。平滑重启这件事我用的最土但最可靠的方案systemd 的ExecReload重新加载进程靠前端负载均衡切换流量。因为机器人的量级远没到需要滚动更新的程度重启过程即使有三五秒的空白也没有用户感知得到。如果你要追求真正的零停机可以在 Webhook 入口加一层队列缓冲收到消息先落 Redis处理进程可以随时重启启动后从 Redis 拉未处理的消息继续跑。这个架构其实很简单但效果很好消息入口和处理进程解耦让重启从事故变成日常操作。7.4 部署流程一键更新脚本部署流程我固定成了一套脚本动作备份当前版本。拉取代码到新目录。安装依赖。跑一遍冒烟测试发一条ping消息看是否正常回复。切换 systemd 指向新版本。旧版本保留三天后删除。这套流程看起来笨拙但胜在稳定。第三方的 API SDK 偶尔会有破坏性升级代码拉下来直接跑可能因为某个依赖版本变化导致启动失败。冒烟测试这一步不能省我在部署上踩过的坑基本都是因为偷懒跳过了它。CloddsBot 从最初的想法到稳定运行前后花了大概两周的业余时间。回过头看最花时间的不是写代码而是反复调整上下文管理策略和排查那些只有在真实流量下才会暴露的问题。我个人的体会是聊天机器人这个方向单一功能都不复杂复杂的是把它们组合成一个在真实环境里稳定不掉链子的系统。每一个看似小的环节——去重、限流、上下文裁剪、消息回环——都值得单独设计。如果你也准备动手造一个自己的机器人我的建议是一个版本只加一个核心能力跑稳了再加下一个。别一上来就想着做成全平台通用的万能助手先让它在自己的群里被真正用起来比什么都重要。