介绍
一个基于 IBus 的大模型拼音输入法。用户输入拼音后按空格,输入法通过 OpenAI-compatible Chat Completions 接口请求中文候选,再由用户选择候选提交到当前输入框。
当前实现偏 MVP:IBus 负责按键捕获、候选窗、提交文本和缓存;大模型负责把拼音转换成中文候选。详细设计见 design.md
。
特性
- 支持 OpenAI-compatible Chat Completions API。
- 支持本地 llama.cpp、Ollama、DeepSeek、OpenRouter 等兼容服务。
- 拼音输入期间不把原始拼音写入当前输入框,只在 IBus 弹出区域显示输入内容和候选。
- 用户选择候选后才提交中文到输入框。
- 焦点切换时自动清空原始拼音、候选和辅助文本。
- 支持快捷键切换中英文输入模式,默认
Ctrl+Space。 - IBus 状态栏/面板会显示当前输入模式:
中或英。 - 支持 SQLite 候选缓存,常用候选会被提升排序。
- 支持导入
.dict.json自定义领域词库,领域词会作为 LLM 上下文参与长拼音纠错。 - 拼音输入过程中输入 ASCII 标点或符号时,不会退出输入,会把符号连同拼音一起发送给 LLM。
- 未开始输入拼音时数字键直通应用;已经开始输入拼音后数字会进入缓冲区,不会中断输入。
- 中文模式下开启 CapsLock 时,英文按键会直接交给当前应用。
- 模型失败或超时时不会阻塞输入,可按回车提交原始拼音。
- 支持本地常用词兜底候选。
- Ctrl、Alt、Super、Meta 快捷键以及功能键、方向键等会直通应用,避免被输入法屏蔽。
项目地址
volsifly/ibus_llm_pinyin_input: ibus使用大模型输入拼音
设计思路
Ubuntu 下基于 IBus 的大模型拼音输入法开发文档
1. 项目目标
本项目目标是在 Ubuntu 桌面环境下开发一个基于 IBus 框架的输入法引擎。用户直接输入拼音,输入法在用户按下触发键后调用大模型,将拼音转换为中文候选项,并允许用户选择候选后提交到当前输入框。
本方案不依赖传统拼音词库,不使用 Rime / Fcitx / libpinyin 作为主转换引擎,而是直接使用大模型完成“拼音到中文”的转换。为了便于模型切换,请求接口采用 OpenAI API 的 Chat Completions 兼容格式,支持配置不同厂商、不同本地模型服务或远程模型服务。
示例交互:
用户输入:nihao
按下空格
候选:1. 你好 2. 你号 3. 倪浩
按下 1 或空格
提交:你好
长句示例:
用户输入:wo yao xie yi pian guan yu ai shuru fa de wen zhang
按下空格
候选:
1. 我要写一篇关于 AI 输入法的文章
2. 我要写一篇关于爱输入法的文章
3. 我要写一篇关于 AI 输入法的文档
2. 设计原则
这个输入法不是传统拼音输入法的复刻,而是一个“大模型驱动的拼音转中文输入法”。因此设计重点与传统输入法不同。
第一,输入法逻辑尽量简单。IBus 负责捕获按键、显示预编辑文本、显示候选项、提交文本;大模型负责根据拼音生成中文候选。
第二,请求接口必须可配置。不同用户可能使用 OpenAI、通义千问兼容接口、智谱兼容接口、本地 llama.cpp server、本地 Ollama OpenAI 兼容接口等,因此不能把模型服务写死。
第三,模型调用必须低延迟。输入法是高频交互工具,不能每输入一个字母就请求大模型。MVP 版本只在用户按下空格、回车或指定触发键时调用模型。
第四,必须有缓存。相同拼音的结果应该优先走本地缓存,不应该反复请求模型。用户选择过的候选也应该提升权重。
第五,错误时不能阻塞输入。模型服务不可用、超时、返回格式错误时,输入法应该允许用户提交原始拼音,避免用户无法继续打字。
3. 技术选型
3.1 输入法框架
使用 IBus。
原因:
Ubuntu / GNOME 默认集成较好
Python 原型开发速度快
支持 preedit、lookup table、candidate、commit text 等输入法核心能力
适合快速验证大模型输入法逻辑
3.2 开发语言
使用 Python 3。
原因:
IBus Python GI 绑定可用
调用 HTTP API 简单
方便读取 JSON 配置
方便接 SQLite 缓存
原型迭代成本低
3.3 大模型接口
使用 OpenAI-compatible Chat Completions 接口。
默认请求形式:
POST {base_url}/chat/completions
Authorization: Bearer {api_key}
Content-Type: application/json
请求体:
{
"model": "qwen3-0.6b",
"messages": [
{
"role": "system",
"content": "你是一个拼音输入法转换器,只输出 JSON 数组。"
},
{
"role": "user",
"content": "把下面的拼音转换成中文候选:nihao"
}
],
"temperature": 0.1,
"max_tokens": 64,
"stream": false
}
响应解析路径:
choices[0].message.content
模型应返回:
["你好", "你号", "倪浩"]
4. 总体架构
┌──────────────────────────┐
│ 当前应用输入框 │
│ GTK / Qt / Chrome / VSCode │
└─────────────▲────────────┘
│ commit_text
┌─────────────┴────────────┐
│ IBus AI Pinyin Engine │
│ │
│ - 捕获键盘事件 │
│ - 维护拼音 buffer │
│ - 更新 preedit │
│ - 展示候选词 lookup table │
│ - 提交候选 │
└─────────────▲────────────┘
│ candidates
┌─────────────┴────────────┐
│ Candidate Manager │
│ │
│ - 查缓存 │
│ - 调用 LLM Client │
│ - 解析 JSON │
│ - 排序和去重 │
└─────────────▲────────────┘
│ HTTP
┌─────────────┴────────────┐
│ OpenAI-compatible API │
│ │
│ - OpenAI │
│ - llama.cpp server │
│ - Ollama compatible API │
│ - vLLM │
│ - 自定义模型服务 │
└──────────────────────────┘
5. 功能范围
5.1 MVP 功能
MVP 版本需要实现以下功能:
输入英文字母,进入拼音 buffer
实时显示 preedit 拼音
按空格触发大模型转换
显示候选词列表
按数字键 1-9 选择候选
按空格提交第一个候选
按回车提交第一个候选或原始拼音
按 Esc 清空 buffer
按 Backspace 删除 buffer 最后一个字符
模型失败时提交原始拼音
支持配置 base_url、api_key、model
支持 SQLite 缓存
5.2 非 MVP 功能
以下能力不建议第一版实现:
实时逐字请求模型
复杂分词
云端用户词库同步
拼音纠错
模糊音
双拼
上下文跨应用记忆
复杂配置 GUI
这些功能可以在 MVP 稳定后逐步增加。
6. 系统依赖
6.1 Ubuntu 依赖
sudo apt update
sudo apt install -y \
ibus \
python3 \
python3-gi \
gir1.2-ibus-1.0 \
python3-requests \
sqlite3
如果系统没有启用 IBus,可执行:
im-config -n ibus
然后注销重新登录。
6.2 Python 依赖
为了降低安装复杂度,MVP 可以只使用系统包:
python3-gi
requests
sqlite3 标准库
json 标准库
threading 标准库
queue 标准库
如果后续需要更复杂配置,可以加入:
pip install pydantic pyyaml
但第一版不建议引入太多依赖。
7. 项目目录结构
建议使用用户级安装,不污染系统目录。
~/.local/share/ibus-ai-pinyin/
├── engine.py # IBus 输入法主程序
├── llm_client.py # OpenAI-compatible API 请求封装
├── config.py # 配置读取
├── cache.py # SQLite 缓存
├── prompts.py # Prompt 模板
├── utils.py # 工具函数
└── README.md
~/.local/share/ibus/component/
└── ai-pinyin.xml # IBus component 注册文件
~/.config/ibus-ai-pinyin/
├── config.json # 用户配置
└── cache.sqlite3 # 候选缓存
8. 配置文件设计
配置文件路径:
~/.config/ibus-ai-pinyin/config.json
示例配置:
{
"api": {
"base_url": "http://127.0.0.1:8080/v1",
"api_key": "",
"api_key_env": "OPENAI_API_KEY",
"model": "qwen3-0.6b",
"endpoint": "/chat/completions",
"timeout_ms": 800,
"temperature": 0.1,
"top_p": 0.8,
"max_tokens": 64,
"stream": false
},
"input": {
"max_buffer_length": 120,
"trigger_keys": ["space"],
"commit_first_candidate_with_space": true,
"enter_commit_raw_when_no_candidate": true,
"candidate_page_size": 5
},
"candidate": {
"max_candidates": 5,
"deduplicate": true,
"allow_ascii": false,
"fallback_to_raw_pinyin": true
},
"cache": {
"enabled": true,
"path": "~/.config/ibus-ai-pinyin/cache.sqlite3",
"ttl_days": 30,
"promote_user_selection": true
},
"prompt": {
"system": "你是一个中文拼音输入法转换器。你的任务是把用户输入的拼音转换成最可能的中文候选。只输出 JSON 字符串数组,不要解释,不要 Markdown,不要代码块。最多输出 5 个候选。",
"user_template": "拼音:{pinyin}\n请输出中文候选 JSON 数组。"
}
}
8.1 本地 llama.cpp server 配置示例
{
"api": {
"base_url": "http://127.0.0.1:8080/v1",
"api_key": "sk-local",
"model": "qwen3-0.6b",
"endpoint": "/chat/completions",
"timeout_ms": 800,
"temperature": 0.1,
"max_tokens": 64,
"stream": false
}
}
8.2 Ollama OpenAI 兼容接口配置示例
{
"api": {
"base_url": "http://127.0.0.1:11434/v1",
"api_key": "ollama",
"model": "qwen3:0.6b",
"endpoint": "/chat/completions",
"timeout_ms": 1200,
"temperature": 0.1,
"max_tokens": 64,
"stream": false
}
}
8.3 远程 OpenAI-compatible 服务配置示例
{
"api": {
"base_url": "https://api.example.com/v1",
"api_key_env": "OPENAI_API_KEY",
"model": "your-fast-model",
"endpoint": "/chat/completions",
"timeout_ms": 1000,
"temperature": 0.1,
"max_tokens": 64,
"stream": false
}
}
9. Prompt 设计
输入法场景的 Prompt 必须短、硬、稳定,不要让模型自由发挥。
推荐 system prompt:
你是一个中文拼音输入法转换器。
你的任务是把用户输入的拼音转换成最可能的中文候选。
只输出 JSON 字符串数组,不要解释,不要 Markdown,不要代码块。
最多输出 5 个候选。
候选必须是中文短语或中文句子。
不要输出拼音本身。
不要输出英文,除非拼音明显表示 AI、API、CPU、GPU、IBus、Linux 等技术词。
推荐 user prompt:
拼音:{pinyin}
请输出中文候选 JSON 数组。
示例:
拼音:wo yao xie yi pian guan yu ai shuru fa de wenzhang
请输出中文候选 JSON 数组。
模型输出:
["我要写一篇关于 AI 输入法的文章", "我要写一篇关于爱输入法的文章"]
10. HTTP 请求封装设计
10.1 请求地址拼接
配置项:
{
"base_url": "http://127.0.0.1:8080/v1",
"endpoint": "/chat/completions"
}
最终 URL:
http://127.0.0.1:8080/v1/chat/completions
10.2 请求头
Authorization: Bearer {api_key}
Content-Type: application/json
如果 api_key 为空,可以仍然传一个本地占位值,例如:
sk-local
部分本地 OpenAI-compatible 服务不校验 key,但仍要求 Authorization header 存在。
10.3 请求体
{
"model": "qwen3-0.6b",
"messages": [
{
"role": "system",
"content": "你是一个中文拼音输入法转换器。只输出 JSON 字符串数组。"
},
{
"role": "user",
"content": "拼音:nihao\n请输出中文候选 JSON 数组。"
}
],
"temperature": 0.1,
"top_p": 0.8,
"max_tokens": 64,
"stream": false
}
10.4 响应格式
OpenAI-compatible Chat Completions 常见响应:
{
"id": "chatcmpl-xxx",
"object": "chat.completion",
"created": 1710000000,
"model": "qwen3-0.6b",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": "[\"你好\", \"你号\", \"倪浩\"]"
},
"finish_reason": "stop"
}
]
}
解析路径:
content = data["choices"][0]["message"]["content"]
然后对 content 做 JSON 解析。
10.5 返回容错
模型有时可能返回:
```json
["你好", "你号"]
或者:
```text
候选:["你好", "你号"]
因此解析器需要做容错:
去除 Markdown 代码块
提取第一个 [ 到最后一个 ] 之间的内容
尝试 json.loads
过滤非字符串元素
去重
限制候选数量
11. 缓存设计
11.1 为什么必须有缓存
输入法是高频操作。如果每次输入同一个拼音都请求模型,体验会很差,也会增加成本。
缓存解决三个问题:
提升速度
降低模型调用成本
让用户选择过的候选越来越靠前
11.2 SQLite 表结构
CREATE TABLE IF NOT EXISTS candidates (
pinyin TEXT NOT NULL,
candidate TEXT NOT NULL,
score INTEGER NOT NULL DEFAULT 0,
source TEXT NOT NULL DEFAULT 'llm',
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
PRIMARY KEY (pinyin, candidate)
);
CREATE INDEX IF NOT EXISTS idx_candidates_pinyin_score
ON candidates (pinyin, score DESC, updated_at DESC);
11.3 查询逻辑
SELECT candidate
FROM candidates
WHERE pinyin = ?
ORDER BY score DESC, updated_at DESC
LIMIT ?;
11.4 写入逻辑
模型返回候选后写入缓存:
INSERT INTO candidates (pinyin, candidate, score, source, created_at, updated_at)
VALUES (?, ?, 0, 'llm', ?, ?)
ON CONFLICT(pinyin, candidate)
DO UPDATE SET updated_at = excluded.updated_at;
用户选择候选后提升分数:
UPDATE candidates
SET score = score + 10,
updated_at = ?
WHERE pinyin = ? AND candidate = ?;
12. IBus 引擎状态机
输入法内部维护以下状态:
buffer: str # 当前拼音输入
candidates: list[str] # 当前候选
candidate_visible: bool # 候选窗口是否显示
selected_index: int # 当前选中候选
is_requesting: bool # 是否正在请求模型
状态流转:
空闲状态
↓ 输入字母
拼音编辑状态
↓ 空格
请求候选状态
↓ 模型返回
候选选择状态
↓ 数字键 / 空格 / 回车
提交文本并回到空闲状态
异常流转:
请求候选状态
↓ 超时 / 异常 / 空结果
候选失败状态
↓ 回车
提交原始拼音
13. 按键设计
| 按键 | 行为 |
|---|---|
| a-z | 加入拼音 buffer |
| ’ | 作为拼音分隔符加入 buffer,可选 |
| 空格 | 有候选时提交当前候选;无候选时触发模型转换 |
| 回车 | 有候选时提交当前候选;无候选时提交原始拼音 |
| 数字 1-9 | 选择对应候选并提交 |
| Backspace | 删除 buffer 最后一个字符;如果候选显示则先返回编辑状态 |
| Esc | 清空 buffer 和候选 |
| PageUp/PageDown | 候选翻页,后续版本实现 |
| 左右方向键 | 候选选择,后续版本实现 |
MVP 可以先不实现复杂翻页,只显示最多 5 个候选。
14. 候选展示设计
IBus 中使用 IBus.LookupTable 展示候选。
候选格式:
1. 你好
2. 你号
3. 倪浩
在 IBus LookupTable 里,候选文本只放中文,不需要手动拼数字序号。IBus UI 会根据当前环境显示候选列表。
候选生成后执行:
self.update_lookup_table(table, True)
清空候选时执行:
self.hide_lookup_table()
15. 异步请求设计
不能在 do_process_key_event 中同步请求模型,否则 UI 会卡住。
错误示例:
def do_process_key_event(...):
candidates = call_llm(self.buffer) # 阻塞 UI,不推荐
正确做法:
按空格
↓
启动后台线程请求模型
↓
立即返回 True
↓
preedit 显示“nihao ...”或“正在转换”
↓
后台线程拿到结果
↓
通过 GLib.idle_add 回到主线程更新候选
示例:
threading.Thread(
target=self.fetch_candidates_worker,
args=(self.buffer,),
daemon=True
).start()
回到 IBus 主线程:
GLib.idle_add(self.on_candidates_ready, pinyin, candidates)
16. 核心代码骨架
16.1 llm_client.py
import json
import os
import re
import requests
class LLMClient:
def __init__(self, config):
self.config = config
api = config.get("api", {})
self.base_url = api.get("base_url", "http://127.0.0.1:8080/v1").rstrip("/")
self.endpoint = api.get("endpoint", "/chat/completions")
self.model = api.get("model", "qwen3-0.6b")
self.timeout = api.get("timeout_ms", 800) / 1000
self.temperature = api.get("temperature", 0.1)
self.top_p = api.get("top_p", 0.8)
self.max_tokens = api.get("max_tokens", 64)
self.stream = api.get("stream", False)
api_key = api.get("api_key", "")
api_key_env = api.get("api_key_env", "OPENAI_API_KEY")
self.api_key = api_key or os.environ.get(api_key_env, "sk-local")
prompt = config.get("prompt", {})
self.system_prompt = prompt.get(
"system",
"你是一个中文拼音输入法转换器。只输出 JSON 字符串数组。"
)
self.user_template = prompt.get(
"user_template",
"拼音:{pinyin}\n请输出中文候选 JSON 数组。"
)
def get_candidates(self, pinyin: str, max_candidates: int = 5):
url = self.base_url + self.endpoint
body = {
"model": self.model,
"messages": [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": self.user_template.format(pinyin=pinyin)}
],
"temperature": self.temperature,
"top_p": self.top_p,
"max_tokens": self.max_tokens,
"stream": self.stream
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
resp = requests.post(url, headers=headers, json=body, timeout=self.timeout)
resp.raise_for_status()
data = resp.json()
content = data["choices"][0]["message"]["content"]
return self.parse_candidates(content, max_candidates=max_candidates)
def parse_candidates(self, content: str, max_candidates: int = 5):
text = content.strip()
text = re.sub(r"^```json\s*", "", text)
text = re.sub(r"^```\s*", "", text)
text = re.sub(r"\s*```$", "", text)
start = text.find("[")
end = text.rfind("]")
if start >= 0 and end > start:
text = text[start:end + 1]
arr = json.loads(text)
if not isinstance(arr, list):
return []
result = []
seen = set()
for item in arr:
if not isinstance(item, str):
continue
item = item.strip()
if not item:
continue
if item in seen:
continue
seen.add(item)
result.append(item)
if len(result) >= max_candidates:
break
return result
16.2 cache.py
import os
import sqlite3
import time
class CandidateCache:
def __init__(self, path: str):
self.path = os.path.expanduser(path)
os.makedirs(os.path.dirname(self.path), exist_ok=True)
self.conn = sqlite3.connect(self.path, check_same_thread=False)
self.init_db()
def init_db(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS candidates (
pinyin TEXT NOT NULL,
candidate TEXT NOT NULL,
score INTEGER NOT NULL DEFAULT 0,
source TEXT NOT NULL DEFAULT 'llm',
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
PRIMARY KEY (pinyin, candidate)
)
""")
self.conn.execute("""
CREATE INDEX IF NOT EXISTS idx_candidates_pinyin_score
ON candidates (pinyin, score DESC, updated_at DESC)
""")
self.conn.commit()
def get(self, pinyin: str, limit: int = 5):
cur = self.conn.execute(
"""
SELECT candidate
FROM candidates
WHERE pinyin = ?
ORDER BY score DESC, updated_at DESC
LIMIT ?
""",
(pinyin, limit)
)
return [row[0] for row in cur.fetchall()]
def put_many(self, pinyin: str, candidates: list[str], source: str = "llm"):
now = int(time.time())
for candidate in candidates:
self.conn.execute(
"""
INSERT INTO candidates (pinyin, candidate, score, source, created_at, updated_at)
VALUES (?, ?, 0, ?, ?, ?)
ON CONFLICT(pinyin, candidate)
DO UPDATE SET updated_at = excluded.updated_at
""",
(pinyin, candidate, source, now, now)
)
self.conn.commit()
def promote(self, pinyin: str, candidate: str):
now = int(time.time())
self.conn.execute(
"""
UPDATE candidates
SET score = score + 10,
updated_at = ?
WHERE pinyin = ? AND candidate = ?
""",
(now, pinyin, candidate)
)
self.conn.commit()
16.3 config.py
import json
import os
DEFAULT_CONFIG = {
"api": {
"base_url": "http://127.0.0.1:8080/v1",
"api_key": "",
"api_key_env": "OPENAI_API_KEY",
"model": "qwen3-0.6b",
"endpoint": "/chat/completions",
"timeout_ms": 800,
"temperature": 0.1,
"top_p": 0.8,
"max_tokens": 64,
"stream": False
},
"input": {
"max_buffer_length": 120,
"candidate_page_size": 5
},
"candidate": {
"max_candidates": 5,
"fallback_to_raw_pinyin": True
},
"cache": {
"enabled": True,
"path": "~/.config/ibus-ai-pinyin/cache.sqlite3"
},
"prompt": {
"system": "你是一个中文拼音输入法转换器。你的任务是把用户输入的拼音转换成最可能的中文候选。只输出 JSON 字符串数组,不要解释,不要 Markdown,不要代码块。最多输出 5 个候选。",
"user_template": "拼音:{pinyin}\n请输出中文候选 JSON 数组。"
}
}
def deep_merge(base, override):
result = dict(base)
for key, value in override.items():
if isinstance(value, dict) and isinstance(result.get(key), dict):
result[key] = deep_merge(result[key], value)
else:
result[key] = value
return result
def load_config():
path = os.path.expanduser("~/.config/ibus-ai-pinyin/config.json")
if not os.path.exists(path):
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
json.dump(DEFAULT_CONFIG, f, ensure_ascii=False, indent=2)
return DEFAULT_CONFIG
with open(path, "r", encoding="utf-8") as f:
user_config = json.load(f)
return deep_merge(DEFAULT_CONFIG, user_config)
16.4 engine.py 核心骨架
#!/usr/bin/env python3
import threading
import gi
gi.require_version("IBus", "1.0")
from gi.repository import IBus, GLib
from config import load_config
from llm_client import LLMClient
from cache import CandidateCache
class AIPinyinEngine(IBus.Engine):
def __init__(self, bus, object_path):
super().__init__(connection=bus.get_connection(), object_path=object_path)
self.config = load_config()
self.llm = LLMClient(self.config)
cache_cfg = self.config.get("cache", {})
self.cache_enabled = cache_cfg.get("enabled", True)
self.cache = CandidateCache(cache_cfg.get("path", "~/.config/ibus-ai-pinyin/cache.sqlite3"))
self.buffer = ""
self.candidates = []
self.selected_index = 0
self.is_requesting = False
def do_process_key_event(self, keyval, keycode, state):
if state & IBus.ModifierType.RELEASE_MASK:
return False
# 数字键选候选
if self.candidates and IBus.KEY_1 <= keyval <= IBus.KEY_9:
index = keyval - IBus.KEY_1
if index < len(self.candidates):
self.commit_candidate(index)
return True
if keyval == IBus.KEY_space:
if self.candidates:
self.commit_candidate(self.selected_index)
return True
if self.buffer:
self.request_candidates()
return True
return False
if keyval == IBus.KEY_Return:
if self.candidates:
self.commit_candidate(self.selected_index)
return True
if self.buffer:
self.commit_raw()
return True
return False
if keyval == IBus.KEY_Escape:
self.clear_all()
return True
if keyval == IBus.KEY_BackSpace:
if self.candidates:
self.candidates = []
self.hide_lookup_table()
self.update_preedit()
return True
if self.buffer:
self.buffer = self.buffer[:-1]
self.update_preedit()
return True
return False
code = IBus.keyval_to_unicode(keyval)
if code == 0:
return False
ch = chr(code)
if self.accept_char(ch):
max_len = self.config.get("input", {}).get("max_buffer_length", 120)
if len(self.buffer) < max_len:
self.buffer += ch.lower()
self.candidates = []
self.hide_lookup_table()
self.update_preedit()
return True
return False
def accept_char(self, ch: str):
return ch.isascii() and (ch.isalpha() or ch in ["'", " "])
def update_preedit(self, suffix: str = ""):
text = self.buffer + suffix
ibus_text = IBus.Text.new_from_string(text)
self.update_preedit_text(ibus_text, len(text), bool(text))
def request_candidates(self):
if self.is_requesting:
return
pinyin = " ".join(self.buffer.split())
max_candidates = self.config.get("candidate", {}).get("max_candidates", 5)
if self.cache_enabled:
cached = self.cache.get(pinyin, limit=max_candidates)
if cached:
self.show_candidates(pinyin, cached)
return
self.is_requesting = True
self.update_preedit(" …")
threading.Thread(
target=self.fetch_candidates_worker,
args=(pinyin, max_candidates),
daemon=True
).start()
def fetch_candidates_worker(self, pinyin: str, max_candidates: int):
try:
candidates = self.llm.get_candidates(pinyin, max_candidates=max_candidates)
except Exception:
candidates = []
GLib.idle_add(self.on_candidates_ready, pinyin, candidates)
def on_candidates_ready(self, pinyin: str, candidates: list[str]):
self.is_requesting = False
# 如果用户已经修改了 buffer,丢弃旧请求结果
current = " ".join(self.buffer.split())
if current != pinyin:
return False
if candidates:
if self.cache_enabled:
self.cache.put_many(pinyin, candidates)
self.show_candidates(pinyin, candidates)
else:
self.update_preedit()
return False
def show_candidates(self, pinyin: str, candidates: list[str]):
self.candidates = candidates
self.selected_index = 0
table = IBus.LookupTable.new(
page_size=self.config.get("input", {}).get("candidate_page_size", 5),
cursor_pos=0,
cursor_visible=True,
round=True
)
for cand in candidates:
table.append_candidate(IBus.Text.new_from_string(cand))
self.update_lookup_table(table, True)
self.update_preedit()
def commit_candidate(self, index: int):
if not self.candidates or index >= len(self.candidates):
return
pinyin = " ".join(self.buffer.split())
text = self.candidates[index]
self.commit_text(IBus.Text.new_from_string(text))
if self.cache_enabled:
self.cache.promote(pinyin, text)
self.clear_all()
def commit_raw(self):
self.commit_text(IBus.Text.new_from_string(self.buffer))
self.clear_all()
def clear_all(self):
self.buffer = ""
self.candidates = []
self.selected_index = 0
self.is_requesting = False
self.update_preedit()
self.hide_lookup_table()
def do_focus_out(self):
self.clear_all()
class EngineFactory(IBus.Factory):
def __init__(self, bus):
super().__init__(connection=bus.get_connection())
self.bus = bus
self.engine_id = 0
def do_create_engine(self, engine_name):
self.engine_id += 1
object_path = f"/org/freedesktop/IBus/Engine/AIPinyin/{self.engine_id}"
return AIPinyinEngine(self.bus, object_path)
def main():
IBus.init()
bus = IBus.Bus()
factory = EngineFactory(bus)
bus.request_name("org.freedesktop.IBus.AIPinyin", 0)
loop = GLib.MainLoop()
loop.run()
if __name__ == "__main__":
main()
17. IBus Component 注册文件
路径:
~/.local/share/ibus/component/ai-pinyin.xml
内容示例:
<?xml version="1.0" encoding="utf-8"?>
<component>
<name>org.freedesktop.IBus.AIPinyin</name>
<description>AI Pinyin Input Method</description>
<exec>/home/YOUR_USERNAME/.local/share/ibus-ai-pinyin/engine.py</exec>
<version>0.1.0</version>
<author>Volsifly</author>
<license>MIT</license>
<homepage>https://example.com</homepage>
<textdomain>ibus-ai-pinyin</textdomain>
<engines>
<engine>
<name>ai-pinyin</name>
<language>zh_CN</language>
<license>MIT</license>
<author>Volsifly</author>
<icon>ibus-keyboard</icon>
<layout>us</layout>
<longname>AI 拼音输入法</longname>
<description>使用大模型将拼音转换为中文的 IBus 输入法</description>
<rank>80</rank>
</engine>
</engines>
</component>
注意替换:
/home/YOUR_USERNAME
可以通过命令查看用户名:
whoami
18. 安装流程
18.1 创建目录
mkdir -p ~/.local/share/ibus-ai-pinyin
mkdir -p ~/.local/share/ibus/component
mkdir -p ~/.config/ibus-ai-pinyin
18.2 放置代码
将以下文件放入:
~/.local/share/ibus-ai-pinyin/engine.py
~/.local/share/ibus-ai-pinyin/llm_client.py
~/.local/share/ibus-ai-pinyin/cache.py
~/.local/share/ibus-ai-pinyin/config.py
增加执行权限:
chmod +x ~/.local/share/ibus-ai-pinyin/engine.py
18.3 注册 IBus Component
将 ai-pinyin.xml 放入:
~/.local/share/ibus/component/ai-pinyin.xml
18.4 重启 IBus
ibus restart
如果没有出现输入法,注销重新登录。
18.5 添加输入法
执行:
ibus-setup
然后添加:
Chinese -> AI 拼音输入法
也可以在 GNOME 设置中进入:
Settings -> Keyboard -> Input Sources
添加对应输入源。
19. 本地模型服务启动示例
19.1 llama.cpp server 示例
假设使用本地 GGUF 模型:
llama-server \
-m ./models/qwen3-0.6b-q4_k_m.gguf \
--host 127.0.0.1 \
--port 8080 \
-c 512 \
-n 64
输入法配置:
{
"api": {
"base_url": "http://127.0.0.1:8080/v1",
"api_key": "sk-local",
"model": "qwen3-0.6b",
"endpoint": "/chat/completions",
"timeout_ms": 800,
"temperature": 0.1,
"max_tokens": 64,
"stream": false
}
}
19.2 Ollama 示例
启动模型:
ollama run qwen3:0.6b
输入法配置:
{
"api": {
"base_url": "http://127.0.0.1:11434/v1",
"api_key": "ollama",
"model": "qwen3:0.6b",
"endpoint": "/chat/completions",
"timeout_ms": 1200,
"temperature": 0.1,
"max_tokens": 64,
"stream": false
}
}
20. 测试命令
在接入 IBus 前,先测试模型接口。
curl http://127.0.0.1:8080/v1/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer sk-local" \
-d '{
"model": "qwen3-0.6b",
"messages": [
{"role": "system", "content": "你是一个中文拼音输入法转换器。只输出 JSON 字符串数组。"},
{"role": "user", "content": "拼音:nihao\n请输出中文候选 JSON 数组。"}
],
"temperature": 0.1,
"max_tokens": 64,
"stream": false
}'
预期返回 content 中包含:
["你好", "你号", "倪浩"]
21. 日志设计
MVP 阶段建议写入用户目录:
~/.cache/ibus-ai-pinyin/engine.log
日志内容包括:
启动时间
配置加载结果,不记录 api_key
模型请求耗时
模型返回候选数量
解析失败原因
HTTP 错误码
IBus 异常
注意不要记录完整用户输入内容,至少默认不要记录。输入法输入内容可能包含隐私信息。
22. 隐私与安全
这个项目涉及输入法,隐私风险非常高。必须明确设计边界。
22.1 默认推荐本地模型
如果使用远程 API,用户输入的拼音内容会被发送到远程服务。虽然只是拼音,但仍可能还原出敏感文本。
因此默认推荐:
本地 llama.cpp server
本地 Ollama
本地 vLLM
局域网私有模型服务
22.2 不记录敏感内容
日志中不要默认记录用户输入的拼音和模型返回结果。
可以配置:
{
"debug": {
"log_user_input": false,
"log_model_output": false
}
}
22.3 API Key 处理
不要把 API Key 写入日志。
推荐优先使用环境变量:
export OPENAI_API_KEY="sk-xxx"
配置中只写:
{
"api": {
"api_key_env": "OPENAI_API_KEY"
}
}
23. 性能优化建议
23.1 只在触发时请求模型
不要逐字符请求。
错误:n、ni、nih、niha、nihao 每次都请求
正确:用户输入完整拼音,按空格后请求
23.2 控制 max_tokens
输入法候选通常很短,建议:
短词:32
长句:64
最多不要超过 128
23.3 控制 timeout
本地模型:
500ms - 1200ms
远程模型:
1000ms - 2000ms
超过超时就回退,不要卡住输入法。
23.4 使用缓存
缓存命中时应该在几十毫秒内返回。
23.5 模型常驻内存
不要每次请求时启动模型。模型服务必须常驻。
24. 错误处理
| 场景 | 处理方式 |
|---|---|
| 模型服务未启动 | 不显示候选,允许回车提交原始拼音 |
| HTTP 超时 | 清除请求状态,恢复 preedit |
| JSON 解析失败 | 尝试提取数组;失败则回退 |
| 候选为空 | 回退原始拼音 |
| 用户请求中途修改 buffer | 丢弃旧请求结果 |
| API Key 错误 | 日志记录 HTTP 状态码,不在 UI 阻塞 |
| IBus focus out | 清空 buffer 和候选 |
25. 开发里程碑
25.1 第一阶段:最小可运行版本
目标:能输入拼音,按空格调用模型,提交第一个候选。
任务:
完成 IBus 注册
完成 buffer 和 preedit
完成 OpenAI-compatible 请求
完成候选解析
完成空格提交
25.2 第二阶段:候选选择
目标:能显示多个候选,并用数字键选择。
任务:
实现 IBus LookupTable
实现数字键 1-9 选择
实现候选去重
实现候选失败回退
25.3 第三阶段:缓存和学习
目标:常用输入越来越快。
任务:
接入 SQLite
缓存模型候选
用户选择后提升 score
缓存优先返回
25.4 第四阶段:体验优化
目标:更像真实输入法。
任务:
候选翻页
方向键选择
标点处理
中英文切换
状态栏提示
配置热重载
25.5 第五阶段:模型增强
目标:提升转换准确率。
任务:
加入上下文短记忆
区分短词和长句 prompt
加入技术词白名单
支持企业词汇提示
支持用户自定义固定短语
26. 推荐默认模型策略
虽然本开发文档允许自定义模型,但默认推荐选择 CPU 可运行的小模型。
推荐优先级:
1. Qwen3-0.6B Q4 / Q5:速度优先,适合 MVP
2. Qwen3-1.7B Q4:质量更好,但 CPU 延迟更高
3. 其他中文小模型:需要测试 JSON 稳定性和拼音转换能力
输入法场景不建议使用太大的模型,因为输入延迟比单次生成质量更重要。
27. 验收标准
MVP 版本验收:
Ubuntu 下能在 IBus 输入源中看到“AI 拼音输入法”
可以切换到该输入法
输入 nihao 按空格,能出现“你好”等候选
按数字键 1 可以提交第一个候选
输入长拼音句子,能返回中文句子候选
模型服务关闭时,输入法不会卡死
相同拼音第二次输入能命中缓存
日志不泄露 API Key
28. 已知限制
这个方案直接使用大模型实现拼音转中文,因此有以下限制:
首次输入延迟高于传统输入法
小模型可能输出不稳定 JSON
小模型可能产生错误候选
没有传统词库时,短词准确率不一定高
远程 API 存在隐私和网络延迟问题
CPU 运行模型时性能取决于机器配置
因此,长期产品化建议采用混合架构:
短词和高频词:本地词库
长句和补全:大模型
候选排序:用户缓存 + 模型重排
但本项目 MVP 按照要求,先实现“直接使用大模型”的版本。
29. 后续可扩展方向
29.1 加入本地词库兜底
即使主转换逻辑由大模型完成,也可以加入极小词库处理高频短词,提升速度。
29.2 加入企业词汇提示
在 prompt 中加入用户自定义词:
常用词:鸿灵、知识库、语义层、向量召回、工作流、智能体
这样模型对专业词会更稳定。
29.3 加入上下文
输入法可以记录最近一次提交的短上下文,但要注意隐私。
示例:
上文:今天我们讨论了输入法
拼音:zhe ge fang an keyi xian zuo mvp
候选:这个方案可以先做 MVP
29.4 加入 AI 指令模式
例如:
/zb
触发:
生成周报模板
或者:
/rw xiufu ibus houxuan chuang wenti
输出:
任务:修复 IBus 候选窗口问题
这会让输入法从“拼音输入法”变成“AI 办公输入入口”。
30. 总结
本开发方案的核心是:
Ubuntu + IBus + Python Engine + OpenAI-compatible API + 可配置模型 + SQLite 缓存
第一版不要追求传统输入法的完整能力,而是先验证一个关键问题:
用户输入拼音后,大模型能否在可接受延迟内返回可用中文候选。
只要这个链路跑通,后续就可以继续叠加词库、缓存、用户学习、企业术语、AI 指令模式和上下文补全,逐步把它演化成真正可用的 AI 输入法。