1. 从零搭建个人AI助手：为什么第一阶段只做Model I/O
很多人一上来就想搞个全能助手，能查天气、能订日程、能写代码、能陪聊，结果三天热情一过，项目文件夹里只剩一个README。我踩过这个坑，而且不止一次。后来复盘发现，问题出在跳过了最枯燥但也最关键的环节——把模型的基础输入输出跑通、跑稳、跑明白。
这个阶段的目标非常克制：让程序能稳定地和模型对话，并且把对话结果结构化地拿回来。听起来简单，但这里面藏着后面所有功能的地基。你后面要做工具调用、要做记忆管理、要做多轮规划，全都依赖这一层是否干净可靠。
我见过太多项目，初期为了"快点看到效果"，把模型调用随手塞在业务逻辑里，参数硬编码、异常不处理、返回格式随缘解析。等到要加第二个模型、要切换供应商、要做流式输出的时候，整个代码推倒重来。所以阶段1我给自己定的原则是：只做Model I/O，但要做到能换模型不改业务代码、能处理异常不崩溃、能看清每次调用的输入输出。
这篇文章适合两类人：一是刚接触Agent开发、想找个靠谱起点的朋友；二是做过一些Demo但总觉得代码越写越乱、想重新梳理底层结构的开发者。我会把项目初始化的目录设计、模型接口的抽象方式、基础对话循环的实现、以及实测中遇到的坑，全部摊开讲清楚。
2. 项目初始化：目录结构和依赖管理的前置决策
2.1 为什么目录结构要在写第一行代码前定好
我刚开始做个人项目时，习惯是"先写个main.py跑起来再说"。这个习惯在脚本阶段没问题，但一旦项目要持续迭代超过一周，就会变成灾难。原因很简单：模型调用、配置管理、日志记录、业务逻辑这四类东西，它们的变更频率完全不同。配置可能天天调，业务逻辑可能一周才动一次，如果混在一起，每次改配置都要在几百行代码里翻找。
所以阶段1我采用的目录结构是这样的：
personal-agent/ ├── config/ │ └── settings.yaml ├── src/ │ ├── core/ │ │ ├── model_client.py │ │ └── message.py │ ├── utils/ │ │ ├── logger.py │ │ └── config_loader.py │ └── main.py ├── logs/ ├── tests/ │ └── test_model_client.py ├── .env ├── .gitignore └── requirements.txt这个结构里，core放的是与模型交互的核心抽象，utils放通用工具，config放配置文件，logs放运行日志。关键点是：core里的代码不应该知道任何具体业务。它只负责"发消息给模型，拿回响应"这件事。
注意：
.env文件一定要放进.gitignore，API密钥泄露是个人项目最常见的安全事故。我习惯在项目根目录放一个.env.example，里面只写键名不写值，方便换机器时快速配置。
2.2 依赖选择：为什么我最终只留了三个核心库
阶段1的依赖我刻意压到最少：
| 依赖 | 用途 | 为什么选它 |
|---|---|---|
openai | 模型调用SDK | 接口设计清晰，兼容多家供应商 |
pyyaml | 配置文件解析 | 比JSON可读性好，支持注释 |
python-dotenv | 环境变量加载 | 轻量，不侵入代码 |
没有引入LangChain、没有引入任何Agent框架。这不是说框架不好，而是阶段1的目标是理解Model I/O的本质。框架帮你封装了太多东西，一旦出问题，你连日志都看不懂。我建议先用原生SDK把一次对话的完整链路跑通，知道请求长什么样、响应长什么样、token怎么算、错误码有哪些，之后再决定要不要上框架。
requirements.txt我习惯锁定小版本：
openai>=1.30.0,<2.0.0 pyyaml>=6.0,<7.0 python-dotenv>=1.0.0,<2.0.0为什么要锁版本？因为模型SDK的接口变动非常频繁，今天能跑的代码下周可能就报参数错误。锁版本能保证你本地环境稳定，等你想升级时再主动升，而不是被动被破坏。
2.3 配置文件的字段设计
settings.yaml我设计成三个区块：
model: provider: "openai" name: "gpt-4o-mini" temperature: 0.7 max_tokens: 1024 timeout: 30 conversation: system_prompt: "你是一个简洁、准确的个人助手。" max_history_rounds: 10 logging: level: "INFO" file: "logs/agent.log"这里有几个设计决策值得说。temperature放在配置里而不是代码里，因为不同任务对创造性的要求不同，后面做工具调用时可能需要调到0，做头脑风暴时可能需要调到1，配置化之后改一个数字就行。max_history_rounds是提前为多轮对话预留的，阶段1虽然只做单轮，但结构先留好，后面加历史管理时不用重构。timeout必须设，我吃过亏，默认超时可能长达几分钟，程序卡死你还以为是逻辑问题。
3. 模型客户端抽象：让换模型不改业务代码
3.1 为什么直接调SDK是短视的
最直接的写法是这样：
from openai import OpenAI client = OpenAI() response = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": "你好"}] ) print(response.choices[0].message.content)这段代码能跑，但它把三件事耦合在了一起：模型供应商、模型名称、消息格式。哪天你想换成另一个供应商的模型，或者想同时支持两个模型做对比，就得改业务代码。更麻烦的是，不同供应商的SDK返回结构不一样，你的解析逻辑会变成一堆if-else。
我的做法是定义一个ModelClient抽象层，业务代码只跟这个抽象层打交道。抽象层的职责是：接收标准化的消息列表，返回标准化的响应对象。至于底层是哪个供应商、用的什么SDK，业务代码不关心。
3.2 消息对象的标准化设计
在message.py里我定义了两个数据类：
from dataclasses import dataclass, field from typing import Literal @dataclass class Message: role: Literal["system", "user", "assistant"] content: str @dataclass class ModelResponse: content: str model: str prompt_tokens: int = 0 completion_tokens: int = 0 total_tokens: int = 0 finish_reason: str = "" raw: dict = field(default_factory=dict)Message用Literal约束role，这样IDE能提示，写错角色名会立刻报错。ModelResponse里保留了raw字段，存原始响应，方便调试时查看完整信息。prompt_tokens和completion_tokens分开记录，因为后面做成本控制时，你需要知道是输入太长还是输出太长。
提示：
finish_reason这个字段很多人忽略，但它很重要。如果是length，说明输出被max_tokens截断了，你需要调大限制或让模型精简；如果是stop，说明正常结束；如果是content_filter，说明内容被拦截了。不检查这个字段，你会拿到半截回答还不知道为什么。
3.3 ModelClient的实现与供应商切换
model_client.py的核心结构：
import os from openai import OpenAI from src.core.message import Message, ModelResponse from src.utils.config_loader import load_config from src.utils.logger import get_logger logger = get_logger(__name__) class ModelClient: def __init__(self, config=None): self.config = config or load_config()["model"] self.client = OpenAI( api_key=os.getenv("MODEL_API_KEY"), base_url=os.getenv("MODEL_BASE_URL"), timeout=self.config.get("timeout", 30) ) def chat(self, messages: list[Message]) -> ModelResponse: payload = [{"role": m.role, "content": m.content} for m in messages] logger.info(f"发送 {len(payload)} 条消息，模型 {self.config['name']}") try: resp = self.client.chat.completions.create( model=self.config["name"], messages=payload, temperature=self.config.get("temperature", 0.7), max_tokens=self.config.get("max_tokens", 1024) ) except Exception as e: logger.error(f"模型调用失败: {e}") raise choice = resp.choices[0] usage = resp.usage return ModelResponse( content=choice.message.content or "", model=resp.model, prompt_tokens=usage.prompt_tokens, completion_tokens=usage.completion_tokens, total_tokens=usage.total_tokens, finish_reason=choice.finish_reason, raw=resp.model_dump() )这里的关键设计是base_url从环境变量读取。很多供应商兼容OpenAI接口协议，只要改base_url和api_key就能切换，业务代码一行不用动。这就是抽象层的价值。
3.4 异常处理的分层策略
模型调用可能出的错分三类，处理方式不同：
- 网络类错误（超时、连接失败）：可以重试，我一般重试2次，间隔1秒和3秒。
- 参数类错误（模型名不存在、token超限）：重试没用，直接抛出，让调用方知道配置有问题。
- 内容类错误（内容被拦截）：返回一个带说明的响应，而不是抛异常，因为这是正常业务场景。
我在chat方法里没有做重试，而是把重试逻辑放在更上层。原因是重试策略可能因场景而异，有的场景需要快速失败，有的场景可以慢慢重试。抽象层只负责"调一次"，重试是调用方的决策。
4. 基础对话循环：从单轮到可控多轮
4.1 单轮对话的最小闭环
阶段1先跑通单轮：
def single_turn(client, user_input): messages = [ Message(role="system", content="你是一个简洁、准确的个人助手。"), Message(role="user", content=user_input) ] response = client.chat(messages) return response.content跑通这个之后，你会看到模型返回的内容、token消耗、结束原因。这时候先别急着加功能，先做一件事：把不同输入下的响应都试一遍。空输入会怎样？超长输入会怎样？包含特殊字符会怎样？这些边界情况现在摸清楚，比后面出问题再查要省事得多。
4.2 多轮对话的历史管理
多轮对话的本质是把历史消息一起发给模型。但历史不能无限增长，否则token消耗会爆炸，而且模型对超长上下文的注意力也会下降。我的策略是保留最近N轮：
class Conversation: def __init__(self, client, system_prompt, max_rounds=10): self.client = client self.system_prompt = system_prompt self.max_rounds = max_rounds self.history = [] def send(self, user_input): self.history.append(Message(role="user", content=user_input)) messages = [Message(role="system", content=self.system_prompt)] + self.history response = self.client.chat(messages) self.history.append(Message(role="assistant", content=response.content)) self._trim() return response def _trim(self): max_messages = self.max_rounds * 2 if len(self.history) > max_messages: self.history = self.history[-max_messages:]_trim方法保留最近的消息，但这里有个细节：裁剪时不能把user和assistant的消息拆散。一轮对话是user加assistant两条消息，如果只保留奇数条，会出现历史里最后一条是user但没有对应assistant的情况，模型可能会困惑。所以max_messages是max_rounds * 2。
4.3 流式输出的取舍
流式输出能让用户更快看到内容，体验更好。但它增加了复杂度：你需要处理分块拼接、需要处理流中断、需要在流式和非流式之间切换。阶段1我的建议是先不做流式，把非流式跑稳。等基础对话循环没问题了，再单独加一个流式方法。
如果要做，SDK层面是stream=True，然后迭代响应块：
def chat_stream(self, messages): payload = [{"role": m.role, "content": m.content} for m in messages] stream = self.client.chat.completions.create( model=self.config["name"], messages=payload, stream=True ) for chunk in stream: delta = chunk.choices[0].delta if delta.content: yield delta.content注意流式模式下拿不到完整的usage信息，token统计需要在流结束后单独处理，或者用SDK提供的stream_options参数。这是很多人踩的坑：以为流式和非流式只是输出方式不同，实际上返回结构差异很大。
5. 实测中暴露的问题与修复过程
5.1 问题一：配置文件路径依赖当前工作目录
第一次跑的时候报FileNotFoundError，因为load_config里写的是相对路径config/settings.yaml。从项目根目录跑没问题，但从其他目录跑就找不到。修复方式是基于文件位置计算绝对路径：
from pathlib import Path BASE_DIR = Path(__file__).resolve().parent.parent.parent def load_config(): config_path = BASE_DIR / "config" / "settings.yaml" with open(config_path, "r", encoding="utf-8") as f: return yaml.safe_load(f)这个坑很典型，所有涉及文件读取的地方都要注意。我的经验是：项目内所有路径都基于一个BASE_DIR计算，永远不要用相对路径。
5.2 问题二：环境变量没加载就调用
OpenAI()初始化时如果api_key是None，会在第一次请求时才报错，而不是初始化时报错。这导致错误信息很晚才出现，排查时容易误以为是网络问题。修复方式是在ModelClient.__init__里主动校验：
api_key = os.getenv("MODEL_API_KEY") if not api_key: raise ValueError("MODEL_API_KEY 未设置，请检查 .env 文件")早失败优于晚失败，这是我在多个项目里总结出的原则。配置类问题应该在启动时就暴露，而不是等到运行到某个分支才报错。
5.3 问题三：日志里看不到完整请求
调试时我想看每次发给模型的完整消息，但日志里只记了消息条数。后来改成记录完整payload，但又有新问题：日志文件迅速变大，而且包含用户隐私内容。最终方案是分级记录：INFO级别记条数和模型名，DEBUG级别记完整内容，并且DEBUG日志单独存文件、定期清理。
logger.info(f"发送 {len(payload)} 条消息，模型 {self.config['name']}") logger.debug(f"完整请求: {payload}")5.4 问题四：token超限没有预警
有一次输入比较长，模型返回的finish_reason是length，内容被截断了，但程序没任何提示。后来我在ModelResponse里加了一个便捷属性：
@property def is_truncated(self): return self.finish_reason == "length"调用方拿到响应后检查这个属性，如果被截断就提示用户或自动调整。不要假设模型一定会完整回答，截断是常态，尤其是max_tokens设得比较小的时候。
6. 阶段1的验收标准与后续扩展方向
6.1 怎么判断阶段1算完成了
我给自己定了四条验收标准：
- 能从配置文件读取模型参数，改配置不需要改代码。
- 能通过环境变量切换供应商，业务代码零改动。
- 单轮和多轮对话都能稳定运行，历史裁剪逻辑正确。
- 所有异常都有日志，token消耗可追踪，截断可识别。
这四条都满足，阶段1就算过关。听起来简单，但每一条背后都对应着后面会反复用到的能力。比如第2条，等你后面想对比两个模型的效果时，就会发现这个抽象层省了多少事。
6.2 阶段2要做什么，以及为什么现在不做
阶段2我计划做工具调用（Function Calling）。但现在不做，是因为工具调用依赖一个稳定的Model I/O层。如果模型调用本身还不稳定，加工具调用只会让问题更难定位。而且工具调用会引入新的消息角色（tool角色）和新的响应结构（tool_calls字段），这些都需要在抽象层里预留位置。
我在Message的role里暂时只定义了system、user、assistant三种，等阶段2再加tool。ModelResponse里也暂时没加tool_calls字段，但raw字段保留了完整响应，需要时可以从raw里取。这种渐进式扩展的方式，比一开始就设计一个大而全的接口要务实得多。
6.3 一个容易被忽略的细节：system prompt的管理
阶段1我把system prompt放在配置文件里，但实际使用中你会发现，不同场景需要不同的system prompt。比如写代码时希望模型严谨，头脑风暴时希望模型发散。我的做法是把system prompt也做成可切换的配置，而不是硬编码一个：
conversation: system_prompts: default: "你是一个简洁、准确的个人助手。" coding: "你是一个严谨的编程助手，回答要包含可运行的代码。" brainstorm: "你是一个创意助手，鼓励发散思考，提供多个角度。" active: "default"这样切换场景只需要改一个字段。后面做多Agent时，每个Agent可以有自己的system prompt，结构上已经预留好了。
6.4 关于测试的一点经验
阶段1我写了test_model_client.py，但测试不调用真实API，而是mock掉client.chat.completions.create。原因是真实API调用有成本、有延迟、结果不确定，不适合做单元测试。mock测试验证的是：消息格式转换是否正确、响应解析是否正确、异常处理是否按预期走。
from unittest.mock import MagicMock, patch def test_chat_returns_model_response(): with patch("src.core.model_client.OpenAI") as mock_openai: mock_client = MagicMock() mock_openai.return_value = mock_client mock_client.chat.completions.create.return_value = fake_response() client = ModelClient(config={"name": "test", "timeout": 30}) result = client.chat([Message(role="user", content="hi")]) assert result.content == "hello"真实API的验证我放在手动测试里，跑几个典型输入看看效果。自动化测试保证逻辑正确，手动测试保证效果符合预期，两者不能互相替代。
6.5 成本控制的早期意识
虽然阶段1调用量很小，但我还是加了一个简单的token累计统计：
class TokenTracker: def __init__(self): self.total_prompt = 0 self.total_completion = 0 def add(self, response): self.total_prompt += response.prompt_tokens self.total_completion += response.completion_tokens def report(self): return { "prompt": self.total_prompt, "completion": self.total_completion, "total": self.total_prompt + self.total_completion }每次对话后打印一下累计消耗，心里有数。等后面做长对话、做批量任务时，这个统计能帮你快速定位是哪个环节在烧token。我见过有人做Agent跑了一晚上，第二天发现账单惊人，就是因为没有早期建立成本意识。
6.6 日志轮转：一个迟早要面对的问题
阶段1日志量不大，但我还是配了轮转，因为不想等到日志文件几百MB时才手忙脚乱：
from logging.handlers import RotatingFileHandler handler = RotatingFileHandler( "logs/agent.log", maxBytes=5 * 1024 * 1024, backupCount=3, encoding="utf-8" )5MB一个文件，保留3个备份，总共不超过20MB。这个配置对个人项目足够用。基础设施类的配置要提前做，不要等出问题才补，因为出问题时你往往正忙着排查业务逻辑，没精力再搞这些。
6.7 关于模型选择的一点个人体会
阶段1我用的是小模型（gpt-4o-mini这类），原因是开发和调试阶段不需要最强模型，小模型响应快、成本低，适合快速迭代。等逻辑跑通、要验证效果时，再切换到强模型对比。配置文件里改一个name字段就行，这就是抽象层带来的便利。
但要注意，不同模型对同一个prompt的响应风格差异很大。用小模型调好的prompt，换到大模型可能效果不同。所以切换模型后要重新验证关键场景，不能假设行为一致。
6.8 最后分享一个调试技巧
调试模型调用时，我习惯把每次请求和响应都存成JSON文件，放在logs/debug/目录下，文件名带时间戳。这样出问题时可以精确复现当时的输入，也可以对比不同版本的prompt效果。比在日志里翻要方便得多。
import json from datetime import datetime def save_debug(payload, response): ts = datetime.now().strftime("%Y%m%d_%H%M%S") path = f"logs/debug/{ts}.json" with open(path, "w", encoding="utf-8") as f: json.dump({"request": payload, "response": response.raw}, f, ensure_ascii=False, indent=2)这个习惯是从做数据标注项目时养成的，后来发现做Agent开发同样适用。可复现的调试记录，比任何调试工具都管用。
阶段1的内容就是这些。核心就一句话：把模型调用这层做干净、做稳、做可替换，后面所有功能都是在这层之上叠加。我踩过的那些坑，希望你能绕过去。
