很多所谓的“AI 机器人”,本质上只是收到关键词后调用一次接口,再把文本答案发回来。本文分享一种更实用的做法:把飞书作为交互入口,把 Codex 作为任务执行核心,再通过一个运行在 Windows 本机的桥接服务,让 AI 能在授权范围内读取文件、执行命令、操作浏览器,并把结果和附件重新发回飞书。

一、它到底能做什么?

我给这套工具起了一个很直白的名字:Codex 1 号机。

它运行在自己的 Windows 电脑上。人在飞书里发消息,机器人会结合最近的群聊上下文理解需求,然后调用本机 Codex 完成任务。

目前它可以处理这些场景:

  • 根据上下文编写、修改和校验 SQL;
  • 读取本地项目代码,定位问题并完成修改;
  • 操作已登录的浏览器,填写表单、编辑文章或处理后台页面;
  • 接收飞书图片和文件,下载到独立工作区后交给 Codex 分析;
  • 将生成的文档、截图和其他文件作为附件发回飞书;
  • 保存明确指定的长期规则,例如固定的发布流程;
  • 在执行过程中持续更新进度,而不是长时间没有响应。

它与普通问答机器人的最大区别是:回答问题只是能力之一,完成任务才是目标。

二、整体架构

整条链路可以概括为:

飞书用户
   │
   │ 消息事件 / 图片 / 文件
   ▼
飞书开放平台
   │
   │ WebSocket 长连接
   ▼
本地 Bridge(Python)
   ├── 身份校验与群聊 @ 判断
   ├── 去重、排队与会话隔离
   ├── 上下文、群消息和长期记忆
   ├── 进度卡片与附件收发
   └── 安全规则与审批边界
   │
   │ codex exec --json
   ▼
Codex 本地执行环境
   ├── PowerShell / Git / 项目文件
   ├── Skills / MCP / 外部工具
   └── Edge CDP(可见浏览器)
   │
   ▼
执行结果、证据、截图或附件回传飞书

OpenAI 官方文档把这种模式称为自托管执行环境:执行器运行在自己的电脑、容器或服务器中,可以按授权访问本地文件、命令和工具;Agent 侧负责规划与调用。官方也明确区分了 Agents API、Agents SDK 和 Responses API,实际项目可以根据状态管理、工具编排和集成深度选择不同入口。

三、为什么要单独做一层 Bridge?

直接把飞书消息丢给模型,看起来只需要几十行代码,但很快会遇到一堆工程问题:

  1. 飞书事件可能重复投递,同一任务不能执行两次;
  2. 群聊里不是每句话都应该触发机器人;
  3. 长任务需要实时显示进度,否则用户会以为机器人已经卡死;
  4. 不同群、不同私聊必须使用独立工作目录;
  5. 图片和附件要先下载,本地结果也要有可靠的回传通道;
  6. 模型上下文有限,短期对话和长期规则必须分开保存;
  7. 发布、删除、付款、发消息等操作必须设置人工确认边界;
  8. 浏览器会话需要复用,不能每个步骤都重新登录。

所以 Bridge 并不是一个“转发器”,而是整个系统的控制平面。

四、准备飞书机器人

在飞书开放平台创建企业自建应用,并开启机器人能力。至少需要配置以下权限,具体名称以控制台当前页面为准:

  • 接收消息事件;
  • 读取群聊或私聊消息;
  • 发送、回复和更新消息;
  • 下载消息中的图片和文件;
  • 上传文件并发送文件消息。

消息接收推荐使用 WebSocket 长连接。它不要求把本机端口暴露到公网,更适合个人电脑和内网环境。

安装 Python 依赖:

pip install lark-oapi

配置文件只保存必要参数,并且不要提交到 Git:

{
  "app_id": "cli_xxx",
  "app_secret": "从环境变量读取更安全",
  "allowed_open_ids": [],
  "bot_mention_names": ["Codex 1号机"],
  "timeout_seconds": 1800,
  "history_max_rounds": 10,
  "group_history_max_messages": 20,
  "outbox_enabled": true,
  "sandbox": "workspace-write",
  "bypass_approvals": false
}

生产环境建议将 app_secret 放进系统环境变量或密钥管理服务,配置文件中只写变量名。

五、消息入口:先判断,再执行

一个可靠的消息处理器至少要完成五件事:校验发送者、判断是否需要响应、事件去重、下载附件、把任务放进队列。

下面是经过简化的结构代码:

from queue import Queue
from threading import Thread, Lock

task_queue = Queue()
processed_ids = set()
processed_lock = Lock()

def is_duplicate(message_id: str) -> bool:
    with processed_lock:
        if message_id in processed_ids:
            return True
        processed_ids.add(message_id)
        return False

def on_message(event):
    message = event.message

    if not sender_is_allowed(event.sender):
        return

    if is_duplicate(message.message_id):
        return

    # 群聊只记录普通消息作为上下文;被 @ 时才执行任务
    record_group_context(message)
    if message.chat_type == "group" and not mentions_bot(message):
        return

    task_queue.put({
        "chat_id": message.chat_id,
        "message_id": message.message_id,
        "text": extract_text(message),
        "attachments": download_attachments(message),
    })

def worker():
    while True:
        task = task_queue.get()
        try:
            process_task(task)
        finally:
            task_queue.task_done()

Thread(target=worker, daemon=True).start()

这里有一个很重要的体验细节:群聊中未 @ 机器人的消息可以进入短期上下文,但不能直接触发执行。 这样用户真正 @ 机器人时,它能理解“按照刚才讨论的方案修改”中的“刚才”指什么,同时又不会在群里频繁插话。

六、为每个会话创建独立工作区

不要让所有会话共享一个临时目录。推荐按 chat_id 隔离:

workspace/
├── chat_a/
│   ├── files/          # 从飞书下载的附件
│   ├── outbox/         # 等待发回飞书的文件
│   ├── history.json    # 当前会话历史
│   └── codex-events.jsonl
└── chat_b/
    └── ...

shared_memory/
└── rules.md            # 用户明确要求长期保存的规则

这里要把两类信息分开:

  • 短期上下文:最近若干轮对话、最近群消息,用于理解当前任务;
  • 长期记忆:只有用户明确说“记住”“以后都按这个规则”时才写入。

如果把所有对话都自动写进长期记忆,系统会越来越混乱,甚至把一次性的临时要求误当成永久规则。

七、调用 Codex,并流式读取进度

Bridge 可以通过 codex exec --json 启动非交互任务。--json 会输出结构化事件,便于提取命令执行、文件修改和最终回答。

import json
import subprocess

def run_codex(cwd: str, prompt: str):
    command = [
        "codex",
        "exec",
        "--json",
        "--skip-git-repo-check",
        "--color", "never",
        "-C", cwd,
        "-",
    ]

    process = subprocess.Popen(
        command,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        encoding="utf-8",
    )

    process.stdin.write(prompt)
    process.stdin.close()

    final_text = ""
    for line in process.stdout:
        event = json.loads(line)
        save_event(event)
        update_progress_card(summarize_event(event))

        if event.get("type") == "agent_message":
            final_text = event.get("text", final_text)

    return final_text

实际实现中还应加入:

  • 超时和取消机制;
  • 标准错误流并发读取,避免管道阻塞;
  • 进度更新节流,避免频繁调用飞书接口;
  • 异常退出后的最后有效消息兜底;
  • 任务级日志和可追踪的事件文件。

OpenAI 官方的自托管 Sandbox 文档也采用了“本地执行器保持运行、接收任务、返回结果”的思路。官方说明执行器可运行命令、读写文件并使用本地 MCP 服务,同时建议隔离不同用户或不同工作负载的环境。

八、让机器人真正会操作网页

很多后台任务没有稳定 API,只能通过浏览器完成。我的方案是让 Edge 以固定调试端口和独立用户目录启动:

Start-Process msedge.exe -ArgumentList `
  '--remote-debugging-port=9555', `
  '--user-data-dir=C:\tmp\codex-edge-profile', `
  '--no-first-run', `
  '--new-window', `
  'https://example.com'

然后通过 Chrome DevTools Protocol(CDP)获取现有标签页:

http://127.0.0.1:9555/json

这样做有三个好处:

  • 浏览器是可见的,用户能随时观察和接管;
  • 登录状态保存在独立 Profile 中,无需每一步重新登录;
  • 后续任务可以按 URL 或标签页 ID 重新绑定,不必重复打开页面。

浏览器自动化必须遵循“读取页面状态 → 执行动作 → 再次验证”的闭环。不能只因为点击函数没有报错,就宣称任务已经完成。

九、用 Outbox 把本地文件发回飞书

让模型直接调用“发送飞书文件”接口会增加工具复杂度。一个简单可靠的设计是使用 Outbox:

  1. Codex 将要发送的文件复制到当前会话的 outbox/;
  2. 后台线程定时扫描;
  3. 先将文件原子重命名为 .sending,防止重复发送;
  4. 上传飞书并发送文件消息;
  5. 成功后移入 outbox/sent/,失败则恢复原名重试。

还需要限制:

  • 只允许发送当前会话工作区内的普通文件;
  • 拒绝目录、软链接、空文件和超大文件;
  • 文件名必须清洗,防止路径穿越;
  • 日志中不能记录密钥和敏感正文。

十、安全边界比“能不能做”更重要

本地 Agent 拥有的能力越强,安全边界越要明确。

我的原则是把动作分成两类:

可以直接执行

  • 读取文件、搜索代码和查看页面;
  • 生成文章、SQL、脚本和报告;
  • 在限定工作区内创建或修改草稿;
  • 执行可逆、低风险的诊断操作。

最后一步必须确认

  • 对外发布内容;
  • 删除或覆盖重要数据;
  • 提交代码、部署生产环境;
  • 付款、发消息、上传敏感文件;
  • 保存密码、Token 或 API Key。

官方 Skills 文档也特别提醒:Skill 会影响规划、工具调用和命令执行,应当把它视为具有权限的代码与指令进行审查;写操作和高影响操作应通过明确审批和策略检查进行约束。

因此,即使是个人机器人,也不建议一开始就配置全盘读写和无条件自动批准。更稳妥的默认值是限定工作区,并对高风险动作启用按需确认。

十一、这套架构最值得复用的经验

经过实际使用,我认为下面几条比“调用哪个模型”更重要:

  1. 先理解意图,再选择工具。 不要把所有消息都套进固定工作流。
  2. 用结果和证据闭环。 操作后读取页面、文件或状态进行验证。
  3. 短期上下文与长期记忆分离。 长期规则必须由用户明确授权保存。
  4. 浏览器会话要持久化。 固定调试端口、Profile 和标签页重绑定策略。
  5. 所有任务都要可追踪。 保留结构化事件、命令摘要和最终结果。
  6. 副作用动作设置确认点。 确认应发生在最后一步,而不是任务开始前反复打断。
  7. 先做单机可靠性,再考虑多 Agent。 去重、超时、队列和恢复比炫酷编排更基础。

十二、可以继续扩展什么?

完成基础桥接后,还可以继续增加:

  • 定时任务和主动通知;
  • Git 仓库级权限隔离;
  • 多台执行机的任务路由;
  • 数据库只读连接和 SQL 审核;
  • 针对固定业务流程的 Skills;
  • 操作审计、成本统计和失败重放;
  • 基于容器或虚拟机的一次性沙箱。

OpenAI 的 Agent Skills 由一个包含 SKILL.md 的目录及其支持文件组成,适合把重复流程固化为可复用能力。与其不断把所有规则堆进一个超长提示词,不如将发布文章、代码审查、数据分析等流程拆成独立 Skill,并为每个 Skill 设置清晰的适用范围和权限边界。

结语

把 Codex 接入飞书并不只是“换一个聊天入口”。真正有价值的是建立一条稳定的本地执行链路:飞书负责触达和协作,Bridge 负责上下文、权限与可靠性,Codex 负责理解、规划和执行,本机环境提供文件、终端和浏览器能力。

当这几层组合起来,机器人就不再只是一个会回答问题的窗口,而是一个能在你授权范围内持续完成工作的本地 AI 助手。

---

参考资料

说明:本文代码为了讲清架构进行了简化。生产使用时,请补齐凭据管理、权限隔离、日志脱敏、任务取消、异常恢复和审计机制。

Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐