高级：让小智 AI 不用 API Key 接入 Claude 和 Codex

你会得到什么

让 ESP32 上的小智 AI 用你自己的 Claude 和 Codex（ChatGPT） 订阅来回答问题，不用申请 API Key，也不用按 token 付费。

做法：给小智的服务端 xiaozhi-esp32-server 加 3 个 LLM 提供者：

提供者	做什么
claude_cli	启动本机已登录的 claude 命令行，流式读取回答
codex_cli	启动本机已登录的 codex 命令行，读取回答
fallback	按顺序尝试：Claude 出错就自动换 Codex，出错的先冷却 300 秒

关键点： OAuth 登录全部交给官方 CLI 自己处理。服务器只是"替你在命令行里打一行字"，不读、不复制、也不刷新任何令牌。令牌过期了，CLI 自己会续。

开始前准备

1. 一块已经刷好小智固件的 ESP32 开发板，能连 Wi-Fi、能对话。还没刷的话，先按小智固件仓库的说明刷好。
2. 用源码方式部署的 xiaozhi-esp32-server（不是 Docker）。CLI 的登录状态存在你本机用户目录里，Docker 容器里看不到，所以服务端要直接跑在登录了 CLI 的这台电脑上。代码在仓库的 main/xiaozhi-server 目录。
3. 两个 CLI 都装好并登录：

# Claude Code：装好后运行一次 claude，按提示用你的 Claude 账号登录
npm install -g @anthropic-ai/claude-code
claude

# Codex：用 ChatGPT 账号登录
npm install -g @openai/codex
codex login

先在终端里各试一句，确认能回答：

echo "你好" | claude -p --model claude-haiku-5-5
echo "你好" | codex exec --skip-git-repo-check -

4. 把设备的 OTA 地址指向你的服务器，比如 http://你的电脑IP:8003/xiaozhi/ota/，让设备连到你自己部署的服务端。

订阅账号只适合自己用。别把这个服务端开放给别人共用你的账号。

原理：小智怎么调用 LLM

小智服务端的每个 LLM 都是 core/providers/llm/<类型>/<类型>.py 里的一个 LLMProvider 类。配置里写 type: openai，它就加载 openai/openai.py。

每个提供者要实现两个方法，照 core/providers/llm/openai 的写法：

• response(session_id, dialogue)：普通对话，一段一段地 yield 文本，边生成边送去语音合成。
• response_with_functions(session_id, dialogue, functions)：带工具的对话，yield (文本, 工具调用)。要调用工具时，文本是 None，工具调用是一个列表，每项有 id、function.name、function.arguments。

所以我们要做的就是：把 dialogue 拼成一段提示词，交给 CLI，把 CLI 的输出拆回这两种格式。

第 1 步：公共部分 cli_common.py

两个 CLI 提供者共用三件事：

• 拼提示词：把系统提示词单独拿出来，其余对话拼成"用户：… 助手：…"的记录。
• 跑子进程：启动 CLI，对话从 stdin 传进去，逐行解析它输出的 JSON；超时就杀掉进程；顺便记下首字延迟和总耗时。
• 工具调用：CLI 已经关掉了自带的工具，所以我们把 functions 写进提示词，要求模型需要调用工具时只输出 <tool_call>{"name":"...","arguments":{...}}</tool_call>。流式输出时一旦看到这个标签就截住，不送去念，结束后解析成工具调用返回。

新建 core/providers/llm/cli_common.py：

"""claude_cli / codex_cli 共用：拼提示词、跑子进程、解析 <tool_call>。"""
import json
import re
import subprocess
import threading
import time
import uuid
from types import SimpleNamespace

from config.logger import setup_logging

TAG = __name__
logger = setup_logging()

TOOL_RULE = """你可以调用下面这些工具。需要调用工具时，只输出：
<tool_call>{"name": "工具名", "arguments": {参数}}</tool_call>
不要输出任何别的文字，不要解释，也不要把工具调用念给用户听。
不需要工具时，直接用简短的口语回答。

工具列表（每行一个，JSON）：
"""


def tools_text(functions):
    lines = []
    for f in functions or []:
        fn = f.get("function", f)
        lines.append(json.dumps({
            "name": fn.get("name"),
            "description": fn.get("description", ""),
            "parameters": fn.get("parameters", {}),
        }, ensure_ascii=False))
    return "\n".join(lines)


def _get(obj, key, default=None):
    return obj.get(key, default) if isinstance(obj, dict) else getattr(obj, key, default)


def build_prompt(dialogue, functions=None):
    """把 OpenAI 格式的 dialogue 拆成 (system, prompt)。"""
    system, turns = [], []
    for m in dialogue:
        role, content = m.get("role"), m.get("content") or ""
        if role == "system":
            system.append(content)
        elif role == "assistant" and m.get("tool_calls"):
            for tc in m["tool_calls"]:
                fn = _get(tc, "function", {})
                call = {"name": _get(fn, "name"), "arguments": json.loads(_get(fn, "arguments") or "{}")}
                turns.append("助手：<tool_call>" + json.dumps(call, ensure_ascii=False) + "</tool_call>")
            if content:
                turns.append("助手：" + content)
        elif role == "tool":
            turns.append("工具结果：" + content)
        else:
            turns.append(("用户：" if role == "user" else "助手：") + content)
    if functions:
        system.append(TOOL_RULE + tools_text(functions))
    prompt = "下面是对话记录。请以“助手”的身份回复最后一条，只输出回复内容本身。\n\n" + "\n".join(turns)
    return "\n\n".join(system), prompt


def run_cli(name, cmd, stdin_text, parse, timeout=60, cwd=None):
    """启动 CLI，逐行读 JSON，parse(obj) 产出文本片段。记录首字和总耗时。"""
    t0 = time.time()
    p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                         stderr=subprocess.PIPE, text=True, encoding="utf-8", cwd=cwd)
    err = []
    threading.Thread(target=lambda: err.append(p.stderr.read()), daemon=True).start()
    timer = threading.Timer(timeout, p.kill)
    timer.start()
    first = None
    try:
        p.stdin.write(stdin_text)
        p.stdin.close()
        for line in p.stdout:
            line = line.strip()
            if not line.startswith("{"):
                continue
            try:
                obj = json.loads(line)
            except ValueError:
                continue
            for text in parse(obj):
                if text:
                    if first is None:
                        first = time.time()
                    yield text
        rc = p.wait()
        if rc != 0 or first is None:
            raise RuntimeError(f"{name} 退出码 {rc}，没有输出：{''.join(err)[-300:]}")
        logger.bind(tag=TAG).info(
            f"{name} 首字 {(first - t0) * 1000:.0f} ms，总耗时 {(time.time() - t0) * 1000:.0f} ms")
    finally:
        timer.cancel()
        if p.poll() is None:
            p.kill()


START = "<tool_call>"


def _cut(buf):
    """返回 (可以安全念出来的长度, 是否已经出现 <tool_call>)。"""
    i = buf.find(START)
    if i >= 0:
        return i, True
    for k in range(min(len(START) - 1, len(buf)), 0, -1):
        if START.startswith(buf[-k:]):  # 末尾可能是半个标签，先扣住
            return len(buf) - k, False
    return len(buf), False


def parse_tool_calls(text):
    calls = []
    for raw in re.findall(r"<tool_call>\s*(\{.*?\})\s*(?:</tool_call>|$)", text, re.S):
        try:
            data = json.loads(raw)
        except ValueError:
            continue
        calls.append(SimpleNamespace(
            index=len(calls),
            id=uuid.uuid4().hex,
            function=SimpleNamespace(
                name=data.get("name"),
                arguments=json.dumps(data.get("arguments") or {}, ensure_ascii=False),
            ),
        ))
    return calls


def with_tool_calls(chunks):
    """文本照常流出 (text, None)；遇到 <tool_call> 就截住，结束时产出 (None, [tool_call])。"""
    buf, tool = "", None
    for c in chunks:
        if tool is not None:
            tool += c
            continue
        buf += c
        n, found = _cut(buf)
        if n:
            yield buf[:n], None
        if found:
            tool, buf = buf[n:], ""
        else:
            buf = buf[n:]
    if tool is None:
        if buf:
            yield buf, None
        return
    calls = parse_tool_calls(tool)
    if calls:
        yield None, calls
    else:
        logger.bind(tag=TAG).warning(f"工具调用格式不对，已丢弃：{tool[:200]}")

with_tool_calls 里有个细节：文本是一小段一小段到的，<tool_call> 可能被拆在两段里。所以末尾如果像半个标签（比如 <tool），先扣住不念，等下一段到了再判断。

第 2 步：claude_cli

新建 core/providers/llm/claude_cli/claude_cli.py。命令就是：

claude -p --output-format stream-json --verbose --model claude-haiku-5-5 --tools ""

再加几个参数让它更快、更干净：

• --include-partial-messages：逐字输出。不加的话要等整段写完才拿到文字。
• --tools ""：关掉 Claude Code 自带的全部工具（读文件、跑命令等），它只负责说话。
• --strict-mcp-config：不加载你本机配的 MCP 服务器，启动更快。
• --system-prompt：用小智的角色提示词替换 Claude Code 默认的那一大段。

from config.logger import setup_logging
from core.providers.llm.base import LLMProviderBase
from core.providers.llm.cli_common import build_prompt, run_cli, with_tool_calls

TAG = __name__
logger = setup_logging()


class LLMProvider(LLMProviderBase):
    """调用本机已登录的 claude CLI。OAuth 由 CLI 自己处理，这里不碰令牌。"""

    def __init__(self, config):
        self.bin = config.get("bin") or "claude"
        self.model = config.get("model_name") or "claude-haiku-5-5"
        self.timeout = int(config.get("timeout") or 60)

    def _cmd(self, system):
        return [
            self.bin, "-p",
            "--output-format", "stream-json", "--verbose",
            "--include-partial-messages",      # 逐字输出，首字更快
            "--model", self.model,
            "--tools", "",                     # 关掉 Claude Code 自带的全部工具
            "--strict-mcp-config",             # 不加载本机的 MCP 服务器
            "--no-session-persistence",        # 不在本机留会话记录
            "--system-prompt", system or "你是一个语音助手，回答简短口语化。",
        ]

    @staticmethod
    def _parser():
        seen_delta = [False]

        def parse(o):
            t = o.get("type")
            if t == "stream_event":
                e = o.get("event") or {}
                d = e.get("delta") or {}
                if e.get("type") == "content_block_delta" and d.get("type") == "text_delta":
                    seen_delta[0] = True
                    yield d.get("text", "")
            elif t == "assistant" and not seen_delta[0]:   # 老版本没有逐字事件
                for b in (o.get("message") or {}).get("content", []):
                    if b.get("type") == "text":
                        yield b.get("text", "")
            elif t == "result" and o.get("is_error"):
                raise RuntimeError(f"claude 报错：{o.get('result')}")
        return parse

    def _stream(self, dialogue, functions=None):
        system, prompt = build_prompt(dialogue, functions)
        return run_cli("claude_cli", self._cmd(system), prompt, self._parser(), self.timeout)

    def response(self, session_id, dialogue, **kwargs):
        yield from self._stream(dialogue)

    def response_with_functions(self, session_id, dialogue, functions=None, **kwargs):
        yield from with_tool_calls(self._stream(dialogue, functions))

第 3 步：codex_cli

新建 core/providers/llm/codex_cli/codex_cli.py。命令是：

codex exec --json --skip-git-repo-check -

最后的 - 表示提示词从 stdin 读。--json 会一行输出一个事件，回答在 type 为 item.completed、item.type 为 agent_message 的那一行里。

另外三处防护：

• --sandbox read-only：Codex 是会跑命令的编程 Agent，这里只让它说话，不让它改任何东西。
• 在一个空的临时目录里运行，不让它读到服务端的代码和配置。
• model_reasoning_effort=low：语音对话不需要深度思考，调低了回答更快。

import tempfile

from config.logger import setup_logging
from core.providers.llm.base import LLMProviderBase
from core.providers.llm.cli_common import build_prompt, run_cli, with_tool_calls

TAG = __name__
logger = setup_logging()


class LLMProvider(LLMProviderBase):
    """调用本机已登录的 codex CLI（codex login）。OAuth 由 CLI 自己处理。"""

    def __init__(self, config):
        self.bin = config.get("bin") or "codex"
        self.model = config.get("model_name")
        self.effort = config.get("reasoning_effort") or "low"
        self.timeout = int(config.get("timeout") or 90)
        self.workdir = tempfile.mkdtemp(prefix="xiaozhi-codex-")   # 空目录，不让它读到项目文件

    def _cmd(self):
        cmd = [self.bin, "exec", "--json", "--skip-git-repo-check",
               "--sandbox", "read-only",
               "-c", f"model_reasoning_effort={self.effort}"]
        if self.model:
            cmd += ["-m", self.model]
        return cmd + ["-"]          # "-" 表示提示词从 stdin 读

    @staticmethod
    def parse(o):
        t = o.get("type")
        item = o.get("item") or {}
        if t == "item.completed" and item.get("type") == "agent_message":
            yield item.get("text", "")
        elif t in ("turn.failed", "error"):
            raise RuntimeError(f"codex 报错：{o.get('error') or o.get('message')}")

    def _stream(self, dialogue, functions=None):
        system, prompt = build_prompt(dialogue, functions)
        # codex exec 没有单独的 system 参数，拼在最前面
        text = system + "\n\n不要运行任何命令，不要读写文件，直接回答。\n\n" + prompt
        return run_cli("codex_cli", self._cmd(), text, self.parse, self.timeout, cwd=self.workdir)

    def response(self, session_id, dialogue, **kwargs):
        yield from self._stream(dialogue)

    def response_with_functions(self, session_id, dialogue, functions=None, **kwargs):
        yield from with_tool_calls(self._stream(dialogue, functions))

codex exec --json 目前不逐字输出，整段回答写完才返回，所以它的首字延迟基本等于总耗时。把它放在链条第二位当备用正合适。

第 4 步：fallback

新建 core/providers/llm/fallback/fallback.py。它本身不调模型，只按 chain 的顺序把请求交给别的提供者：

• 前一个还没说出第一个字就出错（没登录、命令不存在、超时、额度用完），就换下一个。
• 出错的那个记下"冷却到几点"，300 秒内直接跳过，不用每句话都先等它失败一次。
• 如果已经念出一半才出错，就不换了。换一个模型从头说，用户会听到两遍开头。

import os
import time

from config.config_loader import get_project_dir, merge_configs, read_config
from config.logger import setup_logging
from core.providers.llm.base import LLMProviderBase
from core.utils.llm import create_instance

TAG = __name__
logger = setup_logging()


class LLMProvider(LLMProviderBase):
    """按 chain 顺序尝试；出错的提供者冷却 cooldown 秒，期间直接跳过。"""

    def __init__(self, config):
        self.cooldown = int(config.get("cooldown") or 300)
        llm_conf = self._llm_section()
        self.providers = []
        for name in config.get("chain") or []:
            conf = llm_conf.get(name)
            if not conf:
                logger.bind(tag=TAG).error(f"fallback: 找不到 LLM.{name} 的配置，已跳过")
                continue
            self.providers.append((name, create_instance(conf.get("type", name), conf)))
        if not self.providers:
            raise ValueError("fallback: chain 里没有可用的 LLM")
        self.down_until = {}

    @staticmethod
    def _llm_section():
        d = get_project_dir()
        conf = read_config(d + "config.yaml")
        custom = d + "data/.config.yaml"
        if os.path.exists(custom):
            conf = merge_configs(conf, read_config(custom))
        return conf.get("LLM", {})

    def _run(self, call):
        now = time.time()
        ready = [(n, p) for n, p in self.providers if self.down_until.get(n, 0) <= now]
        last = None
        for name, p in ready or self.providers:   # 全在冷却就都再试一次，总比不说话好
            started = False
            try:
                for item in call(p):
                    started = True
                    yield item
                self.down_until.pop(name, None)
                return
            except Exception as e:
                last = e
                self.down_until[name] = time.time() + self.cooldown
                logger.bind(tag=TAG).warning(f"{name} 出错，冷却 {self.cooldown} 秒：{e}")
                if started:   # 已经念出一部分了，不能换一个模型从头再说
                    return
        raise last or RuntimeError("fallback: 没有可用的 LLM")

    def response(self, session_id, dialogue, **kwargs):
        yield from self._run(lambda p: p.response(session_id, dialogue, **kwargs))

    def response_with_functions(self, session_id, dialogue, functions=None, **kwargs):
        yield from self._run(
            lambda p: p.response_with_functions(session_id, dialogue, functions=functions, **kwargs))

fallback 只拿得到自己那一段配置，所以它自己读一遍 config.yaml 和 data/.config.yaml，按名字找到 claude_cli、codex_cli 的配置。如果你用的是智控台（manager-api）下发配置，需要改成从智控台读，或者把子配置直接写进 fallback 里。

第 5 步：改配置

编辑 data/.config.yaml，加上三个 LLM，把选中的 LLM 设成 fallback。意图识别要用 function_call，工具调用才会走 response_with_functions：

selected_module:
  LLM: fallback
  Intent: function_call

LLM:
  claude_cli:
    type: claude_cli
    model_name: claude-haiku-5-5
    timeout: 60
  codex_cli:
    type: codex_cli
    reasoning_effort: low
    timeout: 90
  fallback:
    type: fallback
    chain: [claude_cli, codex_cli]
    cooldown: 300

改完重启服务端：

cd main/xiaozhi-server
python app.py

第 6 步：测试

先不用设备，单独测提供者

在 main/xiaozhi-server 目录下新建 test_cli_llm.py，直接调用 fallback，测普通对话、工具调用和拿到工具结果后的回答：

import sys, time
from core.utils.llm import create_instance
from config.config_loader import read_config
name = sys.argv[1] if len(sys.argv) > 1 else "fallback"
llm = create_instance(read_config("data/.config.yaml")["LLM"][name]["type"], read_config("data/.config.yaml")["LLM"][name])
sysmsg = {"role": "system", "content": "你是小智，一个语音助手，回答简短口语化，不超过两句话。"}
print("== 普通对话")
t = time.time(); print("".join(llm.response("t", [sysmsg, {"role": "user", "content": "你好，介绍一下你自己"}])), f"({time.time()-t:.1f}s)")
tools = [{"type": "function", "function": {"name": "get_weather", "description": "查询某个城市的天气",
          "parameters": {"type": "object", "properties": {"location": {"type": "string"}}, "required": ["location"]}}}]
print("== 工具调用")
for text, calls in llm.response_with_functions("t", [sysmsg, {"role": "user", "content": "北京今天天气怎么样"}], functions=tools):
    if text: print("念出来:", repr(text))
    if calls: print("工具:", [(c.function.name, c.function.arguments) for c in calls])
print("== 工具结果之后")
d = [sysmsg, {"role": "user", "content": "北京今天天气怎么样"},
     {"role": "assistant", "content": "", "tool_calls": [{"id": "1", "type": "function", "function": {"name": "get_weather", "arguments": "{\"location\": \"北京\"}"}}]},
     {"role": "tool", "tool_call_id": "1", "content": "北京：晴，12~22 度，北风 2 级"}]
print("".join(t or "" for t, _ in llm.response_with_functions("t", d, functions=tools)))

python test_cli_llm.py fallback

我们在一台 Mac 上实测的输出（Claude 订阅，Haiku 5.5）：

注意工具调用那一段：模型输出的 <tool_call> 一个字都没被念出来，直接变成了 get_weather 调用。

在设备上测

1. 普通对话：对小智说"你好，介绍一下你自己"。日志里应该出现 claude_cli 首字 … ms。
2. 工具调用：说"今天北京天气怎么样"或"放一首歌"（需要在 Intent.function_call.functions 里开了对应插件）。日志里应该出现"检测到 1 个工具调用"，小智不会把 <tool_call> 念出来。
3. 自动切换：在 claude_cli 下面临时加一行 bin: claude-not-here（一个不存在的命令），重启再问一句。日志会出现：

WARNING claude_cli 出错，冷却 300 秒：[Errno 2] No such file or directory: 'claude-not-here'
INFO codex_cli 首字 … ms，总耗时 … ms

回答照常出来，只是换成了 Codex。测完把这一行删掉。

看延迟

grep -E "首字|冷却" tmp/server.log | tail -20

• 首字延迟：从发出请求到拿到第一个字。小智拿到第一句就开始合成语音，所以这是用户等待感最强的那段。
• 总耗时：整段回答写完。

我们实测 Claude（Haiku 5.5）首字约 1.1～2.3 秒，总耗时 2～3 秒。这里面有 1 秒左右是每次启动 CLI 进程的开销，比直接调 API 慢一些；用户的总等待还要加上语音识别和语音合成的时间。Codex 不逐字输出，延迟一般更高，适合当备用。

常见问题

• 日志里是"退出码 1，没有输出"：多半是 CLI 没登录或登录过期。在同一个系统用户下手动运行一次 claude 或 codex login。服务端用别的用户（比如 systemd 服务）跑时，那个用户也得登录过。
• 报 No such file or directory: 'claude'：服务端进程的 PATH 里找不到命令。在配置里用 bin: 写完整路径，比如 bin: /Users/你/.local/bin/claude（用 which claude 查）。
• 小智把 <tool_call> 念出来了：说明这一轮走的是 response 而不是 response_with_functions。检查 selected_module.Intent 是不是 function_call。
• 回答太长：在小智的角色提示词里写清"回答不超过两句话"。语音场景越短越好。
• 订阅额度用完：CLI 会报错，fallback 自动切到下一个，冷却 300 秒后再回来试。

参考

• xiaozhi-esp32-server：小智服务端
• xiaozhi-esp32：小智固件
• Claude Code 命令行参数
• Codex CLI