AI智能体代码库重构：四大策略降低70%大模型API调用成本-平芜编程栈

最近在推进一个AI智能体项目时，团队遇到了一个棘手的问题：随着功能迭代，每次调用大模型API的成本像坐了火箭一样飙升。仔细排查后发现，问题根源并非业务逻辑复杂，而是项目早期堆砌的“Prompt工程”和臃肿的上下文，导致每次请求都携带了大量无效token。这让我想起了Thoughtworks CTO Rebecca Parsons博士团队近期分享的一个实验，他们通过系统性的代码库重构，成功将AI智能体的token消耗降低了惊人的70%。这不仅仅是技术优化，更直接关乎项目的经济效益和可持续性。本文将深入拆解这个案例，并结合实战，为你展示如何通过重构AI智能体代码库来显著降低token成本，覆盖从问题诊断、重构策略到具体代码实现的完整闭环。

1. 背景与核心概念：为什么AI智能体的token成本如此关键？

在深入重构之前，我们必须理解几个核心概念，以及它们如何与项目成本紧密挂钩。

AI智能体（AI Agent）通常指能够感知环境、进行决策并执行动作以完成特定目标的程序。在现代开发中，它常常通过与大语言模型（LLM）API（如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库：

openai>=1.0.0：OpenAI官方Python SDK。
python-dotenv>=0.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.py

import openai import os from dotenv import load_dotenv load_dotenv() client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY")) def answer_question_about_document(document_path, user_question): # 1. 读取整个文档（可能非常长） with open(document_path, 'r', encoding='utf-8') as f: full_document = f.read() # 2. 构造一个庞大而模糊的提示词 prompt = f""" 你是一个专业的文档分析助手。请基于以下文档内容，回答用户的问题。 文档内容： {full_document} 用户问题：{user_question} 请给出详细、准确的答案。 """ # 3. 发起API调用 response = client.chat.completions.create( model="gpt-4-turbo-preview", messages=[ {"role": "user", "content": prompt} ], temperature=0.7, max_tokens=1000 ) 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.py

SYSTEM_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：实现文档处理与检索服务（模拟RAG）services/document_processor.py

# 简化的文本分割与检索模拟。真实场景应使用LangChain、LlamaIndex或自定义嵌入模型+向量数据库。 import re class DocumentProcessor: def __init__(self, chunk_size=500, chunk_overlap=50): 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_k=3) -> 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(key=lambda x: x[0], reverse=True) return [chunk for _, chunk in scored_chunks[:top_k]]

步骤3：封装LLM客户端并集成结构化输出core/llm_client.py

import 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_key=os.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_mode=True) if result["success"]: try: return json.loads(result["content"]) except json.JSONDecodeError: # 解析失败，记录日志并返回原始内容 print(f"JSON解析失败，原始内容：{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: """一个非常粗略的估算：英文约1token=4字符，中文约1token=2字符。仅用于演示。""" # 这是一个不准确的示例。实际请使用下面的精确方法。 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.py

import 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(model="gpt-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_k=2) # 只取最相关的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', encoding='utf-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_k=2vstop_k=3）进行线上A/B测试，用数据驱动决策。
- 定期回顾：每季度或每半年回顾一次智能体的架构和提示词，随着模型能力的更新和业务需求的变化，新的优化机会会出现。