最近在推进一个AI智能体项目时团队遇到了一个棘手的问题随着功能迭代每次调用大模型API的成本像坐了火箭一样飙升。仔细排查后发现问题根源并非业务逻辑复杂而是项目早期堆砌的“Prompt工程”和臃肿的上下文导致每次请求都携带了大量无效token。这让我想起了Thoughtworks CTO Rebecca Parsons博士团队近期分享的一个实验他们通过系统性的代码库重构成功将AI智能体的token消耗降低了惊人的70%。这不仅仅是技术优化更直接关乎项目的经济效益和可持续性。本文将深入拆解这个案例并结合实战为你展示如何通过重构AI智能体代码库来显著降低token成本覆盖从问题诊断、重构策略到具体代码实现的完整闭环。1. 背景与核心概念为什么AI智能体的token成本如此关键在深入重构之前我们必须理解几个核心概念以及它们如何与项目成本紧密挂钩。AI智能体AI Agent通常指能够感知环境、进行决策并执行动作以完成特定目标的程序。在现代开发中它常常通过与大语言模型LLMAPI如OpenAI GPT、Claude、文心一言等交互来实现“思考”和“内容生成”。每一次与LLM API的交互我们称之为一次“请求”。Token是大语言模型处理文本的基本单位。它可以是一个字、一个词或标点符号。API提供商如OpenAI通常按照请求和响应消耗的token总数来计费。例如GPT-4 Turbo每1000个token可能花费0.01到0.03美元不等。一个复杂的请求消耗数千token是家常便饭。问题的根源在AI智能体项目的快速原型阶段开发者为了快速实现功能往往会采取一些“短平快”的策略冗长的系统提示词System Prompt将所有指令、角色设定、约束条件都塞进一个庞大的提示词中。上下文信息过载每次请求都将完整的对话历史、知识库文档片段不加处理地送入上下文。低效的指令设计使用自然语言描述复杂逻辑而不是让模型输出结构化数据如JSON。代码与提示词耦合过紧业务逻辑变更需要同步修改多处分散的提示词模板容易出错且低效。这些做法直接导致了每次API调用的token消耗居高不下。在项目初期流量较小时成本或许可以接受。但随着用户量增长或智能体复杂度提升token成本将成为项目财务模型中的一个巨大变量甚至可能决定项目的生死。Thoughtworks的实验正是针对这一问题他们证明通过像重构传统软件代码一样去重构“提示词”和与LLM交互的“代码库”可以带来立竿见影的经济效益。2. 环境准备与版本说明本文的实战示例将使用Python语言因为它是在AI应用开发中最流行的语言之一。我们将使用openai这个官方库来模拟与LLM的交互。请注意重构的核心思想是语言和框架无关的同样适用于Node.js、Java、Go等其他技术栈。基础环境操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04)Python版本3.8 或更高版本包管理工具pip核心Python库openai1.0.0OpenAI官方Python SDK。python-dotenv0.19.0用于管理环境变量如API密钥。你可以通过以下命令安装所需库pip install openai python-dotenv项目结构重构前一个典型的、未经优化的AI智能体项目可能杂乱无章my_ai_agent/ ├── main.py # 所有逻辑堆在一起 ├── config.py # 可能存放着巨大的提示词字符串 ├── .env # 存放API密钥 └── knowledge_base/ # 存放原始文档 └── some_document.txt我们的目标是将它重构为一个清晰、高效、易于维护的结构。3. 核心重构策略与原理拆解Thoughtworks团队提出的重构并非简单的代码整理而是一套针对LLM交互特性的优化方法论。我们可以将其归纳为以下四个核心策略3.1 策略一提示词模块化与分层设计将庞大的、单一的提示词拆分为可复用的模块。系统角色层定义智能体的核心身份和基础行为准则这部分相对稳定。任务指令层针对具体任务如总结、翻译、代码生成的指令集。上下文管理层动态管理对话历史和检索到的知识只注入必要信息。输出格式化层明确要求LLM以特定格式JSON、XML、Markdown输出便于后续解析。为什么有效避免了每次请求都重复发送不变的背景信息并且可以动态组合减少冗余。3.2 策略二上下文压缩与精炼LLM的上下文窗口是宝贵的资源。不要将原始文档直接塞入提示词。检索增强生成RAG优化先使用嵌入模型Embedding和向量数据库检索出最相关的文档片段而不是整篇文档。摘要与提炼对于必须保留的较长历史对话可以定期让LLM生成之前对话的简短摘要用摘要替代原始长文本作为后续对话的上下文。剔除无关信息在注入上下文前编写过滤逻辑移除无关的格式标记、广告文本等。为什么有效直接减少了构成请求主体的上下文内容的token数量。3.3 策略三结构化输出与函数调用利用LLM支持的结构化输出功能如OpenAI的JSON Mode和函数调用Function Calling现称为Tool Calling。JSON Mode强制模型输出规范的JSON对象而不是自由文本。这使提示词指令更精确模型输出更简洁、无赘述。Tool Calling让模型选择调用你预先定义好的工具函数。你可以将复杂操作如计算、数据库查询封装为工具模型只需输出调用这些工具的指令参数极大简化了交互。为什么有效减少了模型在自然语言描述上“自由发挥”所产生的冗余token并使交互流程标准化、高效化。3.4 策略四缓存与复用对于频繁出现的、计算结果固定的子问题或子查询将其结果缓存起来。语义缓存不是简单的字符串匹配而是计算用户查询的语义嵌入向量在缓存中查找相似度高的历史查询及其结果。如果找到直接返回缓存结果无需调用LLM。模板结果缓存对于格式固定、仅数据不同的输出如邮件模板、报告框架可以缓存模板本身只让LLM填充变化的数据部分。为什么有效避免了为相同或相似的请求重复支付token费用尤其在高并发场景下效益显著。4. 完整实战案例重构一个文档问答智能体让我们通过一个具体的例子将一个“简陋”的文档问答智能体重构成一个高效、低成本的版本。需求用户上传一篇长文档然后可以针对文档内容进行提问。4.1 重构前低效的“一次性”实现main_inefficient.pyimport openai import os from dotenv import load_dotenv load_dotenv() client openai.OpenAI(api_keyos.getenv(OPENAI_API_KEY)) def answer_question_about_document(document_path, user_question): # 1. 读取整个文档可能非常长 with open(document_path, r, encodingutf-8) as f: full_document f.read() # 2. 构造一个庞大而模糊的提示词 prompt f 你是一个专业的文档分析助手。请基于以下文档内容回答用户的问题。 文档内容 {full_document} 用户问题{user_question} 请给出详细、准确的答案。 # 3. 发起API调用 response client.chat.completions.create( modelgpt-4-turbo-preview, messages[ {role: user, content: prompt} ], temperature0.7, max_tokens1000 ) return response.choices[0].message.content # 使用示例 if __name__ __main__: answer answer_question_about_document(knowledge_base/project_plan.txt, 项目第三阶段的里程碑是什么) print(answer)问题分析Token浪费严重无论用户问什么都把整篇文档可能数万字塞进上下文。提示词模糊指令不清晰导致模型可能输出冗长的前言或不必要的格式。无结构输出答案是非结构化的文本如果后续程序需要处理还需额外解析。无法利用历史多次问答时每次都是独立请求无法基于之前问答进行深入探讨或需要再次传入全部历史成本更高。4.2 重构后模块化、高效、可维护的实现我们按照前述策略重构项目结构my_ai_agent_refactored/ ├── .env ├── requirements.txt ├── main.py # 应用主入口 ├── core/ │ ├── __init__.py │ ├── llm_client.py # 封装LLM客户端 │ ├── prompt_templates.py # 模块化提示词模板 │ └── token_counter.py # Token计算与监控工具 ├── services/ │ ├── __init__.py │ ├── document_processor.py # 文档处理与检索 │ └── cache_manager.py # 语义缓存管理 └── utils/ └── __init__.py步骤1创建模块化的提示词模板core/prompt_templates.pySYSTEM_ROLE 你是一个精准、高效的文档问答助手。你的核心任务是 1. 严格基于提供的**文档片段**回答问题。 2. 如果文档片段中不包含答案所需信息直接回答“根据提供的资料无法回答此问题”。 3. 答案必须简洁、准确优先使用列表或要点形式。 4. 绝对不要添加“根据文档”、“综上所述”等引导性套话。 def format_qa_prompt(context_chunks: list[str], question: str) - list[dict]: 格式化问答提示词。 返回OpenAI API所需的messages列表。 # 将检索到的多个相关片段合并用明确的分隔符隔开 combined_context \n\n---\n\n.join(context_chunks) user_content f 相关文档上下文 {combined_context} 请回答以下问题 {question} return [ {role: system, content: SYSTEM_ROLE}, {role: user, content: user_content} ] def format_summarize_prompt(text: str) - list[dict]: 格式化摘要生成提示词。 return [ {role: system, content: 你是一个文本摘要专家。请将以下文本压缩成不超过3句话的核心摘要。}, {role: user, content: text} ]步骤2实现文档处理与检索服务模拟RAGservices/document_processor.py# 简化的文本分割与检索模拟。真实场景应使用LangChain、LlamaIndex或自定义嵌入模型向量数据库。 import re class DocumentProcessor: def __init__(self, chunk_size500, chunk_overlap50): self.chunk_size chunk_size self.chunk_overlap chunk_overlap def split_document(self, text: str) - list[str]: 按字符数简单分割文档。生产环境应使用更智能的分割器。 words text.split() chunks [] current_chunk [] current_length 0 for word in words: current_chunk.append(word) current_length len(word) 1 # 1 for space if current_length self.chunk_size: chunks.append( .join(current_chunk)) # 保留重叠部分 overlap_words current_chunk[-self.chunk_overlap:] if len(current_chunk) self.chunk_overlap else current_chunk current_chunk overlap_words.copy() current_length sum(len(w) 1 for w in current_chunk) if current_chunk: chunks.append( .join(current_chunk)) return chunks def retrieve_relevant_chunks(self, all_chunks: list[str], question: str, top_k3) - list[str]: 模拟检索最相关的片段。 真实实现应计算question和每个chunk的嵌入向量然后进行相似度排序。 这里使用简单的关键词匹配作为演示。 question_keywords set(re.findall(r\w, question.lower())) scored_chunks [] for chunk in all_chunks: chunk_words set(re.findall(r\w, chunk.lower())) # 计算Jaccard相似度简化版 score len(question_keywords chunk_words) / len(question_keywords | chunk_words) if (question_keywords | chunk_words) else 0 scored_chunks.append((score, chunk)) # 按分数降序排序取前top_k个 scored_chunks.sort(keylambda x: x[0], reverseTrue) return [chunk for _, chunk in scored_chunks[:top_k]]步骤3封装LLM客户端并集成结构化输出core/llm_client.pyimport openai import os import json from typing import Optional, Dict, Any from .token_counter import estimate_tokens # 稍后实现的token计数器 class LLMClient: def __init__(self, model: str gpt-4-turbo-preview): self.client openai.OpenAI(api_keyos.getenv(OPENAI_API_KEY)) self.model model def chat_completion(self, messages: list[dict], json_mode: bool False, max_tokens: int 800) - Dict[str, Any]: 发起聊天补全请求支持JSON模式。 返回包含完整响应信息的字典。 # 在发送前估算token消耗用于监控和日志 input_token_est estimate_tokens(str(messages)) params { model: self.model, messages: messages, temperature: 0.1, # 降低温度使输出更确定、更简洁 max_tokens: max_tokens, } if json_mode: params[response_format] {type: json_object} try: response self.client.chat.completions.create(**params) output_text response.choices[0].message.content # 估算输出token output_token_est estimate_tokens(output_text) return { success: True, content: output_text, input_tokens_est: input_token_est, output_tokens_est: output_token_est, total_tokens_est: input_token_est output_token_est, full_response: response } except Exception as e: return {success: False, error: str(e)} def ask_with_json_output(self, messages: list[dict]) - Optional[dict]: 专门用于要求JSON格式答案的提问。 result self.chat_completion(messages, json_modeTrue) if result[success]: try: return json.loads(result[content]) except json.JSONDecodeError: # 解析失败记录日志并返回原始内容 print(fJSON解析失败原始内容{result[content][:200]}...) return {answer: result[content]} return None步骤4实现Token计数监控辅助优化core/token_counter.py# 注意这是一个简单的、基于规则的估算器。精确计数应使用tiktoken库OpenAI或模型对应的tokenizer。 import tiktoken # 精确计数库需要安装 pip install tiktoken def estimate_tokens_simple(text: str) - int: 一个非常粗略的估算英文约1token4字符中文约1token2字符。仅用于演示。 # 这是一个不准确的示例。实际请使用下面的精确方法。 chinese_chars sum(\u4e00 char \u9fff for char in text) other_chars len(text) - chinese_chars return int(chinese_chars / 2 other_chars / 4) def count_tokens_precise(text: str, model: str gpt-4) - int: 使用tiktoken进行精确token计数。 try: encoding tiktoken.encoding_for_model(model) except KeyError: encoding tiktoken.get_encoding(cl100k_base) # GPT-4, GPT-3.5-turbo的编码 return len(encoding.encode(text)) # 在项目中我们使用精确计数 estimate_tokens count_tokens_precise步骤5主程序集成所有模块main.pyimport os from dotenv import load_dotenv from core.llm_client import LLMClient from core.prompt_templates import format_qa_prompt from services.document_processor import DocumentProcessor from services.cache_manager import SemanticCacheManager # 假设我们有一个缓存管理器 load_dotenv() class EfficientDocQAAgent: def __init__(self): self.llm_client LLMClient(modelgpt-4-turbo-preview) self.doc_processor DocumentProcessor() self.cache_manager SemanticCacheManager() # 初始化缓存 self.document_chunks {} # 缓存文档分块结果 {doc_id: [chunks]} def process_document(self, doc_id: str, document_text: str): 预处理文档分割并存储分块。 chunks self.doc_processor.split_document(document_text) self.document_chunks[doc_id] chunks print(f文档 {doc_id} 已处理分割为 {len(chunks)} 个片段。) # 可选将chunks存入向量数据库 def answer_question(self, doc_id: str, question: str) - dict: 回答关于特定文档的问题。 # 1. 检查语义缓存 cached_answer self.cache_manager.get(question, doc_id) if cached_answer: print(f缓存命中问题{question}) return {source: cache, answer: cached_answer, tokens_saved: significant} # 2. 检索相关文档片段 if doc_id not in self.document_chunks: return {error: f文档 {doc_id} 未处理。} all_chunks self.document_chunks[doc_id] relevant_chunks self.doc_processor.retrieve_relevant_chunks(all_chunks, question, top_k2) # 只取最相关的2段 if not relevant_chunks: return {answer: 未在文档中找到相关信息。} # 3. 构造精准提示词 messages format_qa_prompt(relevant_chunks, question) # 4. 调用LLM并请求结构化输出例如我们希望答案包含引用来源 # 首先定义我们希望JSON输出的结构 json_schema_prompt messages.copy() json_schema_prompt.append({ role: user, content: 请将你的答案以JSON格式输出包含两个字段answer字符串类型你的答案和 confidence浮点数类型0-1表示你的确信度。 }) result self.llm_client.ask_with_json_output(json_schema_prompt) if result: answer result.get(answer, 无答案) confidence result.get(confidence, 0.0) # 5. 将结果存入缓存 self.cache_manager.set(question, doc_id, answer) # 6. 记录token消耗实际应从llm_client返回结果中获取 # 这里仅作演示实际应传递并记录 print(f问题处理完毕。答案置信度{confidence:.2f}) return {source: llm, answer: answer, confidence: confidence, relevant_chunks_used: len(relevant_chunks)} else: return {error: LLM调用失败。} # 使用示例 if __name__ __main__: agent EfficientDocQAAgent() # 模拟读取文档 with open(knowledge_base/project_plan.txt, r, encodingutf-8) as f: doc_text f.read() # 预处理文档只需一次 agent.process_document(project_plan_v1, doc_text) # 进行问答 questions [ 项目第三阶段的里程碑是什么, 项目的总预算是多少, 第三阶段的里程碑是什么, # 与第一个问题语义相似应触发缓存 ] for q in questions: print(f\n用户提问{q}) response agent.answer_question(project_plan_v1, q) print(f系统回答{response.get(answer, response.get(error, 无响应))}) print(f回答来源{response.get(source, N/A)})4.3 重构前后对比与效益分析让我们从几个维度对比重构前后的效果维度重构前低效版本重构后高效版本效益提升Token消耗每次请求包含整篇文档假设10000字≈2500 token 模糊提示词。仅包含2个最相关片段假设每段200字≈50 token共100 token 精准指令。预计减少90%的输入token。提示词质量单一、模糊、包含多余描述。分层、模块化、指令清晰、要求结构化输出。输出更精准、更简洁进一步减少输出token并提升答案质量。可维护性提示词硬编码在业务逻辑中修改困难。提示词模板集中管理易于修改和A/B测试。开发迭代速度提升。扩展性添加新功能如缓存、摘要需大幅改动主逻辑。模块化设计新增服务如缓存管理器易于集成。系统架构清晰支持快速迭代。成本监控无。集成token估算和缓存命中统计成本可视化。便于进行成本分析和优化决策。正如Thoughtworks实验所证明的通过系统性的重构我们完全可以在不牺牲功能、甚至提升功能的前提下将token消耗降低一个数量级。这对于一个日调用量数万甚至数百万的生产级AI应用来说意味着每月可能节省数万至数十万元的API成本。5. 常见问题与排查思路在实施AI智能体重构以降低token消耗的过程中你可能会遇到以下典型问题问题现象可能原因排查思路与解决方案重构后答案质量下降1. 检索的相关片段不准确。2. 提示词指令过于严苛限制了模型发挥。3. 上下文片段过少信息不足。1.优化检索器检查文档分割策略尝试按语义分割优化嵌入模型和相似度阈值。2.迭代提示词进行A/B测试微调指令的严格程度在“简洁”和“准确”间找到平衡。3.调整top_k增加retrieve_relevant_chunks中的top_k参数注入更多上下文。缓存命中率低1. 语义相似度阈值设置过高。2. 缓存键设计不合理如未考虑文档ID。3. 用户问题多样性太高。1.调整相似度阈值根据业务场景调低阈值容忍一定的语义差异。2.优化缓存键将(问题语义向量, 文档ID/版本)共同作为缓存键。3.评估缓存必要性对于高度动态或个性化的场景缓存可能不适用。结构化输出解析失败1. 模型未遵守JSON格式。2. JSON模式json_mode未正确启用或模型不支持。3. 提示词中未明确要求JSON格式。1.强化指令在系统提示词和用户提示词中都明确要求输出JSON并给出示例。2.验证API支持确认所用模型支持response_format: {“type”: “json_object”}。3.添加后处理在解析失败时使用正则表达式尝试从文本中提取JSON或触发一次修正请求。Token估算与实际计费差异大1. 使用的估算方法不准确如简单的字符计数。2. 未计算系统提示词、角色消息的token。1.使用官方库对于OpenAI模型务必使用tiktoken库进行精确计数。2.计算完整消息列表估算时应将整个messages列表序列化为字符串后再计算或使用API返回的usage字段最准确。系统响应变慢1. 检索过程尤其是向量数据库查询耗时。2. 缓存查询逻辑复杂。3. 网络延迟。1.性能剖析使用 profiling 工具定位瓶颈。优化检索查询考虑对文档块建立索引。2.简化缓存检查缓存数据结构确保查询是O(1)或O(log n)复杂度。3.异步处理对于非实时性要求高的步骤如文档预处理采用异步任务。6. 最佳实践与工程建议基于Thoughtworks的实践和行业经验以下最佳实践能帮助你将重构效益最大化并构建健壮的AI智能体系统将提示词视为代码Prompt as Code版本控制像管理源代码一样使用Git管理提示词模板文件。代码审查对提示词的修改进行同行评审评估其清晰度、安全性和潜在token消耗。环境隔离为开发、测试、生产环境配置不同的提示词版本便于测试和回滚。实施成本监控与告警埋点记录在每次LLM调用时记录模型名称、输入/输出token估算值或实际值、时间戳、用户ID等。设置预算和告警在应用层面设置每日/每周token消耗预算当接近阈值时触发告警邮件、Slack等。可视化仪表盘使用Grafana、DataDog等工具创建看板监控平均每次调用成本、token消耗趋势、缓存命中率等关键指标。设计可降级的用户体验当LLM API调用失败或超时时应有备用方案。例如返回一个预定义的通用答案或引导用户简化问题。对于成本极高的操作如处理超长文档可以设置限制并提示用户“正在处理较长内容可能需要更多时间”或提供分步处理的选项。安全与合规性前置输入过滤与审查在用户输入和检索到的上下文注入LLM前进行必要的过滤防止提示词注入攻击。输出审查对模型生成的内容进行安全检查如敏感词过滤特别是在面向公众的应用中。数据隐私确保传入LLM的上下文不包含用户个人身份信息PII或商业机密必要时进行脱敏处理。持续迭代与A/B测试建立评估体系定义答案质量的评估指标如相关性、准确性、流畅度并结合成本指标每次调用token数。进行A/B测试对新旧提示词版本、不同的检索策略如top_k2vstop_k3进行线上A/B测试用数据驱动决策。定期回顾每季度或每半年回顾一次智能体的架构和提示词随着模型能力的更新和业务需求的变化新的优化机会会出现。通过将上述重构策略和最佳实践融入到你的AI智能体开发流程中你不仅能有效控制当下项目的token成本更能建立起一个可持续优化、易于维护的技术基础从容应对未来业务规模的增长和AI技术的快速演进。