1. 项目概述为什么我们需要解耦LLM应用如果你正在构建一个基于大语言模型的应用无论是智能客服、代码助手还是数据分析工具你可能已经感受到了那种“牵一发而动全身”的焦虑。今天改一个提示词明天换一个模型API后天新增一个工具函数整个系统就像一堆积木动一块就可能全盘散架。测试更是无从下手难道每次都要花真金白银去调用昂贵的API或者冒着数据泄露的风险用真实用户对话来验证吗这正是“LLM 应用的依赖注入工程实践”要解决的核心痛点。这个标题听起来很学术但它的内核非常务实通过一套清晰的工程架构将你的AI系统中那些最常变动、最需要独立管理的部分——模型客户端、提示词模板和工具注册表——彻底分离开。让它们从紧密耦合的“铁板一块”变成可以像乐高积木一样自由插拔、独立测试的模块。想象一下你的应用核心逻辑是“大脑”它需要“眼睛”去看Client调用模型、“嘴巴”去说Prompt生成指令、“手”去操作Tool执行具体功能。传统做法是把眼睛、嘴巴、手都焊死在大脑上。而依赖注入就是为大脑设计了一套标准的接口让它可以随时更换不同品牌、不同型号的眼睛、嘴巴和手甚至可以在测试时给大脑接上一个“模拟眼睛”和“假手”完全在本地、零成本地验证大脑的逻辑是否正确。这不仅仅是代码整洁度的问题它直接关系到项目的生死存亡。模型迭代速度以月计今天用的GPT-4明天可能就要评估Claude 3业务需求瞬息万变针对不同用户的提示词需要A/B测试工具链更是会不断膨胀。一个不可测试、不可替换的系统其维护成本会指数级上升最终沦为技术债的泥潭。接下来我们就深入拆解如何一步步构建这样一个既健壮又灵活的系统。2. 核心架构设计Client、Prompt与Tool Registry的三权分立要解耦首先得明确什么该被解耦。在LLM应用这个上下文里经过大量实践我总结出三个最不稳定、最需要被独立管理的核心依赖也就是标题中的三位主角。2.1 模型客户端抽象的“对话能力”Client在这里特指与大语言模型服务进行通信的客户端。它不应该是一个具体的OpenAI或Anthropic的SDK实例而应该是一个抽象的接口。这个接口只定义最基本的能力generate生成和generate_stream流式生成。至于背后是调用GPT-4、Gemini还是你本地部署的Llama 3应用的核心业务逻辑完全不关心。为什么必须抽象第一避免供应商锁定。你的业务逻辑里如果散落着openai.ChatCompletion.create的调用哪天想切到Azure OpenAI或者DeepSeek那就是一场灾难性的全局搜索替换。第二便于测试。你可以轻松创建一个MockClient在测试时返回预设的答案无需网络、无需API密钥、瞬间运行。第三统一监控和治理。你可以在抽象的Client实现层统一添加日志、计量、熔断、重试等跨切面关注点而不是在每个调用处重复编写。一个简单的Client接口定义可能长这样from abc import ABC, abstractmethod from typing import AsyncGenerator class LLMClient(ABC): abstractmethod async def generate(self, messages: list[dict]) - str: 同步生成文本 pass abstractmethod async def generate_stream(self, messages: list[dict]) - AsyncGenerator[str, None]: 流式生成文本 pass2.2 提示词模板可配置的“意图指令”Prompt是LLM应用的灵魂但它也是最容易变成“魔法字符串”散落在代码各处的东西。一个复杂的系统可能有几十上百种提示词欢迎语、总结摘要、代码审查、情感分析……把提示词硬编码在业务逻辑里意味着任何微调都需要改代码、走发布流程。我们的目标是将Prompt工程化。把提示词变成可被查找、可被管理、甚至可被动态渲染的模板。一个PromptTemplate对象应该包含模板内容、所需的输入变量、以及可选的描述信息。这样业务逻辑只需要说“给我取code_review_prompt这个模板并把code变量填充进去”而不需要关心模板具体是什么。更进阶一点你可以建立一个提示词仓库支持版本管理、A/B测试和热更新。业务代码通过一个统一的PromptManager来获取模板彻底实现逻辑与内容的分离。2.3 工具注册表动态的“技能包”Tool Registry工具注册表是Agent类应用的核心。它管理着LLM可以调用的所有函数工具比如“查询天气”、“发送邮件”、“执行SQL”。一个糟糕的实现是在初始化Agent时直接把一堆工具函数作为参数传进去。这会导致工具的定义、描述和注册逻辑与业务代码深度耦合。优雅的做法是建立一个中心化的注册表。这个注册表负责工具的注册、发现和描述生成。每个工具在注册时需要提供其函数本体、自然语言描述、以及参数的模式定义。当Agent需要决定使用什么工具时它向注册表查询可用的工具列表及其描述。当Agent决定调用某个工具时注册表负责找到对应的函数并执行。这样做的好处是巨大的你可以根据不同的用户、场景动态加载不同的工具集可以统一为所有工具添加权限校验、日志记录在测试时可以注册一些“模拟工具”来验证Agent的调用逻辑而无需真正发送邮件或查询数据库。2.4 依赖注入将它们编织在一起的粘合剂现在我们有了三个独立的组件LLMClient、PromptManager、ToolRegistry。我们的核心业务类比如一个CustomerServiceAgent需要它们。依赖注入框架如Spring之于Java或dependency-injector、injector之于Python的作用就是充当一个智能的“装配工”。你不再在CustomerServiceAgent的构造函数里new一个具体的Client而是声明“我需要一个LLMClient和PromptManager”。在应用启动时依赖注入容器会根据配置将配置好的OpenAIClient实例、YamlPromptManager实例和DefaultToolRegistry实例“注入”到CustomerServiceAgent的成员变量中。这个反转控制的过程是解耦的关键。业务类不再负责依赖的创建和生命周期管理它只负责使用接口。所有的配置、组装、替换工作都集中在容器初始化这一个地方。从“我要什么我自己造”变成了“我需要什么你提供给我”。这使得单元测试变得极其简单在测试中你可以给容器配置一套用于测试的Mock依赖然后轻松测试业务类的所有逻辑。3. 实战从零搭建一个可测试的AI智能体服务理论说再多不如动手搭一个。我们以一个简单的“智能任务执行助手”为例它可以根据用户描述调用合适的工具完成任务。我们将使用Python语言和pytest进行演示依赖注入框架选用轻量级的dependency-injector。3.1 第一步定义核心接口与领域模型首先我们定义最核心的几个抽象这是系统的基石。# 1. 模型客户端抽象 class LLMClient(Protocol): def generate(self, messages: List[Dict[str, str]]) - str: ... async def generate_async(self, messages: List[Dict[str, str]]) - str: ... # 2. 提示词管理器抽象 class PromptManager(Protocol): def get_template(self, name: str, **variables) - str: ... # 3. 工具定义 from pydantic import BaseModel class Tool(BaseModel): name: str description: str func: Callable parameters_schema: Dict # 简化版实际可用JSON Schema # 4. 工具注册表抽象 class ToolRegistry(Protocol): def register(self, tool: Tool) - None: ... def get_tool(self, name: str) - Optional[Tool]: ... def get_descriptions(self) - str: # 返回给LLM的工具描述文本注意这里使用了Python的typing.Protocol来定义接口这是一种“鸭子类型”的接口定义方式比ABC更灵活。Pydantic的BaseModel用于工具的数据验证和序列化。3.2 第二步实现具体的依赖组件接着我们实现这些接口的具体版本。这里以OpenAI和内存存储为例。# 具体Client实现 import openai class OpenAIClient: def __init__(self, api_key: str, model: str gpt-4): self.client openai.OpenAI(api_keyapi_key) self.model model def generate(self, messages): response self.client.chat.completions.create( modelself.model, messagesmessages ) return response.choices[0].message.content # 基于内存字典的PromptManager class InMemoryPromptManager: def __init__(self): self._templates { task_decompose: 你是一个任务分解助手。用户目标是{user_goal}。请将目标分解为清晰的步骤。, tool_choice: 根据用户请求和可用工具选择最合适的工具。请求{request}。工具列表{tool_list}。 } def get_template(self, name: str, **variables): template self._templates.get(name) if not template: raise ValueError(fPrompt template {name} not found.) return template.format(**variables) # 简单的工具注册表实现 class SimpleToolRegistry: def __init__(self): self._tools: Dict[str, Tool] {} def register(self, tool: Tool): self._tools[tool.name] tool def get_tool(self, name: str): return self._tools.get(name) def get_descriptions(self): desc [] for name, tool in self._tools.items(): desc.append(f- {name}: {tool.description}) return \n.join(desc)3.3 第三步构建核心业务逻辑现在我们可以构建不依赖于具体实现的智能体了。class TaskAgent: # 通过构造函数声明依赖而不是在内部创建 def __init__(self, llm_client: LLMClient, prompt_manager: PromptManager, tool_registry: ToolRegistry): self.llm llm_client self.prompts prompt_manager self.tools tool_registry async def execute(self, user_request: str) - str: # 1. 使用PromptManager获取模板并渲染 decomposition_prompt self.prompts.get_template( task_decompose, user_goaluser_request ) # 2. 使用抽象的LLMClient进行调用 plan await self.llm.generate_async([ {role: user, content: decomposition_prompt} ]) # 3. 让LLM根据工具描述决定使用哪个工具 tool_choice_prompt self.prompts.get_template( tool_choice, requestuser_request, tool_listself.tools.get_descriptions() ) tool_decision await self.llm.generate_async([ {role: user, content: tool_choice_prompt} ]) # ... 解析LLM返回调用ToolRegistry中的工具执行 return f计划{plan} 执行决策{tool_decision}关键点TaskAgent的代码里没有任何一行涉及openai、yaml读提示词文件或者具体的工具函数。它只和三个抽象接口对话。这就是“依赖倒置”原则的体现高层模块TaskAgent不依赖于低层模块的具体实现二者都依赖于抽象。3.4 第四步使用依赖注入容器进行组装最后我们使用dependency-injector来扮演“装配工”的角色。from dependency_injector import containers, providers class Container(containers.DeclarativeContainer): # 配置信息可以作为Provider config providers.Configuration() # 定义每个依赖的提供方式单例模式 llm_client providers.Singleton( OpenAIClient, api_keyconfig.openai.api_key, modelconfig.openai.model ) prompt_manager providers.Singleton(InMemoryPromptManager) tool_registry providers.Singleton(SimpleToolRegistry) # 定义业务对象并声明其依赖 task_agent providers.Factory( TaskAgent, llm_clientllm_client, prompt_managerprompt_manager, tool_registrytool_registry ) # 应用启动时初始化容器 def create_app(): container Container() # 可以从环境变量或配置文件中加载配置 container.config.openai.api_key.from_env(OPENAI_API_KEY) container.config.openai.model.from_value(gpt-3.5-turbo) # 注册一些工具 registry container.tool_registry() registry.register(Tool(nameget_time, description获取当前时间, funclambda: datetime.now().isoformat(), parameters_schema{})) registry.register(Tool(namesearch_web, description网络搜索, funcsearch_function, parameters_schema{query: {type: string}})) # 获取完全组装好的Agent实例 agent container.task_agent() return agent现在整个应用的依赖关系都在Container类中一目了然。要更换模型只需修改llm_client的提供者。要更换提示词存储方式只需实现一个新的PromptManager并替换提供者。所有改动被隔离在容器配置中业务代码TaskAgent无需任何变动。4. 测试策略如何对解耦后的系统进行高效验证解耦的最大收益之一就是可测试性的巨大提升。我们的测试可以分为三个层次单元测试、集成测试和端到端测试其中前两者在解耦架构下会变得非常高效。4.1 单元测试Mock一切外部依赖单元测试只关心TaskAgent自身的逻辑是否正确。我们可以使用unittest.mock来创建所有依赖的模拟对象。import pytest from unittest.mock import Mock, AsyncMock pytest.mark.asyncio async def test_task_agent_execute_logic(): # 1. 创建Mock依赖 mock_client AsyncMock(specLLMClient) mock_prompts Mock(specPromptManager) mock_registry Mock(specToolRegistry) # 2. 预设Mock行为 mock_prompts.get_template.side_effect lambda name, **vars: fMocked prompt for {name} with {vars} mock_client.generate_async.return_value Mocked LLM response: Use tool A. mock_registry.get_descriptions.return_value - tool_a: A mock tool # 3. 注入Mock创建被测对象 agent TaskAgent(llm_clientmock_client, prompt_managermock_prompts, tool_registrymock_registry) # 4. 执行测试 result await agent.execute(test request) # 5. 验证交互逻辑 # 断言PromptManager被以正确的参数调用 mock_prompts.get_template.assert_any_call(task_decompose, user_goaltest request) mock_prompts.get_template.assert_any_call(tool_choice, requesttest request, tool_list- tool_a: A mock tool) # 断言LLMClient被调用了两次 assert mock_client.generate_async.call_count 2 # 断言最终结果包含我们的Mock响应 assert Mocked LLM response in result这种测试运行速度极快毫秒级不依赖网络和外部API可以轻松覆盖各种分支逻辑比如LLM返回不同格式时Agent的解析逻辑是否正确。4.2 集成测试验证组件间的真实协作集成测试用于验证我们的具体实现如OpenAIClient、InMemoryPromptManager是否能正确地协同工作。这里我们仍然要避免调用真实API但可以使用一些测试专用工具。# 使用 pytest-httpx 来Mock HTTP请求测试OpenAIClient的逻辑 import httpx import pytest from respx import MockRouter pytest.mark.asyncio async def test_openai_client_integration(respx_mock: MockRouter): # 1. Mock OpenAI API的端点 respx_mock.post(https://api.openai.com/v1/chat/completions).mock( return_valuehttpx.Response(200, json{ choices: [{message: {content: Mocked API response}}] }) ) # 2. 使用真实的OpenAIClient类但它发出的请求会被拦截 client OpenAIClient(api_keyfake_key, modelgpt-3.5-turbo) # 3. 执行调用 response await client.generate_async([{role: user, content: Hello}]) # 4. 验证 assert response Mocked API response # 可以进一步验证发出的请求体格式是否正确对于ToolRegistry和PromptManager的集成测试则可以直接使用它们的内存实现验证工具注册、查找和提示词渲染功能是否正常。4.3 使用测试专用容器进行组件测试依赖注入容器在测试时能发挥更大威力。我们可以创建一个专门用于测试的容器覆盖掉所有外部依赖。class TestContainer(containers.DeclarativeContainer): # 覆盖父容器中的Provider提供测试专用的Mock对象 llm_client providers.Singleton(MockLLMClient) # 一个返回固定答案的测试Client prompt_manager providers.Singleton(MockPromptManager) tool_registry providers.Singleton(MockToolRegistry) # task_agent 的依赖会自动被替换成上面的Mock pytest.fixture def test_agent(): container TestContainer() yield container.task_agent() def test_with_test_container(test_agent): result test_agent.execute(test) # 进行断言...通过这种方式你可以为不同的测试场景如异常流测试、性能测试创建不同的测试容器管理测试依赖变得和配置生产依赖一样清晰简单。5. 高级模式与演进让架构适应复杂场景基础的三层解耦已经能解决80%的问题。但随着系统复杂化你可能会遇到更多挑战下面分享几个进阶模式。5.1 动态依赖与运行时上下文有时依赖不是在启动时就能确定的。例如同一个服务需要根据请求中的用户ID选择使用该用户专属的模型配置或提示词集。这需要将依赖注入与运行时上下文Context结合。一种模式是使用“工厂”或“作用域”依赖。在请求入口处根据上下文信息如用户信息动态创建或选择一组依赖然后注入到处理该请求的组件中。许多DI框架如fastapi的Depends原生支持这种模式。from dependency_injector import providers class UserAwareContainer(containers.DeclarativeContainer): # 定义一个工厂它依赖一个“用户上下文”来创建Client llm_client_factory providers.Factory( UserSpecificClient, # 这个类会读取上下文中的用户配置 user_contextproviders.Dependency() # 声明需要一个外部传入的上下文 ) # 在处理请求时 def handle_request(user_id: str): user_context get_user_context(user_id) # 获取用户配置 # 通过工厂方法为该请求创建一个专属的Client client container.llm_client_factory(user_contextuser_context) # ... 使用client处理请求5.2 配置化与热更新将PromptManager和ToolRegistry设计成支持外部配置如YAML、数据库。这样运营人员或产品经理可以在不重启服务的情况下修改提示词、上线新工具。PromptManager可以定期轮询配置中心或数据库加载最新的模板。ToolRegistry可以支持动态注册和注销工具。这要求你的核心业务类Agent对这些依赖的变化是免疫的——它始终通过接口访问不关心背后的实例是否被换成了新的。5.3 组合与装饰器模式增强功能依赖注入的接口是应用装饰器模式的绝佳位置。你可以在不修改核心LLMClient、Tool实现的情况下通过装饰器来增强功能。日志与审计装饰器包装LLMClient记录每一次请求和响应。缓存装饰器包装LLMClient对相同参数的请求返回缓存结果。权限校验装饰器包装Tool在执行前检查当前用户是否有权调用此工具。熔断与降级装饰器包装LLMClient在API持续失败时快速失败或切换到备用模型。class LoggingClientDecorator: def __init__(self, wrapped_client: LLMClient): self._wrapped wrapped_client def generate(self, messages): logger.info(fSending request to LLM: {messages}) start time.time() response self._wrapped.generate(messages) elapsed time.time() - start logger.info(fReceived LLM response in {elapsed:.2f}s: {response[:200]}...) return response # 在容器配置中将装饰器注入链条 container.llm_client.override( providers.Singleton(LoggingClientDecorator, container.llm_client) )5.4 应对LLM特有的挑战Prompt版本管理与工具编排Prompt版本管理当你有上百个提示词模板且需要频繁A/B测试时一个简单的InMemoryPromptManager就不够了。可以考虑引入像Weights Biases、MLflow这样的实验跟踪平台来管理Prompt版本或者自己构建一个简单的服务让PromptManager成为该服务的客户端。复杂工具编排当工具数量众多、且需要复杂编排如顺序执行、条件执行时简单的ToolRegistry可能不够。可以考虑引入工作流引擎如Prefect、Airflow或专门的Agent框架如LangGraph的概念将工具执行逻辑也抽象和配置化。此时ToolRegistry可能演进为一个WorkflowRegistry返回的不再是单个函数而是一个可执行的工作流图。6. 常见陷阱与最佳实践在实际落地这套架构时我踩过不少坑也总结出一些让项目更稳健的经验。6.1 陷阱一过度抽象与接口膨胀解耦是为了应对变化但过早或过度的抽象会引入不必要的复杂性。建议从最可能变化的地方开始抽象。通常LLMClient是第一个需要抽象的因为换模型供应商是高频需求。PromptManager次之。对于内部工具如果工具集非常稳定初期甚至可以直接在Agent构造函数中传入一个工具列表等需要动态管理时再抽象出ToolRegistry。6.2 陷阱二依赖注入容器变成“上帝类”把所有依赖的创建逻辑都塞进一个巨大的容器类会让这个容器难以理解和维护。建议按功能模块拆分容器。例如有一个LLMContainer负责所有模型相关的依赖一个ToolingContainer负责工具相关的依赖一个CoreContainer负责组装前两者并创建业务对象。这样结构更清晰也便于团队协作。6.3 陷阱三忽略依赖的生命周期不同的依赖可能有不同的生命周期配置信息可能是单例且只读的数据库连接可能需要是请求作用域的某些工具实例可能每次使用都需要新建。错误的生命周期管理会导致内存泄漏或状态污染。建议仔细规划每个Provider的作用域Singleton、Factory等。对于持有资源如网络连接、文件句柄的依赖确保容器或框架能正确地在作用域结束时清理它们。6.4 最佳实践一面向接口测试而非实现这是依赖注入带来的最大好处务必充分利用。编写业务逻辑的单元测试时只针对接口契约进行测试完全使用Mock。这能保证测试的稳定和快速。针对具体实现的测试如OpenAIClient是否真的能调用API应归入集成测试且需要有选择地运行比如只在CI/CD的特定阶段运行。6.5 最佳实践二为配置提供清晰的默认值和验证你的Container会读取大量配置API密钥、模型名、超时时间等。务必为所有配置项提供安全的默认值并使用Pydantic之类的库进行强验证。避免在运行时因为配置缺失或错误而崩溃。from pydantic import BaseSettings class LLMSettings(BaseSettings): api_key: str model: str gpt-3.5-turbo timeout: int 30 max_retries: int 3 class Config: env_prefix LLM_ # 会自动从环境变量LLM_API_KEY等读取 # 在容器中 config providers.Configuration() config.settings.from_pydantic(LLMSettings()) llm_client providers.Singleton(OpenAIClient, **config.settings)6.6 最佳实践三建立清晰的依赖关系文档随着模块增多依赖关系网会变复杂。使用依赖注入框架提供的功能如dependency-injector的wire模块或简单的图表工具绘制出核心的依赖关系图。这能帮助新成员快速理解系统结构也在排查问题时提供巨大帮助。最后我想强调的是引入依赖注入和清晰的分层架构在项目初期看起来像是“过度设计”。但一旦你的LLM应用开始处理真实的、多变的需求这种前期投入的工程规范性就会以百倍的回报体现在开发效率、测试覆盖率和系统稳定性上。它让我们的AI应用不再是脆弱的“脚本集合”而是真正可维护、可演进、可信赖的软件系统。
