LLM应用依赖注入工程实践:解耦Client、Prompt与Tool Registry

1. 项目概述:为什么我们需要解耦LLM应用?

如果你正在构建一个基于大语言模型的应用,无论是智能客服、代码助手还是数据分析工具,你可能已经感受到了那种“牵一发而动全身”的焦虑。今天改一个提示词,明天换一个模型API,后天新增一个工具函数,整个系统就像一堆积木,动一块就可能全盘散架。测试?更是无从下手,难道每次都要花真金白银去调用昂贵的API,或者冒着数据泄露的风险用真实用户对话来验证吗?

这正是“LLM 应用的依赖注入工程实践”要解决的核心痛点。这个标题听起来很学术,但它的内核非常务实:通过一套清晰的工程架构,将你的AI系统中那些最常变动、最需要独立管理的部分——模型客户端、提示词模板和工具注册表——彻底分离开。让它们从紧密耦合的“铁板一块”,变成可以像乐高积木一样自由插拔、独立测试的模块。

想象一下,你的应用核心逻辑是“大脑”,它需要“眼睛”去看(Client调用模型)、“嘴巴”去说(Prompt生成指令)、“手”去操作(Tool执行具体功能)。传统做法是把眼睛、嘴巴、手都焊死在大脑上。而依赖注入,就是为大脑设计了一套标准的接口,让它可以随时更换不同品牌、不同型号的眼睛、嘴巴和手,甚至可以在测试时,给大脑接上一个“模拟眼睛”和“假手”,完全在本地、零成本地验证大脑的逻辑是否正确。

这不仅仅是代码整洁度的问题,它直接关系到项目的生死存亡。模型迭代速度以月计,今天用的GPT-4,明天可能就要评估Claude 3;业务需求瞬息万变,针对不同用户的提示词需要A/B测试;工具链更是会不断膨胀。一个不可测试、不可替换的系统,其维护成本会指数级上升,最终沦为技术债的泥潭。接下来,我们就深入拆解,如何一步步构建这样一个既健壮又灵活的系统。

2. 核心架构设计:Client、Prompt与Tool Registry的三权分立

要解耦,首先得明确什么该被解耦。在LLM应用这个上下文里,经过大量实践,我总结出三个最不稳定、最需要被独立管理的核心依赖,也就是标题中的三位主角。

2.1 模型客户端:抽象的“对话能力”

Client,在这里特指与大语言模型服务进行通信的客户端。它不应该是一个具体的OpenAIAnthropic的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]: """流式生成文本""" pass

2.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 依赖注入:将它们编织在一起的粘合剂

现在,我们有了三个独立的组件:LLMClientPromptManagerToolRegistry。我们的核心业务类,比如一个CustomerServiceAgent,需要它们。依赖注入框架(如Spring之于Java,或dependency-injectorinjector之于Python)的作用,就是充当一个智能的“装配工”。

你不再在CustomerServiceAgent的构造函数里new一个具体的Client,而是声明:“我需要一个LLMClientPromptManager”。在应用启动时,依赖注入容器会根据配置,将配置好的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更灵活。PydanticBaseModel用于工具的数据验证和序列化。

3.2 第二步:实现具体的依赖组件

接着,我们实现这些接口的具体版本。这里以OpenAI和内存存储为例。

# 具体Client实现 import openai class OpenAIClient: def __init__(self, api_key: str, model: str = "gpt-4"): self.client = openai.OpenAI(api_key=api_key) self.model = model def generate(self, messages): response = self.client.chat.completions.create( model=self.model, messages=messages ) 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(f"Prompt 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_goal=user_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", request=user_request, tool_list=self.tools.get_descriptions() ) tool_decision = await self.llm.generate_async([ {"role": "user", "content": tool_choice_prompt} ]) # ... 解析LLM返回,调用ToolRegistry中的工具执行 return f"计划:{plan}, 执行决策:{tool_decision}"

关键点TaskAgent的代码里没有任何一行涉及openaiyaml(读提示词文件)或者具体的工具函数。它只和三个抽象接口对话。这就是“依赖倒置”原则的体现:高层模块(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_key=config.openai.api_key, model=config.openai.model ) prompt_manager = providers.Singleton(InMemoryPromptManager) tool_registry = providers.Singleton(SimpleToolRegistry) # 定义业务对象,并声明其依赖 task_agent = providers.Factory( TaskAgent, llm_client=llm_client, prompt_manager=prompt_manager, tool_registry=tool_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(name="get_time", description="获取当前时间", func=lambda: datetime.now().isoformat(), parameters_schema={})) registry.register(Tool(name="search_web", description="网络搜索", func=search_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(spec=LLMClient) mock_prompts = Mock(spec=PromptManager) mock_registry = Mock(spec=ToolRegistry) # 2. 预设Mock行为 mock_prompts.get_template.side_effect = lambda name, **vars: f"Mocked 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_client=mock_client, prompt_manager=mock_prompts, tool_registry=mock_registry) # 4. 执行测试 result = await agent.execute("test request") # 5. 验证交互逻辑 # 断言PromptManager被以正确的参数调用 mock_prompts.get_template.assert_any_call("task_decompose", user_goal="test request") mock_prompts.get_template.assert_any_call("tool_choice", request="test 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 集成测试:验证组件间的真实协作

集成测试用于验证我们的具体实现(如OpenAIClientInMemoryPromptManager)是否能正确地协同工作。这里我们仍然要避免调用真实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_value=httpx.Response(200, json={ "choices": [{"message": {"content": "Mocked API response"}}] }) ) # 2. 使用真实的OpenAIClient类,但它发出的请求会被拦截 client = OpenAIClient(api_key="fake_key", model="gpt-3.5-turbo") # 3. 执行调用 response = await client.generate_async([{"role": "user", "content": "Hello"}]) # 4. 验证 assert response == "Mocked API response" # 可以进一步验证发出的请求体格式是否正确

对于ToolRegistryPromptManager的集成测试,则可以直接使用它们的内存实现,验证工具注册、查找和提示词渲染功能是否正常。

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框架(如fastapiDepends)原生支持这种模式。

from dependency_injector import providers class UserAwareContainer(containers.DeclarativeContainer): # 定义一个工厂,它依赖一个“用户上下文”来创建Client llm_client_factory = providers.Factory( UserSpecificClient, # 这个类会读取上下文中的用户配置 user_context=providers.Dependency() # 声明需要一个外部传入的上下文 ) # 在处理请求时 def handle_request(user_id: str): user_context = get_user_context(user_id) # 获取用户配置 # 通过工厂方法,为该请求创建一个专属的Client client = container.llm_client_factory(user_context=user_context) # ... 使用client处理请求

5.2 配置化与热更新

PromptManagerToolRegistry设计成支持外部配置(如YAML、数据库)。这样,运营人员或产品经理可以在不重启服务的情况下,修改提示词、上线新工具。

PromptManager可以定期轮询配置中心或数据库,加载最新的模板。ToolRegistry可以支持动态注册和注销工具。这要求你的核心业务类Agent对这些依赖的变化是免疫的——它始终通过接口访问,不关心背后的实例是否被换成了新的。

5.3 组合与装饰器模式增强功能

依赖注入的接口,是应用装饰器模式的绝佳位置。你可以在不修改核心LLMClientTool实现的情况下,通过装饰器来增强功能。

  • 日志与审计装饰器:包装LLMClient,记录每一次请求和响应。
  • 缓存装饰器:包装LLMClient,对相同参数的请求返回缓存结果。
  • 权限校验装饰器:包装Tool,在执行前检查当前用户是否有权调用此工具。
  • 熔断与降级装饰器:包装LLMClient,在API持续失败时快速失败或切换到备用模型。
class LoggingClientDecorator: def __init__(self, wrapped_client: LLMClient): self._wrapped = wrapped_client def generate(self, messages): logger.info(f"Sending request to LLM: {messages}") start = time.time() response = self._wrapped.generate(messages) elapsed = time.time() - start logger.info(f"Received 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 & BiasesMLflow这样的实验跟踪平台来管理Prompt版本,或者自己构建一个简单的服务,让PromptManager成为该服务的客户端。

复杂工具编排:当工具数量众多、且需要复杂编排(如顺序执行、条件执行)时,简单的ToolRegistry可能不够。可以考虑引入工作流引擎(如PrefectAirflow)或专门的Agent框架(如LangGraph)的概念,将工具执行逻辑也抽象和配置化。此时,ToolRegistry可能演进为一个WorkflowRegistry,返回的不再是单个函数,而是一个可执行的工作流图。

6. 常见陷阱与最佳实践

在实际落地这套架构时,我踩过不少坑,也总结出一些让项目更稳健的经验。

6.1 陷阱一:过度抽象与接口膨胀

解耦是为了应对变化,但过早或过度的抽象会引入不必要的复杂性。建议:从最可能变化的地方开始抽象。通常,LLMClient是第一个需要抽象的,因为换模型供应商是高频需求。PromptManager次之。对于内部工具,如果工具集非常稳定,初期甚至可以直接在Agent构造函数中传入一个工具列表,等需要动态管理时再抽象出ToolRegistry

6.2 陷阱二:依赖注入容器变成“上帝类”

把所有依赖的创建逻辑都塞进一个巨大的容器类,会让这个容器难以理解和维护。建议:按功能模块拆分容器。例如,有一个LLMContainer负责所有模型相关的依赖,一个ToolingContainer负责工具相关的依赖,一个CoreContainer负责组装前两者并创建业务对象。这样结构更清晰,也便于团队协作。

6.3 陷阱三:忽略依赖的生命周期

不同的依赖可能有不同的生命周期:配置信息可能是单例且只读的;数据库连接可能需要是请求作用域的;某些工具实例可能每次使用都需要新建。错误的生命周期管理会导致内存泄漏或状态污染。建议:仔细规划每个Provider的作用域(SingletonFactory等)。对于持有资源(如网络连接、文件句柄)的依赖,确保容器或框架能正确地在作用域结束时清理它们。

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-injectorwire模块)或简单的图表工具,绘制出核心的依赖关系图。这能帮助新成员快速理解系统结构,也在排查问题时提供巨大帮助。

最后,我想强调的是,引入依赖注入和清晰的分层架构,在项目初期看起来像是“过度设计”。但一旦你的LLM应用开始处理真实的、多变的需求,这种前期投入的工程规范性就会以百倍的回报体现在开发效率、测试覆盖率和系统稳定性上。它让我们的AI应用不再是脆弱的“脚本集合”,而是真正可维护、可演进、可信赖的软件系统。