1. 从零搭建个人AI助手:为什么第一阶段只做Model I/O
很多人一上来就想搞个全能助手,能查天气、能订日程、能写代码、能控制智能家居,结果三天后项目文件夹里堆了二十个半成品脚本,一个都跑不通。我踩过这个坑,所以这次做个人AI助手,阶段1只干两件事:把项目骨架搭稳,把模型输入输出这条最细的管道打通。听起来简单,但这两件事决定了后面所有功能是“长在水泥地上”还是“长在流沙上”。
所谓Model I/O,说白了就是你的代码怎么把一句话发给大模型,大模型怎么把结果还给你。这是整个AI助手最底层的呼吸动作。呼吸不顺,后面加多少工具调用、记忆管理、多轮对话都是白搭。我见过太多项目在阶段1就埋雷:API Key硬编码在脚本里、返回结果不做结构化解析、异常处理全靠print、流式输出和阻塞输出混着用。等到阶段3要接语音输入的时候,发现整个调用链根本没法复用。
这篇文章适合两类人:一是刚接触Agent开发、想从零搭一个个人助手的开发者;二是已经用过一些大模型API、但项目结构一直很乱、想重新整理一套干净底子的人。我会把阶段1拆成可复现的步骤,每一步都解释为什么这么做,以及我实际踩过的坑。代码以Python为主,但思路换成任何语言都通用。
2. 项目初始化:目录结构不是摆好看,是给未来的自己留后路
2.1 为什么我不建议用单文件起步
网上很多教程教你建一个main.py,五十行代码调通API就完事。学习可以,做项目不行。个人AI助手哪怕只做到阶段2,你也会面临:配置要分离、提示词要版本管理、模型调用要抽象成独立模块、日志要单独输出。如果一开始全塞在一个文件里,第一次重构就会让你想弃坑。
我的做法是建一个最小但可扩展的目录骨架。注意“最小”和“可扩展”不矛盾——只建你阶段1真正会用到的文件夹,但每个文件夹的职责边界要清晰。下面是我实际用的结构:
personal-ai-assistant/ ├── config/ │ ├── __init__.py │ └── settings.py ├── core/ │ ├── __init__.py │ ├── model_client.py │ └── message.py ├── prompts/ │ └── system_prompt.txt ├── logs/ │ └── .gitkeep ├── tests/ │ └── test_model_client.py ├── .env ├── .env.example ├── .gitignore └── main.py这个结构里每个东西都有明确用途。config/settings.py负责读取环境变量并暴露配置对象;core/model_client.py封装模型调用;core/message.py定义消息数据结构;prompts/放提示词文本,方便后续做版本对比;logs/存调用日志,排查问题时比翻控制台强一百倍;tests/哪怕只写一个冒烟测试,也能防止你改坏基础调用。
2.2 环境变量与配置分离的实操细节
API Key绝对不能写进代码。这不是“最好这样做”,是“必须这样做”。我习惯用.env文件配合python-dotenv来管理。.env加入.gitignore,同时提交一份.env.example告诉别人需要哪些变量。
# .env.example MODEL_API_KEY=your_api_key_here MODEL_BASE_URL=https://api.example.com/v1 MODEL_NAME=default-model REQUEST_TIMEOUT=30 MAX_RETRIES=3 LOG_LEVEL=INFOconfig/settings.py里用dataclass做一层封装,好处是类型明确、IDE有补全、后续加配置项不会到处改:
import os from dataclasses import dataclass, field from dotenv import load_dotenv load_dotenv() @dataclass class Settings: api_key: str = field(default_factory=lambda: os.getenv("MODEL_API_KEY", "")) base_url: str = field(default_factory=lambda: os.getenv("MODEL_BASE_URL", "")) model_name: str = field(default_factory=lambda: os.getenv("MODEL_NAME", "default-model")) timeout: int = field(default_factory=lambda: int(os.getenv("REQUEST_TIMEOUT", "30"))) max_retries: int = field(default_factory=lambda: int(os.getenv("MAX_RETRIES", "3"))) log_level: str = field(default_factory=lambda: os.getenv("LOG_LEVEL", "INFO")) def validate(self): if not self.api_key: raise ValueError("MODEL_API_KEY 未设置,请检查 .env 文件") if not self.base_url: raise ValueError("MODEL_BASE_URL 未设置")这里有个细节:validate()方法在初始化时主动调用一次,让配置错误在启动阶段就暴露,而不是等到第一次请求才报错。我吃过这个亏——程序跑了十分钟才走到模型调用,结果发现Key是空的,前面十分钟的准备工作全白费。
注意:
.env文件不要提交到版本控制。我见过有人把带Key的.env推到公开仓库,五分钟内就被扫号脚本盗刷。.gitignore第一行就写.env。
2.3 日志系统:阶段1就要接好,别等出问题再补
日志这东西,平时觉得多余,出问题时就是救命稻草。阶段1的日志至少记录四件事:请求时间、请求消息摘要、响应耗时、响应状态。不要打印完整消息内容,尤其是涉及个人对话时,日志里存全文既不安全也没必要。
import logging import sys def setup_logger(level: str = "INFO") -> logging.Logger: logger = logging.getLogger("assistant") logger.setLevel(getattr(logging, level.upper(), logging.INFO)) handler = logging.StreamHandler(sys.stdout) formatter = logging.Formatter( "%(asctime)s | %(levelname)s | %(name)s | %(message)s", datefmt="%Y-%m-%d %H:%M:%S" ) handler.setFormatter(formatter) if not logger.handlers: logger.addHandler(handler) return logger我习惯把日志同时输出到控制台和文件,文件按天切分。阶段1可以先只做控制台,但logs/目录要建好,后面加文件handler时不用改结构。
3. Model I/O的核心:消息结构设计与调用封装
3.1 消息角色不是随便定的:system、user、assistant的分工逻辑
大模型对话里,每条消息都有角色。最常见的三种是system、user、assistant。很多人随手就把所有内容塞进user,结果模型表现不稳定。这三个角色的作用完全不同:
system:设定模型的整体行为边界。比如“你是一个简洁的个人助手,回答不超过三句话”。它只在对话开头出现一次,影响整个会话的基调。user:用户输入。每次用户说话就是一条user消息。assistant:模型之前的回复。多轮对话时,需要把历史回复按顺序放回去,模型才能理解上下文。
阶段1虽然只做单轮调用,但消息结构必须按多轮来设计。否则阶段2加多轮对话时,整个调用层要重写。我定义了一个简单的Message数据类:
from dataclasses import dataclass from typing import Literal Role = Literal["system", "user", "assistant"] @dataclass class Message: role: Role content: str def to_dict(self) -> dict: return {"role": self.role, "content": self.content}用Literal限定角色取值范围,写错角色名时IDE会直接提示。to_dict()方法负责转换成API需要的格式,这样上层代码不用关心底层API的字段名变化。
3.2 模型客户端封装:把重试、超时、异常都收进一个类
直接调HTTP接口不是不行,但每个调用点都要写一遍超时、重试、异常处理,代码会迅速膨胀。我的做法是写一个ModelClient类,把所有这些横切关注点收进去。上层只需要client.chat(messages),拿到干净的结果。
import time import requests from typing import List from core.message import Message class ModelClient: def __init__(self, settings, logger): self.settings = settings self.logger = logger self.session = requests.Session() self.session.headers.update({ "Authorization": f"Bearer {settings.api_key}", "Content-Type": "application/json" }) def chat(self, messages: List[Message], temperature: float = 0.7) -> str: payload = { "model": self.settings.model_name, "messages": [m.to_dict() for m in messages], "temperature": temperature } last_error = None for attempt in range(1, self.settings.max_retries + 1): try: start = time.time() resp = self.session.post( f"{self.settings.base_url}/chat/completions", json=payload, timeout=self.settings.timeout ) elapsed = time.time() - start resp.raise_for_status() data = resp.json() content = data["choices"][0]["message"]["content"] self.logger.info( f"请求成功 | 耗时={elapsed:.2f}s | 消息数={len(messages)}" ) return content except requests.exceptions.Timeout as e: last_error = e self.logger.warning(f"第{attempt}次请求超时,准备重试") except requests.exceptions.HTTPError as e: status = e.response.status_code if e.response is not None else "unknown" if status in (429, 500, 502, 503, 504): last_error = e self.logger.warning(f"第{attempt}次请求返回{status},准备重试") else: self.logger.error(f"请求失败且不可重试 | 状态码={status}") raise except (KeyError, IndexError) as e: self.logger.error(f"响应结构异常 | 错误={e}") raise if attempt < self.settings.max_retries: time.sleep(2 ** attempt) raise RuntimeError(f"重试{self.settings.max_retries}次后仍失败") from last_error这段代码里有几个关键决策值得展开说。
第一,为什么用requests.Session而不是每次新建连接?Session会复用TCP连接,连续调用时省去握手开销。阶段1可能感觉不明显,但阶段3如果要做流式输出或高频调用,连接复用能显著降低延迟。
第二,重试策略为什么用指数退避?time.sleep(2 ** attempt)意味着第一次重试等2秒,第二次等4秒,第三次等8秒。这是为了应对服务端的限流——如果被限流了还密集重试,只会让情况更糟。指数退避给服务端恢复的时间。
第三,哪些错误该重试、哪些不该?超时和429/5xx可以重试,因为可能是临时问题。但401(认证失败)、400(请求格式错误)重试多少次都没用,直接抛异常让上层处理。这个判断逻辑如果不写清楚,会出现“Key错了还傻傻重试三次”的尴尬。
3.3 温度参数:阶段1就要理解它,别等回答跑偏才后悔
temperature控制输出的随机性。取值通常在0到2之间。低温度(0到0.3)让输出更确定、更保守;高温度(0.7到1.2)让输出更发散、更有创意。个人AI助手场景下,我默认用0.7,但不同任务要区别对待:
| 任务类型 | 建议温度 | 原因 |
|---|---|---|
| 事实问答 | 0.1-0.3 | 减少胡编乱造 |
| 日常对话 | 0.6-0.8 | 保持自然但不失控 |
| 创意写作 | 0.9-1.2 | 需要多样性 |
| 代码生成 | 0.0-0.2 | 要确定性,不要“创意” |
阶段1虽然只做基础调用,但把temperature作为参数暴露出来,后面接不同功能时直接传不同值,不用改客户端代码。
4. 提示词管理:别把system prompt硬编码在代码里
4.1 为什么提示词要独立成文件
我见过太多项目把system prompt写成代码里的一个长字符串。改一次提示词就要改代码、重新测试、重新部署。更麻烦的是,你没法对比两个版本的提示词效果差异。把提示词放到prompts/system_prompt.txt里,好处很直接:改提示词不用动代码逻辑,可以用版本控制追踪每次修改,可以写脚本做A/B对比。
阶段1的system prompt不需要复杂,但要有基本的行为约束。我实际用的版本是这样的:
你是一个个人AI助手,名字叫小助。 你的回答风格:简洁、直接、不废话。 如果用户的问题你不确定,直接说“我不确定”,不要编造。 如果用户的问题超出你的知识范围,建议用户查阅相关资料。 回答使用中文,除非用户明确要求其他语言。这个提示词只有五条,但每条都有用意。“简洁直接”防止模型写小作文;“不确定就说不确定”降低幻觉;“建议查阅资料”给模型一个安全的退路;“使用中文”避免中英混杂。
4.2 提示词加载与缓存
加载提示词很简单,但要注意文件不存在时的处理。我习惯在启动时加载一次并缓存,而不是每次请求都读文件:
from pathlib import Path class PromptManager: def __init__(self, prompt_dir: str = "prompts"): self.prompt_dir = Path(prompt_dir) self._cache = {} def load(self, name: str) -> str: if name in self._cache: return self._cache[name] path = self.prompt_dir / f"{name}.txt" if not path.exists(): raise FileNotFoundError(f"提示词文件不存在: {path}") content = path.read_text(encoding="utf-8").strip() self._cache[name] = content return content缓存的意义在于:阶段2如果要做多轮对话,每轮都要拼system prompt,每次都读磁盘是浪费。缓存后只在第一次读,后续直接命中内存。
提示:提示词文件用UTF-8编码保存。我遇到过用GBK保存导致中文乱码的情况,排查了半天才发现是编码问题。
5. 跑通第一个完整调用:从main.py到实际响应
5.1 main.py的组装逻辑
把所有模块拼起来,main.py应该很薄——只负责初始化、组装、调用、输出。业务逻辑都在core/里。这样做的原因是:阶段2要加命令行交互、阶段3要加Web接口,main.py可以整个替换,而core/不用动。
from config.settings import Settings from core.model_client import ModelClient from core.message import Message from prompts.manager import PromptManager from utils.logger import setup_logger def main(): settings = Settings() settings.validate() logger = setup_logger(settings.log_level) prompt_manager = PromptManager() client = ModelClient(settings, logger) system_prompt = prompt_manager.load("system_prompt") messages = [ Message(role="system", content=system_prompt), Message(role="user", content="你好,请用一句话介绍你自己。") ] try: reply = client.chat(messages) print(f"助手: {reply}") except Exception as e: logger.error(f"调用失败: {e}") print("抱歉,助手暂时无法响应,请稍后再试。") if __name__ == "__main__": main()注意异常处理那一段:日志里记录详细错误,但给用户看的是友好提示。不要把堆栈信息直接抛给终端用户,既不友好也不安全。
5.2 第一次运行最容易遇到的三个问题
问题一:ModuleNotFoundError。通常是因为在项目根目录外运行,或者__init__.py没建。确保每个包目录下都有__init__.py,并且从项目根目录运行python main.py。
问题二:SSL证书错误。某些环境下requests会报证书验证失败。这不是让你关掉验证,而是检查系统证书链是否完整。如果确实在受限环境,可以指定证书路径,但不要用verify=False——那等于把安全门拆了。
问题三:响应结构不符合预期。不同服务商的返回JSON结构可能有差异。我的建议是在chat方法里加一层防御性解析,用.get()逐层取值,取不到时抛出明确异常,而不是让KeyError直接冒出来。
5.3 冒烟测试:三行代码防止改坏基础调用
阶段1至少写一个测试,验证Message能正确转字典、ModelClient能初始化。不需要真的调API(那会消耗额度且不稳定),用mock就行:
from core.message import Message def test_message_to_dict(): msg = Message(role="user", content="测试") assert msg.to_dict() == {"role": "user", "content": "测试"} def test_message_role_validation(): msg = Message(role="system", content="你是助手") assert msg.to_dict()["role"] == "system"这两个测试看起来简单,但当你后面重构Message类时,它们能立刻告诉你有没有改坏基础结构。测试不是给阶段1写的,是给阶段3的你写的。
6. 阶段1的边界:哪些事现在不做,为什么
6.1 不做流式输出,但预留接口
流式输出(逐字返回)体验好,但阶段1不做。原因有两个:一是流式输出的错误处理比阻塞式复杂得多,中途断流、部分响应、重试逻辑都要重新设计;二是阶段1的目标是验证调用链通畅,不是优化体验。但我会在ModelClient里预留一个stream参数,默认False,阶段2再实现。
6.2 不做多轮对话,但消息结构已支持
阶段1的main.py只发一轮消息。但Message列表结构天然支持多轮——阶段2只需要把历史消息追加进去就行。这就是为什么阶段1要坚持用列表而不是单个字符串。
6.3 不做工具调用,但客户端已可扩展
工具调用(function calling)是Agent的核心能力,但阶段1不碰。不过ModelClient.chat的签名设计成接收消息列表、返回字符串,阶段2加工具调用时,可以新增一个chat_with_tools方法,而不影响现有调用。
6.4 阶段1的验收标准
怎么判断阶段1做完了?我给自己定了四条:
- 从零克隆项目,按README步骤能在五分钟内跑通第一次调用。
- API Key不在代码里,
.env.example能说明所有必需变量。 - 调用失败时有明确日志,且不会泄露敏感信息。
- 有一个可运行的测试文件,
pytest能通过。
这四条都满足,阶段1才算收工。少一条,后面都会还债。
7. 实操心得:那些文档里不会写的细节
关于超时设置。默认30秒对大多数模型够用,但如果你用的是推理型模型或者长输出任务,30秒可能不够。我的做法是把超时做成可配置,并且在日志里记录每次请求的实际耗时。跑一段时间后,根据P95耗时来调整超时值,而不是拍脑袋定。
关于重试的副作用。重试意味着同一个请求可能被发送多次。如果请求本身有副作用(比如计费),重试会导致重复计费。阶段1的对话请求没有副作用,可以放心重试。但阶段2如果加了“写文件”“发消息”这类工具,重试逻辑要重新设计——不能让模型因为重试而重复执行动作。
关于日志脱敏。阶段1的日志只记录消息数量和耗时,不记录内容。但如果你调试时需要看内容,记得在日志输出前做脱敏处理。我习惯写一个mask_sensitive函数,把疑似Key、手机号、邮箱的字符串替换成***。
关于依赖版本锁定。requirements.txt里不要只写requests,要写requests==2.31.0这样的精确版本。我遇到过因为requests小版本升级导致SSL行为变化的情况,锁定版本能避免这类意外。
关于项目README。阶段1就要写README,哪怕只有十行。写清楚:项目是什么、怎么安装依赖、怎么配置环境变量、怎么运行。三个月后你回来看这个项目,README就是你的记忆。
8. 从阶段1到阶段2的衔接点
阶段1结束后,你手里应该有一个能跑通单轮对话的干净底子。阶段2要做的第一件事通常是加多轮对话循环——把main.py里的单次调用改成while循环,每次把用户输入追加到消息列表,把模型回复也追加进去。因为阶段1的消息结构已经是列表,这个改动只涉及main.py,core/完全不用动。
第二件事通常是加命令行交互,用input()读取用户输入,加一个exit命令退出。这时候你会发现阶段1的日志系统开始发挥作用——每次对话的耗时和消息数都有记录,方便你观察多轮对话下上下文长度对响应时间的影响。
第三件事可能是加简单的记忆——把对话历史存到本地文件,下次启动时加载。这时候Message的to_dict和反向的from_dict就派上用场了。阶段1没写from_dict?没关系,加一个就行,因为结构是清晰的。
我在多个项目里反复验证过:阶段1多花两天把Model I/O做扎实,阶段2和阶段3能省下一周的重构时间。那些一开始图快、把所有逻辑塞进一个文件的项目,最后都卡在“想加功能但不敢改”的困境里。个人AI助手是个长期项目,底子比速度重要。