1. 为什么我要自己动手写一个 Hermes 式智能体市面上现成的对话机器人框架其实不少，但真正用起来总会遇到几个绕不开的坎：要么是框架太重，光依赖就装了几百兆，启动一次要等半分钟；要么是扩展性差，想接一个自定义的工具函数得改好几层源码；要么是消息通道绑定得太死，想从命令行切到飞书这种团队协作平台，几乎等于重写一遍。我最初的需求其实很简单：团队日常在飞书里沟通，很多重复性的查询、整理、提醒类工作希望能有一个机器人来分担。但我不想要那种"什么都帮你做好"的黑盒方案，因为一旦出问题，排查成本极高。我更倾向于自己掌控核心逻辑，用尽量少的代码把"消息进来、意图理解、工具调用、结果返回"这条链路跑通。Hermes 式智能体的核心思路其实不复杂——它本质上是一个带工具调用能力的对话循环。用户发一条消息，模型判断是否需要调用外部工具，如果需要就执行工具、把结果喂回模型，再由模型生成最终回复。这个循环可以跑一轮，也可以跑多轮，直到模型认为任务完成。听起来简单，但真正落地时，消息通道的接入、工具注册的灵活性、上下文的管理，每一个环节都有讲究。我最终用大约 100 行核心代码实现了这套逻辑，并且成功接入了飞书。这里的"100 行"指的是智能体核心循环加工具调度的部分，不包括飞书 SDK 的适配层和配置文件。下面我会把这套方案的完整思路拆开讲清楚，包括为什么这样设计、每一步的关键细节、以及我在实测中踩过的坑。这篇文章适合两类人看：一类是想理解智能体底层运行机制、不想被框架绑架的开发者；另一类是想在飞书里快速落地一个实用机器人、但不想引入过重依赖的团队。无论你之前有没有接触过智能体开发，跟着思路走一遍，应该都能自己复现出来。2. 拆解 Hermes 式智能体的最小运行闭环2.1 一次完整的对话循环到底经历了什么很多人第一次接触智能体，容易被"Agent"这个词吓到，觉得里面有什么高深的调度算法。其实剥开来看，核心就是一个 while 循环加上几次模型调用。我用一个具体的例子来说明。假设用户在飞书里发了一句"帮我查一下今天有哪些待办任务"。这条消息进入系统后，会经历以下步骤：第一步，消息被封装成模型能理解的格式。通常是一个消息列表，包含系统提示词和用户消息。系统提示词里会写清楚"你是一个助手，可以调用以下工具"，并把可用工具的描述一并塞进去。第二步，模型返回结果。这时候模型有两种可能：一种是直接给出文字回复，说明它不需要调用工具；另一种是返回一个工具调用请求，告诉系统"我要调用 query_tasks 这个工具，参数是 date=today"。第三步，如果是工具调用请求，系统就执行对应的函数，拿到结果。比如 query_tasks 返回了一个任务列表。第四步，把工具执行结果作为一条新消息追加到对话历史里，再次调用模型。这次模型拿到工具结果后，通常会生成一段自然语言回复，比如"今天你有三项待办：……"。第五步，把最终回复发回飞书。整个循环的关键在于：模型负责决策，代码负责执行。模型不直接操作任何外部系统，它只输出"我想调用什么工具、传什么参数"，真正的执行权在代码手里。这个边界划清楚之后，安全性和可控性都有了保障。2.2 为什么选择"工具调用"而不是"意图分类"早期我做对话机器人的时候，用的是意图分类的路子：先训练一个分类模型，把用户消息分到预定义的意图类别里，再根据类别走不同的处理分支。这套方案在意图数量少、表达方式固定的场景下能用，但一旦用户换种说法，分类就崩了。工具调用模式最大的好处是意图理解交给大模型。我不需要预先定义"查任务"这个意图有哪些说法，模型自己能从自然语言里提取出结构化参数。用户说"今天有啥要做的"、"帮我看看待办"、"今天的任务列表"，模型都能映射到同一个工具调用上。另一个好处是多工具组合。用户可能说"查一下今天的任务，然后帮我提醒还没完成的"。这句话涉及两个动作：查询和提醒。在意图分类模式下，我得专门设计一个复合意图；但在工具调用模式下，模型会先调用查询工具，拿到结果后再调用提醒工具，天然支持组合。当然，工具调用也不是没有代价。它依赖模型的能力，如果模型本身对工具描述理解不到位，可能会调用错误的工具或者传错参数。所以工具的描述写得清不清楚，直接决定了整套系统的可靠性。这一点我在后面会专门讲。2.3 核心循环的代码骨架长什么样下面是我实际使用的核心循环，用 Python 写的，去掉空行和注释大概 40 行左右。我把它拆成几个关键部分来说明。def run_agent(user_message, history, tools, max_turns=5): history.append({"role": "user", "content": user_message}) for turn in range(max_turns): response = call_llm(history, tools) msg = response.choices[0].message if not msg.tool_calls: history.append({"role": "assistant", "content": msg.content}) return msg.content history.append(msg) for tool_call in msg.tool_calls: name = tool_call.function.name args = json.loads(tool_call.function.arguments) result = execute_tool(name, args, tools) history.append({ "role": "tool", "tool_call_id": tool_call.id, "content": str(result) }) return "任务处理轮次超出限制，请简化你的请求。"这段代码里有几个设计决策值得展开说。max_turns 的限制。我设了 5 轮上限，防止模型陷入死循环。实际测试中，绝大多数请求 1 到 2 轮就能完成，3 轮以上的情况非常少见。如果没有这个上限，遇到模型反复调用同一个工具的情况，会一直烧 token。history 的传递方式。我把整个对话历史传进去，而不是只传当前消息。这样做的好处是模型能看到之前的上下文，支持多轮对话。但代价是 token 消耗会随对话轮次增长。生产环境里需要加一个截断策略，比如只保留最近 10 轮。工具结果的序列化。str(result)这个处理比较粗糙，实际使用中建议用 JSON 序列化，并且控制返回内容的长度。如果工具返回了一大段文本，直接塞回模型会占用大量 token，还可能干扰模型判断。2.4 工具注册表的设计：让扩展变得无痛工具注册这块，我用了一个字典来管理，每个工具包含名称、描述、参数 schema 和实际执行函数。TOOLS = { "query_tasks": { "description": "查询指定日期的待办任务", "parameters": { "type": "object", "properties": { "date": {"type": "string", "description": "日期，格式 YYYY-MM-DD，默认今天"} } }, "handler": query_tasks_impl }, "send_reminder": { "description": "给指定用户发送提醒消息", "parameters": { "type": "object", "properties": { "user": {"type": "string"}, "content": {"type": "string"} }, "required": ["user", "content"] }, "handler": send_reminder_impl } }新增一个工具只需要往这个字典里加一项，不需要改核心循环的任何代码。这是我觉得这套设计最舒服的地方——核心逻辑稳定，扩展点集中在一处。工具描述的质量直接决定模型调用的准确率。我踩过的坑是：描述写得太简略，模型经常把参数传错。比如 query_tasks 的 date 参数，如果只写"日期"，模型有时会传"今天"这种自然语言，而不是标准格式。后来我在描述里明确写了格式要求，并在 handler 里做了容错解析，问题才解决。3. 接入飞书：消息通道适配的关键细节3.1 飞书机器人有两种接入方式，选哪种飞书开放平台提供两种机器人接入模式：一种是Webhook 机器人，只能被动接收消息、主动发送消息，但无法读取群聊历史、无法获取用户详细信息；另一种是应用机器人，需要创建企业自建应用，配置权限后可以调用更多 API。我选的是应用机器人，原因是 Webhook 模式的能力太受限。比如我想让机器人读取用户发的图片、或者查询群成员列表，Webhook 都做不到。应用机器人虽然配置步骤多一些，但一次配好之后，后续扩展空间大很多。配置应用机器人的核心步骤是：在开放平台创建应用、开启机器人能力、配置事件订阅地址、申请所需权限。事件订阅地址需要是一个公网可访问的 HTTPS 接口，飞书会把用户消息推送到这个地址。3.2 事件订阅的验证与消息解密飞书的事件推送有一个验证机制：配置订阅地址时，飞书会先发一个验证请求，你需要原样返回 challenge 字段的值，才算验证通过。这一步很多人会卡住，因为验证请求的格式和普通消息不一样。def handle_event(body): if "challenge" in body: return {"challenge": body["challenge"]} event = body.get("event", {}) msg_type = event.get("message", {}).get("message_type") if msg_type != "text": return {"code": 0} content = json.loads(event["message"]["content"]) user_text = content.get("text", "").strip() user_id = event["sender"]["sender_id"]["open_id"] reply = run_agent(user_text, get_history(user_id), TOOLS) send_message(user_id, reply) return {"code": 0}这段代码里有两个容易忽略的点。消息类型过滤。飞书推送的事件里，message_type 可能是 text、image、file 等多种类型。如果不做过滤，遇到非文本消息时解析 content 会报错。我一开始没加这个判断，结果用户发了一张图片，整个服务直接抛异常。open_id 的使用。飞书用户的标识有 open_id、union_id、user_id 三种。open_id 是应用维度的，同一个用户在不同应用里 open_id 不同。用 open_id 做会话隔离是最稳妥的，不会跨应用串数据。3.3 消息发送的限流与重试飞书的消息发送 API 有频率限制，单个应用每秒能发的消息数量有限。如果短时间内大量发送，会被限流返回错误码。我的处理方式是在发送层加一个简单的队列和重试机制。消息先入队列，由一个单独的工作线程按固定速率消费，遇到限流错误就等待一段时间后重试。这样即使智能体短时间内产生多条回复，也不会触发限流。import time from collections import deque send_queue = deque() def send_worker(): while True: if not send_queue: time.sleep(0.1) continue user_id, text = send_queue.popleft() try: send_message_api(user_id, text) except RateLimitError: send_queue.appendleft((user_id, text)) time.sleep(1) time.sleep(0.2)这个 worker 的消费速率我设的是每秒 5 条，实测下来比较稳。如果你的应用消息量更大，需要根据实际配额调整。3.4 会话上下文的隔离与持久化每个用户的对话历史需要单独维护，不能混在一起。我用一个字典按 open_id 存储历史，简单直接。user_histories = {} def get_history(user_id): if user_id not in user_histories: user_histories[user_id] = [{"role": "system", "content": SYSTEM_PROMPT}] return user_histories[user_id]但这个方案有个明显问题：服务重启后历史全丢。生产环境需要把历史存到 Redis 或数据库里。我目前的处理是内存存储加定期落盘，重启时从文件恢复。对于小规模使用够用了，但如果要支撑更多用户，还是得上 Redis。另外，历史不能无限增长。我的策略是保留最近 20 条消息，超出部分从头部丢弃。系统提示词始终保留在第一位，不参与截断。4. 工具函数的实现与踩坑记录4.1 工具函数的返回值格式很关键工具函数返回什么，直接影响模型能不能正确理解结果。我最初的做法是直接返回 Python 对象，让 str() 去转换。结果模型经常看不懂，比如返回一个字典{'tasks': ['写周报', '开会']}，模型有时会把它当成一段乱码。后来我改成返回结构化的自然语言描述：def query_tasks_impl(date=None): tasks = fetch_tasks_from_db(date) if not tasks: return f"{date} 没有待办任务。" lines = [f"{i+1}. {t['title']}（截止 {t['deadline']}）" for i, t in enumerate(tasks)] return f"{date} 共有 {len(tasks)} 项待办：\n" + "\n".join(lines)这样模型拿到结果后，几乎不需要额外加工就能生成回复。工具函数的输出应该尽量接近最终展示给用户的形式，把格式化的负担放在工具层，而不是让模型去猜。4.2 参数校验和容错处理模型传参数不可能 100% 准确。我遇到过几种典型情况：日期格式不对、必填参数缺失、参数类型错误。如果工具函数不做校验，直接抛异常，整个对话就中断了。我的做法是在每个工具函数入口做一层校验，校验失败时返回一个友好的错误提示，而不是抛异常。这样模型能看到错误信息，有机会在下一轮修正参数。def query_tasks_impl(date=None): if date is None: date = datetime.now().strftime("%Y-%m-%d") try: datetime.strptime(date, "%Y-%m-%d") except ValueError: return f"日期格式不正确：{date}，请使用 YYYY-MM-DD 格式。" ...这个错误提示会作为工具结果返回给模型，模型看到后通常会重新调用一次，传正确的格式。实测下来，这种自我修正的成功率挺高的。4.3 耗时工具的超时控制有些工具函数可能耗时较长，比如调用外部 API 查询数据。如果工具执行卡住，整个对话循环就阻塞了。我一开始没做超时控制，结果有一次外部服务响应慢，用户等了半分钟才收到回复。后来我给每个工具函数加了超时装饰器，超过 5 秒就中断并返回超时提示。import signal def timeout_handler(signum, frame): raise TimeoutError("工具执行超时") def with_timeout(seconds=5): def decorator(func): def wrapper(*args, **kwargs): signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(seconds) try: return func(*args, **kwargs) finally: signal.alarm(0) return wrapper return decorator需要注意的是，signal 方案只在主线程有效。如果你的工具函数跑在子线程里，得用其他方式实现超时，比如用 concurrent.futures 的 timeout 参数。4.4 工具权限的边界控制不是所有用户都能调用所有工具。比如"发送全员提醒"这种工具，应该只对管理员开放。我在工具注册表里加了一个权限字段，执行前先检查用户身份。TOOLS = { "send_broadcast": { "description": "发送全员广播", "permission": "admin", ... } } def execute_tool(name, args, tools, user_role="member"): tool = tools.get(name) if not tool: return f"未知工具：{name}" required = tool.get("permission") if required and required != user_role: return f"你没有权限调用 {name}。" return tool["handler"](**args)这个权限检查放在 execute_tool 里，而不是放在每个工具函数内部，好处是统一管理，不会遗漏。用户角色从飞书事件里获取，或者维护一张用户角色表。5. 实测中的性能表现与优化方向5.1 响应延迟的构成分析用户从发消息到收到回复，中间经历了几段延迟：飞书推送事件到服务器（通常 100 毫秒以内）、模型调用（1 到 3 秒，取决于模型和网络）、工具执行（几十毫秒到几秒不等）、消息发送回飞书（100 毫秒左右）。大头在模型调用上。如果一轮对话涉及两次模型调用（一次决策、一次生成回复），总延迟就是 2 到 6 秒。这个延迟在聊天场景里可以接受，但如果用户期望秒回，就需要优化。我的优化手段是：对于明显不需要工具调用的消息，先用一个轻量规则判断，直接走模型生成回复，省掉一轮调用。比如用户说"你好"，没必要走工具决策流程。这个规则很简单，就是维护一个问候语列表，命中就直接回复。5.2 Token 消耗的控制策略Token 消耗主要来自三部分：系统提示词、对话历史、工具描述。系统提示词和工具描述是固定开销，每次调用都要带上。对话历史随轮次增长。我的控制策略是：系统提示词尽量精简，只保留必要的行为约束；工具描述只写关键信息，不写废话；对话历史保留最近 10 轮，超出截断。这样单次调用的 token 量能控制在 2000 以内，成本可控。如果工具数量很多，工具描述本身就会占用大量 token。这时候可以考虑工具分组，根据用户消息先判断属于哪个组，只加载该组的工具描述。不过这会增加一次判断调用，需要权衡。5.3 并发场景下的状态管理如果多个用户同时发消息，每个用户的对话循环是独立的，互不干扰。但共享的资源比如数据库连接、HTTP 客户端，需要做好并发控制。我用的是连接池加线程池的方案。每个用户请求分配一个线程，线程内完成整个对话循环。数据库连接从池里取，用完归还。这样即使几十个用户同时发消息，也不会出现资源竞争。需要注意的是，Python 的 GIL 限制了多线程的并行能力。如果并发量真的很大，需要考虑异步方案或者多进程部署。不过对于团队内部使用，几十个并发已经绰绰有余了。5.4 日志与可观测性智能体出问题时，排查起来比普通服务麻烦，因为涉及模型决策这个黑盒。我的做法是把每次对话的完整链路都记下来：用户消息、模型返回、工具调用、工具结果、最终回复。这样出问题时能完整复现。def log_turn(user_id, turn, msg, tool_results=None): entry = { "user_id": user_id, "turn": turn, "timestamp": time.time(), "assistant_message": msg.content if hasattr(msg, "content") else None, "tool_calls": [tc.function.name for tc in msg.tool_calls] if msg.tool_calls else [], "tool_results": tool_results } logger.info(json.dumps(entry, ensure_ascii=False))日志按天切分，保留 7 天。有了这些日志，模型调用错误、工具执行异常、参数传递问题，都能快速定位。6. 从 100 行到生产可用还差什么6.1 错误恢复与降级策略核心循环跑通只是第一步，生产环境要考虑各种异常情况。模型 API 超时怎么办？工具执行失败怎么办？飞书推送重复怎么办？我的处理是分层降级。模型调用失败时，重试两次，仍失败则返回"服务暂时不可用，请稍后再试"。工具执行失败时，把错误信息返回给模型，让模型决定是重试还是告知用户。飞书推送重复时，用消息 ID 做去重，同一条消息只处理一次。这些逻辑加起来大概又多了 50 行代码，但都是必要的。没有这些保护，服务在真实环境里跑不了几天就会出问题。6.2 配置与密钥管理模型 API Key、飞书应用密钥这些敏感信息，绝对不能硬编码在代码里。我用的是环境变量加配置文件的方式，敏感信息只放环境变量，配置文件里只放非敏感的配置项。import os CONFIG = { "llm_api_key": os.environ["LLM_API_KEY"], "llm_base_url": os.environ.get("LLM_BASE_URL", "https://api.example.com/v1"), "feishu_app_id": os.environ["FEISHU_APP_ID"], "feishu_app_secret": os.environ["FEISHU_APP_SECRET"], "max_turns": 5, "history_limit": 10 }这样代码可以放心提交到版本库，部署时通过环境变量注入密钥。如果团队有密钥管理服务，也可以对接上去，实现自动轮换。6.3 部署方式的选择我试过两种部署方式：一种是直接跑在一台常驻服务器上，用 systemd 管理进程；另一种是打包成容器，用容器编排平台管理。小规模使用，systemd 就够了，简单直接，出问题看日志方便。如果要多实例部署、自动扩缩容，那就得上容器方案。我目前的场景是团队内部使用，systemd 完全够用。部署时要注意的是：飞书事件订阅地址必须是 HTTPS，所以需要配置证书。我用的是反向代理加自动证书续期，这部分配置一次就好，后续不用管。6.4 后续可以扩展的方向这套 100 行的核心代码，扩展空间其实很大。我目前想到几个方向：一是增加更多工具，比如日历查询、文档检索、数据统计；二是支持多模态输入，让机器人能处理图片和文件；三是引入记忆机制，让机器人记住用户的偏好和历史行为。但扩展的前提是核心循环保持稳定。我的原则是：核心循环尽量不改，所有新能力都通过工具的形式接入。这样每加一个功能，风险都局限在那个工具函数里，不会影响整体稳定性。7. 一些实操心得和避坑建议工具描述要写得像给新人看的文档。模型对工具的理解完全依赖描述，描述里该写清楚的格式、范围、默认值，一个都不能省。我见过太多人工具描述写得含糊，然后抱怨模型调用不准。对话历史不要无限增长。我一开始图省事，把所有历史都留着，结果跑了几十轮之后，token 消耗暴涨，响应也变慢。后来加了截断策略，体验立刻好转。错误信息要返回给模型，而不是直接抛给用户。模型看到错误信息后，有很大概率能自我修正。直接抛给用户，用户也不知道该怎么办。权限检查要放在工具执行层，不要依赖模型判断。模型可能会被诱导调用不该调用的工具，代码层面的权限检查是最后一道防线。日志要记全，尤其是模型返回的原始内容。排查问题时，模型到底返回了什么，是定位问题的关键。只记最终回复是不够的。飞书的消息推送有重试机制，同一条消息可能推送多次。一定要做去重，否则用户会收到重复回复。去重可以用消息 ID，也可以用时间窗口加内容哈希。最后说一个我踩过的坑：飞书的事件订阅地址验证，challenge 字段必须原样返回，不能做任何加工。我一开始以为要加密或者编码，折腾了半天才发现直接返回就行。这种细节文档里写得清楚，但容易看漏。
1. 为什么我要自己动手写一个 Hermes 式智能体市面上现成的对话机器人框架其实不少，但真正用起来总会遇到几个绕不开的坎：要么是框架太重，光依赖就装了几百兆，启动一次要等半分钟；要么是扩展性差，想接一个自定义的工具函数得改好几层源码；要么是消息通道绑定得太死，想从命令行切到飞书这种团队协作平台，几乎等于重写一遍。我最初的需求其实很简单：团队日常在飞书里沟通，很多重复性的查询、整理、提醒类工作希望能有一个机器人来分担。但我不想要那种"什么都帮你做好"的黑盒方案，因为一旦出问题，排查成本极高。我更倾向于自己掌控核心逻辑，用尽量少的代码把"消息进来、意图理解、工具调用、结果返回"这条链路跑通。Hermes 式智能体的核心思路其实不复杂——它本质上是一个带工具调用能力的对话循环。用户发一条消息，模型判断是否需要调用外部工具，如果需要就执行工具、把结果喂回模型，再由模型生成最终回复。这个循环可以跑一轮，也可以跑多轮，直到模型认为任务完成。听起来简单，但真正落地时，消息通道的接入、工具注册的灵活性、上下文的管理，每一个环节都有讲究。我最终用大约 100 行核心代码实现了这套逻辑，并且成功接入了飞书。这里的"100 行"指的是智能体核心循环加工具调度的部分，不包括飞书 SDK 的适配层和配置文件。下面我会把这套方案的完整思路拆开讲清楚，包括为什么这样设计、每一步的关键细节、以及我在实测中踩过的坑。这篇文章适合两类人看：一类是想理解智能体底层运行机制、不想被框架绑架的开发者；另一类是想在飞书里快速落地一个实用机器人、但不想引入过重依赖的团队。无论你之前有没有接触过智能体开发，跟着思路走一遍，应该都能自己复现出来。2. 拆解 Hermes 式智能体的最小运行闭环2.1 一次完整的对话循环到底经历了什么很多人第一次接触智能体，容易被"Agent"这个词吓到，觉得里面有什么高深的调度算法。其实剥开来看，核心就是一个 while 循环加上几次模型调用。我用一个具体的例子来说明。假设用户在飞书里发了一句"帮我查一下今天有哪些待办任务"。这条消息进入系统后，会经历以下步骤：第一步，消息被封装成模型能理解的格式。通常是一个消息列表，包含系统提示词和用户消息。系统提示词里会写清楚"你是一个助手，可以调用以下工具"，并把可用工具的描述一并塞进去。第二步，模型返回结果。这时候模型有两种可能：一种是直接给出文字回复，说明它不需要调用工具；另一种是返回一个工具调用请求，告诉系统"我要调用 query_tasks 这个工具，参数是 date=today"。第三步，如果是工具调用请求，系统就执行对应的函数，拿到结果。比如 query_tasks 返回了一个任务列表。第四步，把工具执行结果作为一条新消息追加到对话历史里，再次调用模型。这次模型拿到工具结果后，通常会生成一段自然语言回复，比如"今天你有三项待办：……"。第五步，把最终回复发回飞书。整个循环的关键在于：模型负责决策，代码负责执行。模型不直接操作任何外部系统，它只输出"我想调用什么工具、传什么参数"，真正的执行权在代码手里。这个边界划清楚之后，安全性和可控性都有了保障。2.2 为什么选择"工具调用"而不是"意图分类"早期我做对话机器人的时候，用的是意图分类的路子：先训练一个分类模型，把用户消息分到预定义的意图类别里，再根据类别走不同的处理分支。这套方案在意图数量少、表达方式固定的场景下能用，但一旦用户换种说法，分类就崩了。工具调用模式最大的好处是意图理解交给大模型。我不需要预先定义"查任务"这个意图有哪些说法，模型自己能从自然语言里提取出结构化参数。用户说"今天有啥要做的"、"帮我看看待办"、"今天的任务列表"，模型都能映射到同一个工具调用上。另一个好处是多工具组合。用户可能说"查一下今天的任务，然后帮我提醒还没完成的"。这句话涉及两个动作：查询和提醒。在意图分类模式下，我得专门设计一个复合意图；但在工具调用模式下，模型会先调用查询工具，拿到结果后再调用提醒工具，天然支持组合。当然，工具调用也不是没有代价。它依赖模型的能力，如果模型本身对工具描述理解不到位，可能会调用错误的工具或者传错参数。所以工具的描述写得清不清楚，直接决定了整套系统的可靠性。这一点我在后面会专门讲。2.3 核心循环的代码骨架长什么样下面是我实际使用的核心循环，用 Python 写的，去掉空行和注释大概 40 行左右。我把它拆成几个关键部分来说明。def run_agent(user_message, history, tools, max_turns=5): history.append({"role": "user", "content": user_message}) for turn in range(max_turns): response = call_llm(history, tools) msg = response.choices[0].message if not msg.tool_calls: history.append({"role": "assistant", "content": msg.content}) return msg.content history.append(msg) for tool_call in msg.tool_calls: name = tool_call.function.name args = json.loads(tool_call.function.arguments) result = execute_tool(name, args, tools) history.append({ "role": "tool", "tool_call_id": tool_call.id, "content": str(result) }) return "任务处理轮次超出限制，请简化你的请求。"这段代码里有几个设计决策值得展开说。max_turns 的限制。我设了 5 轮上限，防止模型陷入死循环。实际测试中，绝大多数请求 1 到 2 轮就能完成，3 轮以上的情况非常少见。如果没有这个上限，遇到模型反复调用同一个工具的情况，会一直烧 token。history 的传递方式。我把整个对话历史传进去，而不是只传当前消息。这样做的好处是模型能看到之前的上下文，支持多轮对话。但代价是 token 消耗会随对话轮次增长。生产环境里需要加一个截断策略，比如只保留最近 10 轮。工具结果的序列化。str(result)这个处理比较粗糙，实际使用中建议用 JSON 序列化，并且控制返回内容的长度。如果工具返回了一大段文本，直接塞回模型会占用大量 token，还可能干扰模型判断。2.4 工具注册表的设计：让扩展变得无痛工具注册这块，我用了一个字典来管理，每个工具包含名称、描述、参数 schema 和实际执行函数。TOOLS = { "query_tasks": { "description": "查询指定日期的待办任务", "parameters": { "type": "object", "properties": { "date": {"type": "string", "description": "日期，格式 YYYY-MM-DD，默认今天"} } }, "handler": query_tasks_impl }, "send_reminder": { "description": "给指定用户发送提醒消息", "parameters": { "type": "object", "properties": { "user": {"type": "string"}, "content": {"type": "string"} }, "required": ["user", "content"] }, "handler": send_reminder_impl } }新增一个工具只需要往这个字典里加一项，不需要改核心循环的任何代码。这是我觉得这套设计最舒服的地方——核心逻辑稳定，扩展点集中在一处。工具描述的质量直接决定模型调用的准确率。我踩过的坑是：描述写得太简略，模型经常把参数传错。比如 query_tasks 的 date 参数，如果只写"日期"，模型有时会传"今天"这种自然语言，而不是标准格式。后来我在描述里明确写了格式要求，并在 handler 里做了容错解析，问题才解决。3. 接入飞书：消息通道适配的关键细节3.1 飞书机器人有两种接入方式，选哪种飞书开放平台提供两种机器人接入模式：一种是Webhook 机器人，只能被动接收消息、主动发送消息，但无法读取群聊历史、无法获取用户详细信息；另一种是应用机器人，需要创建企业自建应用，配置权限后可以调用更多 API。我选的是应用机器人，原因是 Webhook 模式的能力太受限。比如我想让机器人读取用户发的图片、或者查询群成员列表，Webhook 都做不到。应用机器人虽然配置步骤多一些，但一次配好之后，后续扩展空间大很多。配置应用机器人的核心步骤是：在开放平台创建应用、开启机器人能力、配置事件订阅地址、申请所需权限。事件订阅地址需要是一个公网可访问的 HTTPS 接口，飞书会把用户消息推送到这个地址。3.2 事件订阅的验证与消息解密飞书的事件推送有一个验证机制：配置订阅地址时，飞书会先发一个验证请求，你需要原样返回 challenge 字段的值，才算验证通过。这一步很多人会卡住，因为验证请求的格式和普通消息不一样。def handle_event(body): if "challenge" in body: return {"challenge": body["challenge"]} event = body.get("event", {}) msg_type = event.get("message", {}).get("message_type") if msg_type != "text": return {"code": 0} content = json.loads(event["message"]["content"]) user_text = content.get("text", "").strip() user_id = event["sender"]["sender_id"]["open_id"] reply = run_agent(user_text, get_history(user_id), TOOLS) send_message(user_id, reply) return {"code": 0}这段代码里有两个容易忽略的点。消息类型过滤。飞书推送的事件里，message_type 可能是 text、image、file 等多种类型。如果不做过滤，遇到非文本消息时解析 content 会报错。我一开始没加这个判断，结果用户发了一张图片，整个服务直接抛异常。open_id 的使用。飞书用户的标识有 open_id、union_id、user_id 三种。open_id 是应用维度的，同一个用户在不同应用里 open_id 不同。用 open_id 做会话隔离是最稳妥的，不会跨应用串数据。3.3 消息发送的限流与重试飞书的消息发送 API 有频率限制，单个应用每秒能发的消息数量有限。如果短时间内大量发送，会被限流返回错误码。我的处理方式是在发送层加一个简单的队列和重试机制。消息先入队列，由一个单独的工作线程按固定速率消费，遇到限流错误就等待一段时间后重试。这样即使智能体短时间内产生多条回复，也不会触发限流。import time from collections import deque send_queue = deque() def send_worker(): while True: if not send_queue: time.sleep(0.1) continue user_id, text = send_queue.popleft() try: send_message_api(user_id, text) except RateLimitError: send_queue.appendleft((user_id, text)) time.sleep(1) time.sleep(0.2)这个 worker 的消费速率我设的是每秒 5 条，实测下来比较稳。如果你的应用消息量更大，需要根据实际配额调整。3.4 会话上下文的隔离与持久化每个用户的对话历史需要单独维护，不能混在一起。我用一个字典按 open_id 存储历史，简单直接。user_histories = {} def get_history(user_id): if user_id not in user_histories: user_histories[user_id] = [{"role": "system", "content": SYSTEM_PROMPT}] return user_histories[user_id]但这个方案有个明显问题：服务重启后历史全丢。生产环境需要把历史存到 Redis 或数据库里。我目前的处理是内存存储加定期落盘，重启时从文件恢复。对于小规模使用够用了，但如果要支撑更多用户，还是得上 Redis。另外，历史不能无限增长。我的策略是保留最近 20 条消息，超出部分从头部丢弃。系统提示词始终保留在第一位，不参与截断。4. 工具函数的实现与踩坑记录4.1 工具函数的返回值格式很关键工具函数返回什么，直接影响模型能不能正确理解结果。我最初的做法是直接返回 Python 对象，让 str() 去转换。结果模型经常看不懂，比如返回一个字典{'tasks': ['写周报', '开会']}，模型有时会把它当成一段乱码。后来我改成返回结构化的自然语言描述：def query_tasks_impl(date=None): tasks = fetch_tasks_from_db(date) if not tasks: return f"{date} 没有待办任务。" lines = [f"{i+1}. {t['title']}（截止 {t['deadline']}）" for i, t in enumerate(tasks)] return f"{date} 共有 {len(tasks)} 项待办：\n" + "\n".join(lines)这样模型拿到结果后，几乎不需要额外加工就能生成回复。工具函数的输出应该尽量接近最终展示给用户的形式，把格式化的负担放在工具层，而不是让模型去猜。4.2 参数校验和容错处理模型传参数不可能 100% 准确。我遇到过几种典型情况：日期格式不对、必填参数缺失、参数类型错误。如果工具函数不做校验，直接抛异常，整个对话就中断了。我的做法是在每个工具函数入口做一层校验，校验失败时返回一个友好的错误提示，而不是抛异常。这样模型能看到错误信息，有机会在下一轮修正参数。def query_tasks_impl(date=None): if date is None: date = datetime.now().strftime("%Y-%m-%d") try: datetime.strptime(date, "%Y-%m-%d") except ValueError: return f"日期格式不正确：{date}，请使用 YYYY-MM-DD 格式。" ...这个错误提示会作为工具结果返回给模型，模型看到后通常会重新调用一次，传正确的格式。实测下来，这种自我修正的成功率挺高的。4.3 耗时工具的超时控制有些工具函数可能耗时较长，比如调用外部 API 查询数据。如果工具执行卡住，整个对话循环就阻塞了。我一开始没做超时控制，结果有一次外部服务响应慢，用户等了半分钟才收到回复。后来我给每个工具函数加了超时装饰器，超过 5 秒就中断并返回超时提示。import signal def timeout_handler(signum, frame): raise TimeoutError("工具执行超时") def with_timeout(seconds=5): def decorator(func): def wrapper(*args, **kwargs): signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(seconds) try: return func(*args, **kwargs) finally: signal.alarm(0) return wrapper return decorator需要注意的是，signal 方案只在主线程有效。如果你的工具函数跑在子线程里，得用其他方式实现超时，比如用 concurrent.futures 的 timeout 参数。4.4 工具权限的边界控制不是所有用户都能调用所有工具。比如"发送全员提醒"这种工具，应该只对管理员开放。我在工具注册表里加了一个权限字段，执行前先检查用户身份。TOOLS = { "send_broadcast": { "description": "发送全员广播", "permission": "admin", ... } } def execute_tool(name, args, tools, user_role="member"): tool = tools.get(name) if not tool: return f"未知工具：{name}" required = tool.get("permission") if required and required != user_role: return f"你没有权限调用 {name}。" return tool["handler"](**args)这个权限检查放在 execute_tool 里，而不是放在每个工具函数内部，好处是统一管理，不会遗漏。用户角色从飞书事件里获取，或者维护一张用户角色表。5. 实测中的性能表现与优化方向5.1 响应延迟的构成分析用户从发消息到收到回复，中间经历了几段延迟：飞书推送事件到服务器（通常 100 毫秒以内）、模型调用（1 到 3 秒，取决于模型和网络）、工具执行（几十毫秒到几秒不等）、消息发送回飞书（100 毫秒左右）。大头在模型调用上。如果一轮对话涉及两次模型调用（一次决策、一次生成回复），总延迟就是 2 到 6 秒。这个延迟在聊天场景里可以接受，但如果用户期望秒回，就需要优化。我的优化手段是：对于明显不需要工具调用的消息，先用一个轻量规则判断，直接走模型生成回复，省掉一轮调用。比如用户说"你好"，没必要走工具决策流程。这个规则很简单，就是维护一个问候语列表，命中就直接回复。5.2 Token 消耗的控制策略Token 消耗主要来自三部分：系统提示词、对话历史、工具描述。系统提示词和工具描述是固定开销，每次调用都要带上。对话历史随轮次增长。我的控制策略是：系统提示词尽量精简，只保留必要的行为约束；工具描述只写关键信息，不写废话；对话历史保留最近 10 轮，超出截断。这样单次调用的 token 量能控制在 2000 以内，成本可控。如果工具数量很多，工具描述本身就会占用大量 token。这时候可以考虑工具分组，根据用户消息先判断属于哪个组，只加载该组的工具描述。不过这会增加一次判断调用，需要权衡。5.3 并发场景下的状态管理如果多个用户同时发消息，每个用户的对话循环是独立的，互不干扰。但共享的资源比如数据库连接、HTTP 客户端，需要做好并发控制。我用的是连接池加线程池的方案。每个用户请求分配一个线程，线程内完成整个对话循环。数据库连接从池里取，用完归还。这样即使几十个用户同时发消息，也不会出现资源竞争。需要注意的是，Python 的 GIL 限制了多线程的并行能力。如果并发量真的很大，需要考虑异步方案或者多进程部署。不过对于团队内部使用，几十个并发已经绰绰有余了。5.4 日志与可观测性智能体出问题时，排查起来比普通服务麻烦，因为涉及模型决策这个黑盒。我的做法是把每次对话的完整链路都记下来：用户消息、模型返回、工具调用、工具结果、最终回复。这样出问题时能完整复现。def log_turn(user_id, turn, msg, tool_results=None): entry = { "user_id": user_id, "turn": turn, "timestamp": time.time(), "assistant_message": msg.content if hasattr(msg, "content") else None, "tool_calls": [tc.function.name for tc in msg.tool_calls] if msg.tool_calls else [], "tool_results": tool_results } logger.info(json.dumps(entry, ensure_ascii=False))日志按天切分，保留 7 天。有了这些日志，模型调用错误、工具执行异常、参数传递问题，都能快速定位。6. 从 100 行到生产可用还差什么6.1 错误恢复与降级策略核心循环跑通只是第一步，生产环境要考虑各种异常情况。模型 API 超时怎么办？工具执行失败怎么办？飞书推送重复怎么办？我的处理是分层降级。模型调用失败时，重试两次，仍失败则返回"服务暂时不可用，请稍后再试"。工具执行失败时，把错误信息返回给模型，让模型决定是重试还是告知用户。飞书推送重复时，用消息 ID 做去重，同一条消息只处理一次。这些逻辑加起来大概又多了 50 行代码，但都是必要的。没有这些保护，服务在真实环境里跑不了几天就会出问题。6.2 配置与密钥管理模型 API Key、飞书应用密钥这些敏感信息，绝对不能硬编码在代码里。我用的是环境变量加配置文件的方式，敏感信息只放环境变量，配置文件里只放非敏感的配置项。import os CONFIG = { "llm_api_key": os.environ["LLM_API_KEY"], "llm_base_url": os.environ.get("LLM_BASE_URL", "https://api.example.com/v1"), "feishu_app_id": os.environ["FEISHU_APP_ID"], "feishu_app_secret": os.environ["FEISHU_APP_SECRET"], "max_turns": 5, "history_limit": 10 }这样代码可以放心提交到版本库，部署时通过环境变量注入密钥。如果团队有密钥管理服务，也可以对接上去，实现自动轮换。6.3 部署方式的选择我试过两种部署方式：一种是直接跑在一台常驻服务器上，用 systemd 管理进程；另一种是打包成容器，用容器编排平台管理。小规模使用，systemd 就够了，简单直接，出问题看日志方便。如果要多实例部署、自动扩缩容，那就得上容器方案。我目前的场景是团队内部使用，systemd 完全够用。部署时要注意的是：飞书事件订阅地址必须是 HTTPS，所以需要配置证书。我用的是反向代理加自动证书续期，这部分配置一次就好，后续不用管。6.4 后续可以扩展的方向这套 100 行的核心代码，扩展空间其实很大。我目前想到几个方向：一是增加更多工具，比如日历查询、文档检索、数据统计；二是支持多模态输入，让机器人能处理图片和文件；三是引入记忆机制，让机器人记住用户的偏好和历史行为。但扩展的前提是核心循环保持稳定。我的原则是：核心循环尽量不改，所有新能力都通过工具的形式接入。这样每加一个功能，风险都局限在那个工具函数里，不会影响整体稳定性。7. 一些实操心得和避坑建议工具描述要写得像给新人看的文档。模型对工具的理解完全依赖描述，描述里该写清楚的格式、范围、默认值，一个都不能省。我见过太多人工具描述写得含糊，然后抱怨模型调用不准。对话历史不要无限增长。我一开始图省事，把所有历史都留着，结果跑了几十轮之后，token 消耗暴涨，响应也变慢。后来加了截断策略，体验立刻好转。错误信息要返回给模型，而不是直接抛给用户。模型看到错误信息后，有很大概率能自我修正。直接抛给用户，用户也不知道该怎么办。权限检查要放在工具执行层，不要依赖模型判断。模型可能会被诱导调用不该调用的工具，代码层面的权限检查是最后一道防线。日志要记全，尤其是模型返回的原始内容。排查问题时，模型到底返回了什么，是定位问题的关键。只记最终回复是不够的。飞书的消息推送有重试机制，同一条消息可能推送多次。一定要做去重，否则用户会收到重复回复。去重可以用消息 ID，也可以用时间窗口加内容哈希。最后说一个我踩过的坑：飞书的事件订阅地址验证，challenge 字段必须原样返回，不能做任何加工。我一开始以为要加密或者编码，折腾了半天才发现直接返回就行。这种细节文档里写得清楚，但容易看漏。