背景在之前的文章中Nanobot 轻量级的个人AI助手,我们分析了nanobot onboard命令的实现,该命令的主要作用是做一系列的初始化工作 这次我们分析另一个命令nanobot gateway,从整理上来看该nanobot用到了Typer,Rich,Questionary,prompt_toolkit这种现代、美观且交互式命令行界面 (CLI) 的强大工具组合。Typer 用于定义 CLI 结构和参数Rich 负责文本样式、表格、面板和 Markdown 渲染Questionary 用于创建交互式问答界面其中Rich中的Console,Markdown,Table,Text用来进行渲染支持颜色、表格、面板、语法高亮和 Markdown 以更好的进行个性化的展示。nanobot gateway命令gateway 是 Typer 子命令用于 启动 nanobot 网关进程加载配置、同步工作区模板、创建消息总线与 LLM Provider、会话管理、Cron、AgentLoop、ChannelManager各聊天频道、Heartbeat最后用 asyncio.run 并发跑 agent 主循环 所有channel并在退出时做清理.defgateway(port:int|Nonetyper.Option(None,--port,-p,helpGateway port),workspace:str|Nonetyper.Option(None,--workspace,-w,helpWorkspace directory),verbose:booltyper.Option(False,--verbose,-v,helpVerbose output),config:str|Nonetyper.Option(None,--config,-c,helpPath to config file),):参数说明-p/--port端口未传入时使用配置中的config.gateway.port。貌似在这里也没有太大的作用-w/--workspace覆盖配置中的工作区路径经_load_runtime_config写回Config。-v/--verbose为True时对标准库logging设置DEBUG便于查看更底层日志。-c/--config指定配置文件路径同样由_load_runtime_config处理。核心组件实例化MessageBus()进程内入站/出站消息队列。_make_provider(config)按当前默认模型解析 provider构造具体 Provider 实例含 generation 设置若缺少 API key 等会typer.Exit。SessionManager(config.workspace_path)会话持久化JSONL 等与工作区目录绑定。CronService(cron_store_path)随后将cron.on_job设为下文所述异步回调。MessageBus这是消息总线主要包括从channel传递过来的Message和需要发送给channel的Message:self.inbound:asyncio.Queue[InboundMessage]asyncio.Queue()self.outbound:asyncio.Queue[OutboundMessage]asyncio.Queue()asyncio.Queue 是 Python asyncio 库中用于在异步协程Coroutine之间安全传递数据的先进先出FIFO数据结构。这里默认允许放入任意数量的元素。它是协程同步的能安全地在协程间共享数据类似于线程同步。make_provider根据当前Config主要是默认 model 和 providers.*构造并返回一个 LLM Provider 实例供 gateway、agent 等命令里的 AgentLoop 使用。这里的provider主要有以下几种azure_openai 对应 AzureOpenAIProvideropenai_compat 对应 OpenAICompatProvideropenai_codex 对应 OpenAICodexProvideranthropic 对应 AnthropicProvider这些provider提供与大模型的交互SessionManager这里主要负责在工作区里 按「会话键」加载、缓存、保存对话会话。每个会话对应磁盘上的一个 JSONL 文件键一般是 channel:chat_id如下├── sessions │ ├── cli_direct.jsonl │ └── feishu_ou_52720638d0cxxxxf.jsonl这里的目录在workspace/session下2. Cron 服务任务持久化路径workspace/cron/jobs.json。CronService(cron_store_path)下面会把 cron.on_job 设为异步回调if is_default_workspace(config.workspace_path): _migrate_cron_store(config) # Create cron service with workspace-scoped store cron_store_path config.workspace_path / cron / jobs.json cron CronService(cron_store_path) ... async def on_cron_job(job: CronJob) - str | None: Execute a cron job through the agent. from nanobot.agent.tools.cron import CronTool from nanobot.agent.tools.message import MessageTool from nanobot.utils.evaluator import evaluate_response reminder_note ( [Scheduled Task] Timer finished.\n\n fTask {job.name} has been triggered.\n fScheduled instruction: {job.payload.message} ) cron_tool agent.tools.get(cron) cron_token None if isinstance(cron_tool, CronTool): cron_token cron_tool.set_cron_context(True) try: resp await agent.process_direct( reminder_note, session_keyfcron:{job.id}, channeljob.payload.channel or cli, chat_idjob.payload.to or direct, ) finally: if isinstance(cron_tool, CronTool) and cron_token is not None: cron_tool.reset_cron_context(cron_token) response resp.content if resp else message_tool agent.tools.get(message) if isinstance(message_tool, MessageTool) and message_tool._sent_in_turn: return response if job.payload.deliver and job.payload.to and response: should_notify await evaluate_response( response, job.payload.message, provider, agent.model, ) if should_notify: from nanobot.bus.events import OutboundMessage await bus.publish_outbound(OutboundMessage( channeljob.payload.channel or cli, chat_idjob.payload.to, contentresponse, )) return response cron.on_job on_cron_jobon_cron_job 这个是在定时任务触发时调用该方法主要是逻辑是构建一个Message信息并直接 让大模型去处理对应的信息并且保留该job对应的channel和chat_id(这个会在消息处理的时候进行传递具体是在_run_agent_loop方法中调用_set_tool_context),同时也会保留当前协程的上下文(用contextvars保留)防止在执行任务的增加任务而导致重复增加调度任务(具体见CronTool的execute方法中add的判断)根据大模型的反馈来决定需不需要进行通知。ChannelManagerChannelManager(config, bus)按配置启动feishu/dingding等channel插件与MessageBus对接,用于从channel发送消息到处理队列以及传递消息给对应的Channel这里会通过pkgutil.iter_modules(nanobot.channels)遍历nanobot.channels路径下所有子模块和子包,并结合importlib.import_module如果模块已导入则直接返回 dir(mod)动态的获取模块的BaseChannel子类。Heartbeat 服务周期性的读取 Heartbeat.md 文件并使用大模型总结该文件下的活跃任务调用大模型去执行最后还是调用大模型去判断是否需要把任务的结果回传给Channel对于 heartbeat session则只会保留最近几条信息,默认是8条由gateway.heartbeat.keepRecentMessages来决定对于 回传给哪个 Channel会按照激活的Channel从最新的活跃时间排序取最新的Channel(排除纯本地(CLI) / 系统(System))启动cron hearbeat 服务并开启主流程循环启动所有的channelawaitcron.start()awaitheartbeat.start()awaitasyncio.gather(agent.run(),channels.start_all(),注意这里的AgentLoop.run是个死循环。所以该命令会使进程hang在这里。