首页 / 资讯中心 / 文章详情

从零搭建个人AI助手:Model I/O基础与模型客户端抽象

从零搭建个人AI助手:Model I/O基础与模型客户端抽象 ★ FEATURED ARTICLE
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文件一定要放进.gitignoreAPI密钥泄露是个人项目最常见的安全事故。我习惯在项目根目录放一个.env.example里面只写键名不写值方便换机器时快速配置。2.2 依赖选择为什么我最终只留了三个核心库阶段1的依赖我刻意压到最少依赖用途为什么选它openai模型调用SDK接口设计清晰兼容多家供应商pyyaml配置文件解析比JSON可读性好支持注释python-dotenv环境变量加载轻量不侵入代码没有引入LangChain、没有引入任何Agent框架。这不是说框架不好而是阶段1的目标是理解Model I/O的本质。框架帮你封装了太多东西一旦出问题你连日志都看不懂。我建议先用原生SDK把一次对话的完整链路跑通知道请求长什么样、响应长什么样、token怎么算、错误码有哪些之后再决定要不要上框架。requirements.txt我习惯锁定小版本openai1.30.0,2.0.0 pyyaml6.0,7.0 python-dotenv1.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( modelgpt-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_factorydict)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, configNone): self.config config or load_config()[model] self.client OpenAI( api_keyos.getenv(MODEL_API_KEY), base_urlos.getenv(MODEL_BASE_URL), timeoutself.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( modelself.config[name], messagespayload, temperatureself.config.get(temperature, 0.7), max_tokensself.config.get(max_tokens, 1024) ) except Exception as e: logger.error(f模型调用失败: {e}) raise choice resp.choices[0] usage resp.usage return ModelResponse( contentchoice.message.content or , modelresp.model, prompt_tokensusage.prompt_tokens, completion_tokensusage.completion_tokens, total_tokensusage.total_tokens, finish_reasonchoice.finish_reason, rawresp.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(rolesystem, content你是一个简洁、准确的个人助手。), Message(roleuser, contentuser_input) ] response client.chat(messages) return response.content跑通这个之后你会看到模型返回的内容、token消耗、结束原因。这时候先别急着加功能先做一件事把不同输入下的响应都试一遍。空输入会怎样超长输入会怎样包含特殊字符会怎样这些边界情况现在摸清楚比后面出问题再查要省事得多。4.2 多轮对话的历史管理多轮对话的本质是把历史消息一起发给模型。但历史不能无限增长否则token消耗会爆炸而且模型对超长上下文的注意力也会下降。我的策略是保留最近N轮class Conversation: def __init__(self, client, system_prompt, max_rounds10): self.client client self.system_prompt system_prompt self.max_rounds max_rounds self.history [] def send(self, user_input): self.history.append(Message(roleuser, contentuser_input)) messages [Message(rolesystem, contentself.system_prompt)] self.history response self.client.chat(messages) self.history.append(Message(roleassistant, contentresponse.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层面是streamTrue然后迭代响应块def chat_stream(self, messages): payload [{role: m.role, content: m.content} for m in messages] stream self.client.chat.completions.create( modelself.config[name], messagespayload, streamTrue ) 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, encodingutf-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(roleuser, contenthi)]) 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, maxBytes5 * 1024 * 1024, backupCount3, encodingutf-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 flogs/debug/{ts}.json with open(path, w, encodingutf-8) as f: json.dump({request: payload, response: response.raw}, f, ensure_asciiFalse, indent2)这个习惯是从做数据标注项目时养成的后来发现做Agent开发同样适用。可复现的调试记录比任何调试工具都管用。阶段1的内容就是这些。核心就一句话把模型调用这层做干净、做稳、做可替换后面所有功能都是在这层之上叠加。我踩过的那些坑希望你能绕过去。
阅读完成 · 觉得有帮助?
咨询建站