提示系统原型设计中的可扩展性:架构师如何设计支持功能扩展的原型?
在大模型(LLM)应用爆发的今天,提示系统(Prompt System)已经成为连接用户需求与模型能力的核心层——它负责将用户问题转化为模型能理解的提示(Prompt),管理上下文、整合工具调用,甚至优化输出结果。但很多团队在原型设计阶段容易陷入“快速实现”的陷阱:硬编码提示模板、直接耦合单一模型、缺乏扩展机制,导致后期要加功能(比如支持多模态、切换模型、整合工具)时,不得不重构核心代码,效率极低
提示系统原型设计的可扩展性指南:架构师如何打造支持功能扩展的基础框架
副标题:从模块化到插件化,构建能应对未来需求的大模型提示系统
摘要/引言
在大模型(LLM)应用爆发的今天,提示系统(Prompt System) 已经成为连接用户需求与模型能力的核心层——它负责将用户问题转化为模型能理解的提示(Prompt),管理上下文、整合工具调用,甚至优化输出结果。但很多团队在原型设计阶段容易陷入“快速实现”的陷阱:硬编码提示模板、直接耦合单一模型、缺乏扩展机制,导致后期要加功能(比如支持多模态、切换模型、整合工具)时,不得不重构核心代码,效率极低。
本文要解决的核心问题是:如何在提示系统原型阶段就植入可扩展性设计,让系统能快速适应未来的功能迭代?
我们的解决方案是:通过 分层架构+模块化组件+插件化扩展+配置驱动 的设计原则,将提示系统拆解为“稳定核心”与“可变扩展点”,让新功能的添加无需修改核心逻辑。
读完本文,你将掌握:
- 提示系统可扩展性的核心设计维度;
- 如何用分层架构隔离变化;
- 如何通过插件化机制快速扩展功能;
- 一个可落地的可扩展提示系统原型实现。
目标读者与前置知识
目标读者
- 负责大模型应用架构设计的 资深工程师/架构师;
- 正在开发提示系统(如RAG、Agent)的 算法工程师;
- 想提升原型扩展性的 技术负责人。
前置知识
- 了解大模型基础:熟悉Prompt Engineering、LLM调用流程;
- 掌握软件架构原则:理解分层、模块化、依赖倒置(SOLID);
- 会用Python开发:能读懂FastAPI、Pydantic代码;
- (可选)接触过LangChain/LlamaIndex:但本文会避免绑定特定框架。
文章目录
- 引言与基础
- 问题背景:为什么提示系统需要可扩展性?
- 核心概念:可扩展性的4个设计维度
- 架构设计:分层+模块化的核心框架
- 分步实现:从0到1构建可扩展原型
- 关键设计解析:为什么要这么做?
- 验证与扩展:测试、优化与未来方向
- 总结
一、问题背景:为什么提示系统需要可扩展性?
在讨论设计之前,我们先明确 “提示系统”的核心职责:
- 接收用户输入(文本/图像/语音);
- 管理上下文(多轮对话、历史记录);
- 渲染提示模板(结合用户问题+上下文+工具结果);
- 调用大模型(OpenAI/Claude/本地模型);
- 解析输出(格式化为用户需要的结构);
- 扩展功能(如工具调用、多模态处理)。
现有原型的痛点
很多团队的初始原型是 “单体式” 的:
- 提示模板硬编码在代码里(比如
prompt = f"回答用户问题:{user_input}"
); - 直接调用OpenAI API(
openai.ChatCompletion.create(...)
); - 上下文存在内存里(
context = {}
); - 功能耦合(比如工具调用逻辑写在提示生成函数里)。
这种原型的问题是 “抗变化能力弱”:
- 要支持Claude模型?得改所有模型调用的代码;
- 要加图像输入?得重构整个输入处理流程;
- 要整合天气工具?得修改提示生成的核心逻辑;
- 要换上下文存储(比如从内存到Redis)?得改所有上下文操作的代码。
可扩展性的价值
可扩展的提示系统原型能帮你:
- 快速响应需求:加新模型/工具/模态时,不用改核心代码;
- 降低维护成本:核心逻辑稳定,扩展功能通过“插件”实现;
- 支持团队协作:不同角色(算法/后端/产品)可以并行开发扩展点;
- 应对未来变化:大模型技术迭代快,可扩展架构能“兼容未来”。
二、核心概念:可扩展性的4个设计维度
要设计可扩展的提示系统,首先得明确 “扩展什么”——也就是可扩展性的核心维度:
1. 模型扩展:支持多模型切换
- 需求:能快速切换OpenAI、Claude、Gemini、本地Llama 3等模型;
- 设计目标:核心逻辑不依赖具体模型,新增模型只需加“适配器”。
2. 功能扩展:支持插件化新增功能
- 需求:能加工具调用、多模态处理、输出格式校验等功能;
- 设计目标:新功能通过“插件”注入,不修改核心流程。
3. 数据扩展:支持多模态输入输出
- 需求:能处理文本、图像、语音等输入,输出结构化数据(JSON/表格);
- 设计目标:输入/输出层抽象,支持新增数据类型。
4. 配置扩展:支持动态调整参数
- 需求:能通过配置文件修改模型参数(温度、top_p)、模板路径、插件启用状态;
- 设计目标:核心逻辑由配置驱动,无需重启服务改参数。
三、架构设计:分层+模块化的核心框架
基于上述维度,我们提出 “四层三组件” 的可扩展架构:
1. 架构分层(从外到内)
层级 | 职责 | 变化频率 | 设计原则 |
---|---|---|---|
API层 | 接收用户请求(HTTP/GRPC)、返回结果 | 低 | 轻量、标准化(RESTful) |
核心服务层 | 业务逻辑 orchestration(调度组件) | 中 | 依赖抽象、无状态 |
扩展组件层 | 可替换的功能模块(模型/模板/上下文) | 高 | 模块化、插件化 |
基础设施层 | 底层依赖(数据库、缓存、模型API) | 低 | 抽象化、可配置 |
2. 核心组件(扩展点)
核心服务层通过 依赖抽象 调用扩展组件,实现“核心稳定、扩展灵活”:
- 提示模板引擎(Prompt Template Engine):管理提示模板,支持动态渲染;
- 模型适配器(Model Adapter):抽象模型调用,支持多模型切换;
- 上下文管理器(Context Manager):管理多轮对话上下文,支持多种存储;
- 插件系统(Plugin System):注入自定义逻辑(如工具调用、多模态处理)。
四、分步实现:从0到1构建可扩展原型
接下来我们用 Python+FastAPI 实现一个可扩展的提示系统原型,覆盖上述所有设计点。
环境准备
- 安装依赖:
pip install fastapi uvicorn pydantic jinja2 openai anthropic python-dotenv
- 项目结构:
prompt-system-prototype/ ├── app/ │ ├── api/ # API层 │ │ └── endpoints.py │ ├── core/ # 核心服务层 │ │ ├── service.py # 核心业务逻辑 │ │ └── config.py # 配置管理 │ ├── components/ # 扩展组件层 │ │ ├── template.py # 提示模板引擎 │ │ ├── model.py # 模型适配器 │ │ ├── context.py # 上下文管理器 │ │ └── plugin.py # 插件系统 │ └── plugins/ # 插件实现 │ └── tool_caller.py # 工具调用插件 ├── templates/ # 提示模板文件 │ ├── text.j2 │ └── multimodal.j2 ├── .env # 环境变量 └── main.py # 启动文件
步骤1:配置驱动设计(核心服务的“开关”)
首先实现 配置管理,让核心逻辑由配置驱动。用Pydantic管理配置,支持从.env
文件加载:
app/core/config.py
:
from pydantic_settings import BaseSettings
from typing import List, Type
from app.components.model import BaseModelAdapter
from app.components.plugin import BasePlugin
class Settings(BaseSettings):
# 模型配置
DEFAULT_MODEL: str = "openai"
MODEL_ADAPTERS: List[Type[BaseModelAdapter]] = [] # 注册的模型适配器
# 模板配置
TEMPLATE_DIR: str = "templates"
DEFAULT_TEMPLATE: str = "text.j2"
# 上下文配置
CONTEXT_STORE_TYPE: str = "memory" # 可选:memory/redis
# 插件配置
ENABLED_PLUGINS: List[Type[BasePlugin]] = [] # 启用的插件
class Config:
env_file = ".env"
# 全局配置实例
settings = Settings()
步骤2:实现扩展组件的抽象接口
扩展组件的核心是 面向接口编程——定义抽象基类(ABC),让核心服务依赖抽象而非具体实现。
(1)模型适配器抽象(支持多模型)
app/components/model.py
:
from abc import ABC, abstractmethod
from pydantic import BaseModel
from typing import Dict, Any
# 模型输入输出的标准化定义
class ModelInput(BaseModel):
prompt: str
model_params: Dict[str, Any] = {} # 温度、top_p等参数
class ModelOutput(BaseModel):
text: str
raw_response: Dict[str, Any] # 模型原始响应(便于调试)
# 模型适配器抽象基类
class BaseModelAdapter(ABC):
model_name: str # 适配器对应的模型名(如"openai")
@abstractmethod
def generate(self, input: ModelInput) -> ModelOutput:
pass
# 注册模型适配器到配置(示例:OpenAI适配器)
from openai import OpenAI
class OpenAIAdapter(BaseModelAdapter):
model_name: str = "openai"
def __init__(self):
self.client = OpenAI()
def generate(self, input: ModelInput) -> ModelOutput:
response = self.client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": input.prompt}],
**input.model_params
)
return ModelOutput(
text=response.choices[0].message.content,
raw_response=response.dict()
)
# 将适配器注册到配置
settings.MODEL_ADAPTERS.append(OpenAIAdapter)
(2)提示模板引擎抽象(支持多模板)
用Jinja2实现模板渲染,支持加载不同模板文件:app/components/template.py
:
from abc import ABC, abstractmethod
from jinja2 import Environment, FileSystemLoader
from app.core.config import settings
class BaseTemplateEngine(ABC):
@abstractmethod
def render(self, template_name: str, context: Dict[str, Any]) -> str:
pass
class Jinja2TemplateEngine(BaseTemplateEngine):
def __init__(self):
self.env = Environment(
loader=FileSystemLoader(settings.TEMPLATE_DIR),
autoescape=True
)
def render(self, template_name: str, context: Dict[str, Any]) -> str:
template = self.env.get_template(template_name)
return template.render(context)
# 全局模板引擎实例
template_engine = Jinja2TemplateEngine()
模板文件示例(templates/text.j2
):
{% if context.history %}
历史对话:{{ context.history | join('\n') }}
{% endif %}
用户现在的问题:{{ user_input }}
请用简洁的语言回答。
(3)上下文管理器抽象(支持多存储)
app/components/context.py
:
from abc import ABC, abstractmethod
from typing import Dict, Optional
from app.core.config import settings
class BaseContextStore(ABC):
@abstractmethod
def get(self, key: str) -> Optional[Dict[str, Any]]:
pass
@abstractmethod
def set(self, key: str, value: Dict[str, Any]) -> None:
pass
@abstractmethod
def delete(self, key: str) -> None:
pass
# 内存存储实现(默认)
class MemoryContextStore(BaseContextStore):
def __init__(self):
self.store: Dict[str, Dict[str, Any]] = {}
def get(self, key: str) -> Optional[Dict[str, Any]]:
return self.store.get(key)
def set(self, key: str, value: Dict[str, Any]) -> None:
self.store[key] = value
def delete(self, key: str) -> None:
self.store.pop(key, None)
# 根据配置创建上下文存储实例
def create_context_store() -> BaseContextStore:
if settings.CONTEXT_STORE_TYPE == "memory":
return MemoryContextStore()
# 未来加Redis存储只需加分支:elif settings.CONTEXT_STORE_TYPE == "redis": ...
else:
raise ValueError(f"Unsupported context store type: {settings.CONTEXT_STORE_TYPE}")
context_store = create_context_store()
步骤3:实现插件系统(功能扩展的核心)
插件系统是 “不修改核心代码加功能” 的关键。我们定义插件的生命周期钩子:
app/components/plugin.py
:
from abc import ABC, abstractmethod
from typing import Dict, Any
from app.components.model import ModelInput, ModelOutput
# 插件上下文:传递核心数据
class PluginContext(BaseModel):
user_input: str # 用户原始输入
context: Dict[str, Any] # 对话上下文
model_input: ModelInput # 模型输入(渲染后的提示)
model_output: ModelOutput # 模型输出
config: Dict[str, Any] # 插件配置
# 插件抽象基类:定义生命周期钩子
class BasePlugin(ABC):
plugin_name: str # 插件名
priority: int = 0 # 执行顺序(数字小先执行)
@abstractmethod
def pre_process(self, plugin_ctx: PluginContext) -> PluginContext:
"""生成提示前的处理(如工具调用、补充上下文)"""
pass
@abstractmethod
def post_process(self, plugin_ctx: PluginContext) -> PluginContext:
"""模型输出后的处理(如格式校验、结果优化)"""
pass
# 插件管理器:加载、执行插件
class PluginManager:
def __init__(self, plugins: List[Type[BasePlugin]]):
# 按priority排序插件
self.plugins = sorted([p() for p in plugins], key=lambda x: x.priority)
def run_pre_process(self, plugin_ctx: PluginContext) -> PluginContext:
for plugin in self.plugins:
plugin_ctx = plugin.pre_process(plugin_ctx)
return plugin_ctx
def run_post_process(self, plugin_ctx: PluginContext) -> PluginContext:
for plugin in self.plugins:
plugin_ctx = plugin.post_process(plugin_ctx)
return plugin_ctx
# 全局插件管理器实例
plugin_manager = PluginManager(settings.ENABLED_PLUGINS)
插件示例:工具调用插件
实现一个“天气查询”插件,当用户问天气时,自动调用天气API补充上下文:
app/plugins/tool_caller.py
:
import requests
from app.components.plugin import BasePlugin, PluginContext
from app.core.config import settings
class ToolCallerPlugin(BasePlugin):
plugin_name: str = "tool_caller"
priority: int = 10 # 先于其他插件执行
def pre_process(self, plugin_ctx: PluginContext) -> PluginContext:
# 检测用户输入是否包含天气关键词
if "天气" in plugin_ctx.user_input:
# 调用天气API(示例用免费接口)
city = self._extract_city(plugin_ctx.user_input)
weather_data = self._get_weather(city)
# 将天气数据加到上下文
plugin_ctx.context["weather"] = weather_data
# 更新模型输入的提示(补充天气信息)
plugin_ctx.model_input.prompt += f"\n当前{city}的天气:{weather_data}"
return plugin_ctx
def post_process(self, plugin_ctx: PluginContext) -> PluginContext:
# 无需后处理,直接返回
return plugin_ctx
def _extract_city(self, user_input: str) -> str:
# 简单提取城市名(实际需更 robust 的NLP处理)
cities = ["北京", "上海", "广州", "深圳"]
for city in cities:
if city in user_input:
return city
return "北京" # 默认北京
def _get_weather(self, city: str) -> str:
# 调用免费天气API(示例:https://www.tianqiapi.com/)
api_key = "your_api_key"
url = f"https://v0.yiketianqi.com/api?unescape=1&version=v61&appid=xxx&appsecret=xxx&city={city}"
response = requests.get(url)
data = response.json()
return f"{data['wea']},温度{data['tem']}℃"
# 注册插件到配置
settings.ENABLED_PLUGINS.append(ToolCallerPlugin)
步骤4:实现核心服务层(业务逻辑调度)
核心服务层负责 调度扩展组件,实现端到端的提示生成流程:
app/core/service.py
:
from typing import Dict, Any
from app.core.config import settings
from app.components.model import ModelInput, ModelOutput, BaseModelAdapter
from app.components.template import template_engine
from app.components.context import context_store
from app.components.plugin import PluginContext, plugin_manager
class PromptService:
def __init__(self):
# 根据配置加载默认模型适配器
self.model_adapters = {
adapter.model_name: adapter()
for adapter in settings.MODEL_ADAPTERS
}
self.default_model = self.model_adapters[settings.DEFAULT_MODEL]
def generate(self, user_input: str, user_id: str, model_name: str = None) -> ModelOutput:
# 1. 获取上下文(用户ID作为key)
context = context_store.get(user_id) or {"history": []}
# 2. 渲染提示模板
template_context = {
"user_input": user_input,
"context": context
}
prompt = template_engine.render(settings.DEFAULT_TEMPLATE, template_context)
# 3. 初始化模型输入
model_input = ModelInput(prompt=prompt)
# 4. 执行插件pre_process
plugin_ctx = PluginContext(
user_input=user_input,
context=context,
model_input=model_input,
model_output=None,
config={}
)
plugin_ctx = plugin_manager.run_pre_process(plugin_ctx)
# 5. 调用模型
model = self.model_adapters.get(model_name, self.default_model)
model_output = model.generate(plugin_ctx.model_input)
# 6. 执行插件post_process
plugin_ctx.model_output = model_output
plugin_ctx = plugin_manager.run_post_process(plugin_ctx)
# 7. 更新上下文(保存历史对话)
context["history"].append(f"用户:{user_input}")
context["history"].append(f"助手:{plugin_ctx.model_output.text}")
context_store.set(user_id, context)
# 8. 返回结果
return plugin_ctx.model_output
# 全局核心服务实例
prompt_service = PromptService()
步骤5:实现API层(对外接口)
用FastAPI实现RESTful接口,接收用户请求:
app/api/endpoints.py
:
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from app.core.service import prompt_service
from app.components.model import ModelOutput
router = APIRouter()
# 请求体定义
class GenerateRequest(BaseModel):
user_input: str
user_id: str
model_name: str = None # 可选:指定模型
# 响应体定义
class GenerateResponse(BaseModel):
result: str
raw_response: Dict[str, Any]
@router.post("/generate", response_model=GenerateResponse)
async def generate(request: GenerateRequest):
try:
output = prompt_service.generate(
user_input=request.user_input,
user_id=request.user_id,
model_name=request.model_name
)
return GenerateResponse(
result=output.text,
raw_response=output.raw_response
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
步骤6:启动服务
main.py
:
from fastapi import FastAPI
from app.api.endpoints import router as api_router
app = FastAPI(title="可扩展提示系统原型", version="1.0")
# 注册API路由
app.include_router(api_router, prefix="/api")
if __name__ == "__main__":
import uvicorn
uvicorn.run("main.py:app", host="0.0.0.0", port=8000, reload=True)
五、关键设计解析:为什么要这么做?
1. 为什么用抽象基类(ABC)?
依赖倒置原则:核心服务依赖抽象的BaseModelAdapter
,而不是具体的OpenAIAdapter
。这样新增Claude模型时,只需实现ClaudeAdapter
并注册到配置,无需修改核心服务的代码。
比如加Claude适配器:
from anthropic import Anthropic
class ClaudeAdapter(BaseModelAdapter):
model_name: str = "claude"
def __init__(self):
self.client = Anthropic()
def generate(self, input: ModelInput) -> ModelOutput:
response = self.client.messages.create(
model="claude-3-sonnet-20240229",
max_tokens=1000,
messages=[{"role": "user", "content": input.prompt}]
)
return ModelOutput(
text=response.content[0].text,
raw_response=response.dict()
)
# 注册到配置
settings.MODEL_ADAPTERS.append(ClaudeAdapter)
2. 为什么用插件系统?
开放封闭原则:核心流程对扩展开放(可以加插件),对修改封闭(不用改核心代码)。比如加“多模态处理”插件时,只需实现MultimodalPlugin
,在pre_process
中处理图像输入,无需修改PromptService
的逻辑。
3. 为什么用配置驱动?
分离配置与代码:模型参数、模板路径、插件启用状态都放在配置文件中,修改时无需重启服务(如果用动态配置中心如Nacos,还能实时生效)。比如要切换默认模型,只需改.env
文件:
DEFAULT_MODEL=claude
六、验证与扩展
1. 功能验证
启动服务后,用Postman调用POST /api/generate
接口:
-
请求体(调用OpenAI):
{ "user_input": "北京的天气怎么样?", "user_id": "user_123", "model_name": "openai" }
-
响应(工具调用插件生效):
{ "result": "当前北京的天气是晴,温度25℃。北京的主要景点有故宫、长城、颐和园等...", "raw_response": { ... } }
-
请求体(切换Claude模型):
{ "user_input": "北京的天气怎么样?", "user_id": "user_123", "model_name": "claude" }
-
响应(Claude模型返回结果):
{ "result": "根据最新数据,北京今日晴,气温25℃。推荐你去故宫参观,提前预约门票哦~", "raw_response": { ... } }
2. 性能优化与最佳实践
- 缓存模板:将常用模板缓存到内存,避免每次渲染都读取文件;
- 异步模型调用:用
async
/await
优化模型API调用,提升并发; - 上下文过期:给上下文加过期时间,避免存储过多历史数据;
- 插件懒加载:只加载启用的插件,减少初始化时间;
- 监控扩展点:给插件、模型适配器加监控(如调用次数、耗时),便于排查问题。
3. 未来扩展方向
- 动态插件加载:支持从远程仓库加载插件(如Git),无需重启服务;
- 可视化配置:做一个Web界面管理模板、模型、插件配置;
- AI辅助优化:加一个“提示优化插件”,自动调整提示模板(如根据模型反馈优化prompt);
- 多租户支持:给不同用户/租户配置不同的模型、插件、模板;
- 多模态扩展:加
MultimodalAdapter
支持图像输入,MultimodalTemplate
处理图文混合提示。
七、总结
设计可扩展的提示系统原型,核心是 “分离稳定与变化”:
- 稳定的核心:API层、核心服务层(调度逻辑);
- 可变的扩展点:模型适配器、提示模板、上下文存储、插件。
通过 分层架构+模块化组件+插件化扩展+配置驱动 的设计,你可以:
- 快速支持新模型(加适配器);
- 快速新增功能(加插件);
- 快速调整配置(改配置文件);
- 避免后期重构的痛苦。
最后送你一句架构设计的名言:“今天的最佳设计,是能应对明天变化的设计”。希望本文的思路能帮你打造一个“抗造”的提示系统原型!
参考资料
- FastAPI官方文档:https://fastapi.tiangolo.com/
- Pydantic配置管理:https://docs.pydantic.dev/latest/usage/settings/
- 依赖倒置原则(DIP):https://en.wikipedia.org/wiki/Dependency_inversion_principle
- LangChain的模块化设计:https://python.langchain.com/docs/modules/
- 插件化架构设计:https://martinfowler.com/articles/modularization-patterns.html
附录:完整代码仓库
GitHub地址:https://github.com/your-name/prompt-system-prototype
包含:
- 完整的项目结构;
- 示例模板、插件、配置文件;
- Postman测试用例;
- Dockerfile(一键部署)。
欢迎Star、Fork,如有问题请提Issue!
更多推荐
所有评论(0)