说实话，我一开始觉得写个AI助手挺难的。直到真的动手做了才发现，核心代码可能不到100行。

这篇文章就是想告诉你：门槛比你想象的低得多。

为什么想写这个

去年ChatGPT火了之后，我就在想：能不能自己搞一个？不是那种套壳的，而是真正能"做事"的助手。

后来发现了香港大学开源的nanobot项目，功能很强大，支持20多个平台。我基于这个项目自己部署了一套，现在它每天帮我自动发小红书、推新闻热点、还能语音转文字。

这篇文章把核心思路分享出来，帮你理解AI助手是怎么工作的。

最简单的AI助手长什么样

说白了就三件事：

1. 接收消息 - 用户说什么
2. 调用LLM - 让AI理解并决定做什么
3. 执行动作 - 调用工具、返回结果
   用代码表示大概是这样：

import asyncio

async def agent_loop():
    while True:
        # 1. 接收消息
        message = await get_message()
        
        # 2. 调用LLM
        response = await call_llm(message)
        
        # 3. 执行动作
        if response.needs_tool:
            result = await execute_tool(response.tool_name, response.tool_args)
            await send_message(result)
        else:
            await send_message(response.text)

就这么简单。当然，这是最简化的版本。但核心逻辑就是这样。

消息循环：asyncio是关键

我一开始用的是同步代码，结果发现一个问题：阻塞。

当你在调用LLM API的时候，整个程序都卡住了。用户发来新消息，根本收不到。

后来换成asyncio，问题迎刃而解：

import asyncio
from asyncio import Queue

# 消息队列
message_queue = Queue()

async def message_receiver():
    """接收消息，放入队列"""
    async for message in listen_for_messages():
        await message_queue.put(message)

async def message_processor():
    """从队列取消息，处理"""
    while True:
        message = await message_queue.get()
        response = await call_llm(message)
        await send_message(response)

# 同时运行
async def main():
    await asyncio.gather(
        message_receiver(),
        message_processor()
    )

asyncio.run(main())

这样，接收消息和处理消息可以同时进行。LLM调用再慢，也不会阻塞新消息的接收。

调用LLM：LiteLLM真香

一开始我直接用OpenAI的SDK，后来发现一个问题：模型切换太麻烦。

想换成Claude？改代码。想换成国产模型？改代码。想多个模型轮着用？更麻烦。

然后发现了LiteLLM这个库，一个API统一所有模型：

from litellm import completion

# OpenAI
response = completion(
    model="gpt-4",
    messages=[{"role": "user", "content": "你好"}]
)

# Claude
response = completion(
    model="claude-3-opus-20240229",
    messages=[{"role": "user", "content": "你好"}]
)

# 国产模型（比如智谱）
response = completion(
    model="zhipu/glm-4",
    messages=[{"role": "user", "content": "你好"}]
)

API格式完全一样，换模型就改个字符串。太香了。

工具系统：让AI能"动手"

光聊天没意思，我要的是能做事的助手。

OpenAI的Function Calling让这件事变得很简单：

# 定义工具
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称"
                    }
                },
                "required": ["city"]
            }
        }
    }
]

# 调用LLM时传入工具
response = completion(
    model="gpt-4",
    messages=messages,
    tools=tools
)

# LLM决定调用工具
if response.choices[0].message.tool_calls:
    tool_call = response.choices[0].message.tool_calls[0]
    tool_name = tool_call.function.name
    tool_args = json.loads(tool_call.function.arguments)
    
    # 执行工具
    if tool_name == "get_weather":
        result = get_weather(tool_args["city"])

LLM会自己决定什么时候调用工具、传什么参数。你只需要定义好工具，然后执行就行了。

踩坑记录

搞这个的过程中踩了不少坑，分享几个印象深的：

1. 消息历史爆炸

一开始我把所有消息都塞进context，结果token消耗飞快。后来加了滑动窗口，只保留最近N条消息。

MAX_HISTORY = 20
messages = messages[-MAX_HISTORY:]

2. 工具调用死循环

有一次LLM调用工具失败后，一直重试，陷入死循环。后来加了最大重试次数限制：

MAX_RETRIES = 3
for attempt in range(MAX_RETRIES):
    try:
        result = await execute_tool(...)
        break
    except Exception as e:
        if attempt == MAX_RETRIES - 1:
            result = "工具调用失败，请稍后重试"

3. 异步陷阱

asyncio的坑很多，比如在异步函数里用了time.sleep()而不是asyncio.sleep()，整个事件循环都卡住了。

# 错误
time.sleep(1)  # 阻塞整个事件循环

# 正确
await asyncio.sleep(1)  # 不阻塞

写在最后

这篇文章只是个入门，帮你搭建起AI助手的基本骨架。更复杂的功能——比如记忆系统、多渠道接入、定时任务——可以在这个基础上慢慢加。

如果你也想动手做一个，建议从最简单的开始：先跑通消息循环，再加LLM调用，最后加工具。一步一步来，别一上来就想搞个大而全的。

有问题欢迎评论区交流。下一篇打算聊聊LiteLLM的进阶用法，感兴趣的可以关注一下。

---

开源项目推荐：OpenClaw/nanobot - 香港大学开源的AI助手框架，支持20+平台，功能完善，欢迎star