1. 项目概述为什么我们需要解耦LLM应用最近在折腾几个基于大语言模型LLM的智能应用项目从简单的客服机器人到复杂的自动化工作流踩的坑一个接一个。最让我头疼的不是模型效果调优而是整个系统的“脆弱性”。今天改个提示词Prompt明天换个模型接口Client后天加个新工具Tool牵一发而动全身测试起来更是噩梦。代码里到处都是硬编码的OpenAI()实例、散落各处的提示词模板字符串、以及一个臃肿不堪、全局共享的工具注册表。想给某个功能写个单元测试Mock起来复杂到想放弃。想从GPT-4换成Claude或者本地模型几乎要重写一半的业务逻辑。这让我意识到我们很多LLM应用还处在“脚本”阶段远未达到“工程系统”的标准。其核心问题在于紧耦合。Client、Prompt、Tool Registry这三者像一团乱麻纠缠在一起导致系统难以测试、难以维护、更难以扩展。于是我开始尝试将经典的软件工程思想——依赖注入Dependency Injection, DI——引入LLM应用开发。实践下来发现这简直是给混乱的AI系统开发套上了一个清晰的框架。它不仅能让你轻松替换模型、切换提示策略、管理工具集更重要的是它让整个系统变得真正可测试。今天我就来详细拆解一下这套“LLM应用的依赖注入工程实践”聊聊如何解耦Client、Prompt和Tool Registry让你的AI系统从“玩具”进化到“工业级”。2. 核心痛点紧耦合的LLM应用长什么样在深入解决方案前我们得先看清问题。一个典型的、未经过设计的LLM应用代码可能长这样# 糟糕的紧耦合示例 import openai from some_tool_module import calculator, search_web class ChatAgent: def __init__(self): # 痛点1: Client硬编码 self.client openai.OpenAI(api_keysk-...) # API Key也可能硬编码或来自全局配置 # 痛点2: Prompt硬编码且分散 self.system_prompt 你是一个有帮助的助手。 # 痛点3: Tool Registry是全局的或静态的 self.tools [calculator, search_web] def chat(self, user_input): # 痛点4: 业务逻辑与具体实现深度绑定 messages [ {role: system, content: self.system_prompt}, {role: user, content: user_input} ] # 直接调用特定Client的方法 response self.client.chat.completions.create( modelgpt-4, messagesmessages, toolsself.tools # 工具直接传入 ) # 处理工具调用... return response.choices[0].message.content这段代码在项目初期跑起来很快但一旦需要变化问题就全暴露了难以测试你怎么对chat方法做单元测试你需要一个真实的OpenAI API Key并且会产生实际费用和网络调用。Mockopenai.OpenAI这个类及其方法非常繁琐。难以替换组件想把模型从GPT-4换成Azure OpenAI或Anthropic Claude你需要找到所有调用self.client.chat.completions.create的地方修改参数甚至调用方式。提示词想根据用户上下文动态生成你需要侵入业务逻辑修改messages的组装过程。配置管理混乱API Key、模型名称、基础URL等配置散落在代码各处或者依赖一个全局的单例配置对象不利于环境隔离开发、测试、生产。工具管理僵化所有Agent共享同一套工具无法做到细粒度的权限控制或场景化工具集。想为某个特定任务启用一个特殊工具可能需要修改全局工具列表影响其他功能。依赖注入要解决的正是这种“创建依赖”的责任与“使用依赖”的逻辑混杂在一起的问题。它的核心思想是一个类不应该自己创建它所需要的对象依赖而应该由外部容器或调用者来提供。这样类就只关心自己的核心业务逻辑而不关心依赖的具体实现和生命周期。3. 架构设计定义清晰的抽象与接口解耦的第一步是定义边界。我们需要为Client、Prompt和Tool Registry这三个核心概念建立抽象接口在Python中通常使用抽象基类abc.ABC。3.1 定义LLM Client抽象Client的职责是封装与LLM服务的一切通信细节。一个良好的Client接口应该只暴露业务所需的方法隐藏底层SDK的差异。from abc import ABC, abstractmethod from typing import List, Dict, Any, Optional class LLMClient(ABC): LLM客户端的抽象接口。 abstractmethod async def chat_completion( self, messages: List[Dict[str, str]], model: Optional[str] None, temperature: float 0.7, **kwargs ) - str: 发起聊天补全请求返回纯文本内容。 pass abstractmethod async def chat_completion_with_tools( self, messages: List[Dict[str, str]], tools: List[Dict[str, Any]], # 工具定义列表 tool_choice: Optional[str] None, model: Optional[str] None, **kwargs ) - Dict[str, Any]: 发起支持工具调用的聊天补全请求。 返回一个字典包含可能的文本回复和工具调用请求。 例如{content: ..., tool_calls: [...]} pass abstractmethod def get_model_list(self) - List[str]: 获取该客户端支持的模型列表。 pass为什么这样设计异步优先LLM调用通常是I/O密集型操作使用async/await能更好地利用资源提升系统吞吐量。返回标准结构chat_completion_with_tools返回字典而不是某个SDK特有的对象如OpenAI的ChatCompletion对象。这隔离了底层SDK的变化上游业务逻辑只处理标准化的数据结构。参数通用化只包含最通用的参数messages,model,temperature其他供应商特定参数通过**kwargs传递由具体实现类处理。3.2 定义Prompt Provider抽象Prompt的职责是根据当前对话上下文、用户信息或其他状态生成最终发送给LLM的消息列表messages。它不应该只是一个静态字符串。from abc import ABC, abstractmethod from typing import List, Dict, Any class PromptProvider(ABC): 提示词提供者的抽象接口。 abstractmethod async def get_messages( self, user_input: str, conversation_history: List[Dict[str, str]], context: Dict[str, Any] None, **kwargs ) - List[Dict[str, str]]: 根据输入、历史记录和上下文生成LLM消息列表。 Args: user_input: 用户当前输入。 conversation_history: 之前的对话消息列表格式同OpenAI API。 context: 额外的上下文信息如用户ID、会话ID、业务数据等。 Returns: 符合LLM API要求的消息列表例如 [{role: system, content: ...}, {role: user, content: ...}] pass为什么这样设计动态化Prompt可以基于丰富的上下文动态生成实现多角色扮演、个性化回复、上下文注入等高级功能。可测试你可以轻松创建一个返回固定消息列表的Mock PromptProvider用于测试Agent的核心逻辑而无需关心提示词的具体内容。职责分离Agent不再需要知道如何组装system_prompt、user_prompt和history这部分逻辑被封装在独立的、可替换的组件中。3.3 定义Tool Registry抽象Tool Registry的职责是管理一组可被LLM调用的工具函数并提供工具描述信息的获取、工具的执行和结果处理。from abc import ABC, abstractmethod from typing import List, Dict, Any, Callable class ToolRegistry(ABC): 工具注册表的抽象接口。 abstractmethod def register_tool(self, tool_func: Callable, tool_schema: Dict[str, Any]): 向注册表中注册一个工具。 Args: tool_func: 可调用的工具函数。 tool_schema: 符合OpenAI Function Calling格式的工具模式定义。 pass abstractmethod def get_tools_schema(self) - List[Dict[str, Any]]: 获取所有已注册工具的Schema列表用于传递给LLM Client。 pass abstractmethod async def execute_tool( self, tool_name: str, tool_arguments: Dict[str, Any] ) - Any: 根据工具名和参数执行对应的工具并返回结果。 pass abstractmethod def get_tool(self, tool_name: str) - Optional[Callable]: 根据工具名获取工具函数。 pass为什么这样设计动态注册与发现工具可以在运行时动态添加或移除支持插件化架构。Schema与执行分离get_tools_schema方法专门为LLM提供描述信息而execute_tool处理实际调用。这允许你在提供Schema时进行过滤或转换例如基于用户权限隐藏某些工具。统一执行入口所有工具调用都通过execute_tool这一个入口便于添加统一的日志、监控、权限校验或错误处理逻辑。4. 实现与集成构建可注入的组件有了清晰的接口接下来就是实现它们并组装成一个松耦合的Agent。4.1 实现具体的组件OpenAI Client实现import openai from typing import List, Dict, Any, Optional from .llm_client import LLMClient class OpenAIClient(LLMClient): def __init__(self, api_key: str, base_url: Optional[str] None, default_model: str gpt-4): # 依赖通过构造器注入 self.client openai.AsyncOpenAI(api_keyapi_key, base_urlbase_url) self.default_model default_model async def chat_completion(self, messages: List[Dict[str, str]], model: Optional[str] None, temperature: float 0.7, **kwargs) - str: model_to_use model or self.default_model try: response await self.client.chat.completions.create( modelmodel_to_use, messagesmessages, temperaturetemperature, **kwargs ) return response.choices[0].message.content except openai.APIError as e: # 统一的错误处理与转换 raise LLMClientError(fOpenAI API调用失败: {e}) from e async def chat_completion_with_tools(self, messages: List[Dict[str, str]], tools: List[Dict[str, Any]], tool_choice: Optional[str] None, model: Optional[str] None, **kwargs) - Dict[str, Any]: model_to_use model or self.default_model response await self.client.chat.completions.create( modelmodel_to_use, messagesmessages, toolstools, tool_choicetool_choice, **kwargs ) message response.choices[0].message result {content: message.content} if message.tool_calls: result[tool_calls] [{ id: tc.id, type: tc.type, function: { name: tc.function.name, arguments: tc.function.arguments } } for tc in message.tool_calls] return result def get_model_list(self) - List[str]: # 这里可以调用模型列表API或返回一个预设列表 return [gpt-4, gpt-4-turbo, gpt-3.5-turbo]动态PromptProvider实现from jinja2 import Template from .prompt_provider import PromptProvider class DynamicPromptProvider(PromptProvider): def __init__(self, system_template: str): # 可以使用Jinja2等模板引擎 self.system_template Template(system_template) async def get_messages(self, user_input: str, conversation_history: List[Dict[str, str]], context: Dict[str, Any] None, **kwargs) - List[Dict[str, str]]: context context or {} # 动态渲染系统提示词 system_content self.system_template.render(**context) messages [{role: system, content: system_content}] # 添加上下文历史 messages.extend(conversation_history[-10:]) # 限制历史长度 # 添加当前用户输入 messages.append({role: user, content: user_input}) return messages内存Tool Registry实现from typing import Dict, Any, Callable, Optional from .tool_registry import ToolRegistry class InMemoryToolRegistry(ToolRegistry): def __init__(self): self._tools: Dict[str, Dict[str, Any]] {} # name - {func: ..., schema: ...} def register_tool(self, tool_func: Callable, tool_schema: Dict[str, Any]): tool_name tool_schema[function][name] self._tools[tool_name] {func: tool_func, schema: tool_schema} def get_tools_schema(self) - List[Dict[str, Any]]: return [tool_info[schema] for tool_info in self._tools.values()] async def execute_tool(self, tool_name: str, tool_arguments: Dict[str, Any]) - Any: if tool_name not in self._tools: raise ValueError(f工具 {tool_name} 未注册。) tool_func self._tools[tool_name][func] # 注意实际执行可能需要处理异步函数 import inspect if inspect.iscoroutinefunction(tool_func): return await tool_func(**tool_arguments) else: return tool_func(**tool_arguments) def get_tool(self, tool_name: str) - Optional[Callable]: tool_info self._tools.get(tool_name) return tool_info[func] if tool_info else None4.2 组装依赖注入的Agent现在我们可以构建一个高度可配置、可测试的Agent核心类。from typing import Optional, List, Dict, Any class DIAgent: 通过依赖注入构建的智能体。 def __init__( self, llm_client: LLMClient, prompt_provider: PromptProvider, tool_registry: ToolRegistry, max_tool_iterations: int 5 # 防止无限工具调用循环 ): # 所有依赖通过构造器注入 self.llm_client llm_client self.prompt_provider prompt_provider self.tool_registry tool_registry self.max_tool_iterations max_tool_iterations self.conversation_history: List[Dict[str, str]] [] async def chat(self, user_input: str, context: Dict[str, Any] None) - str: 主要的聊天交互方法。 context context or {} full_response_text for iteration in range(self.max_tool_iterations): # 1. 通过PromptProvider获取消息 messages await self.prompt_provider.get_messages( user_inputuser_input if iteration 0 else , # 首次迭代传入用户输入 conversation_historyself.conversation_history, contextcontext ) # 2. 获取可用工具Schema tools_schema self.tool_registry.get_tools_schema() # 3. 调用LLM Client llm_response await self.llm_client.chat_completion_with_tools( messagesmessages, toolstools_schema if tools_schema else None, tool_choiceauto if tools_schema else none ) # 4. 处理LLM响应 content llm_response.get(content) tool_calls llm_response.get(tool_calls, []) if content: full_response_text content \n # 将LLM的回复添加到历史中 self.conversation_history.append({role: assistant, content: content}) # 5. 如果没有工具调用结束循环 if not tool_calls: break # 6. 执行工具调用 tool_messages [] for tc in tool_calls: tool_name tc[function][name] tool_args tc[function][arguments] # 注意实际需要解析JSON字符串 import json try: arguments_dict json.loads(tool_args) except json.JSONDecodeError: arguments_dict {} try: tool_result await self.tool_registry.execute_tool(tool_name, arguments_dict) # 将工具执行结果作为消息反馈给LLM tool_messages.append({ role: tool, tool_call_id: tc[id], content: str(tool_result) }) except Exception as e: tool_messages.append({ role: tool, tool_call_id: tc[id], content: f工具执行错误: {e} }) # 将工具执行结果添加到历史作为下一轮迭代的输入 self.conversation_history.extend(tool_messages) # 下一轮迭代user_input为空LLM会基于工具结果继续思考 user_input # 更新历史添加最终的用户输入和完整的助手回复可选 self.conversation_history.append({role: user, content: user_input}) # 注意实际中可能需要更精细的历史管理避免过长 return full_response_text.strip()这个设计的精妙之处在于构造器注入DIAgent的所有核心依赖Client, Prompt, Tools都通过__init__方法传入。它不关心这些对象是怎么来的从配置文件读取、从工厂创建、还是测试时Mock的它只负责使用它们。面向接口编程DIAgent只依赖于三个抽象接口而不是具体实现。这意味着你可以随时替换它们。单一职责每个类都有明确且单一的职责代码可读性和可维护性大大提升。5. 依赖注入容器的使用在大型应用中手动创建和传递所有依赖会很繁琐。这时可以引入一个轻量的依赖注入容器如dependency-injector库或自己实现一个简单的Container来管理组件的创建和生命周期。# 一个简单的配置容器示例 class AppContainer: def __init__(self, config: Dict[str, Any]): self.config config self._llm_client None self._prompt_provider None self._tool_registry None property def llm_client(self) - LLMClient: if self._llm_client is None: provider self.config[llm][provider] if provider openai: from .clients import OpenAIClient self._llm_client OpenAIClient( api_keyself.config[llm][api_key], base_urlself.config[llm].get(base_url), default_modelself.config[llm].get(model, gpt-4) ) elif provider anthropic: from .clients import AnthropicClient self._llm_client AnthropicClient(...) # ... 其他实现 return self._llm_client property def prompt_provider(self) - PromptProvider: if self._prompt_provider is None: from .prompts import DynamicPromptProvider system_prompt self.config[prompts][system] self._prompt_provider DynamicPromptProvider(system_templatesystem_prompt) return self._prompt_provider property def tool_registry(self) - ToolRegistry: if self._tool_registry is None: from .tools import InMemoryToolRegistry self._tool_registry InMemoryToolRegistry() # 自动注册配置中声明的工具 for tool_config in self.config.get(tools, []): self._register_tool_from_config(tool_config) return self._tool_registry def create_agent(self) - DIAgent: 工厂方法创建注入所有依赖的Agent实例。 return DIAgent( llm_clientself.llm_client, prompt_providerself.prompt_provider, tool_registryself.tool_registry ) def _register_tool_from_config(self, tool_config: dict): # ... 根据配置动态导入和注册工具 pass # 应用入口 config { llm: {provider: openai, api_key: sk-..., model: gpt-4}, prompts: {system: 你是一个专业的助手。}, tools: [...] } container AppContainer(config) agent container.create_agent() # 现在可以愉快地使用agent了 response await agent.chat(今天的天气怎么样)容器将对象的创建逻辑集中管理应用的其他部分只需要从容器中获取完全组装好的、可用的对象。这使得配置变更比如切换LLM提供商只需要修改一处配置而无需改动业务代码。6. 测试实践Mock让一切变得简单依赖注入最大的优势之一就是极大地简化了测试。我们可以轻松地为每个抽象接口创建Mock模拟对象。import pytest from unittest.mock import AsyncMock, Mock from .di_agent import DIAgent from .llm_client import LLMClient from .prompt_provider import PromptProvider from .tool_registry import ToolRegistry pytest.fixture def mock_llm_client(): client AsyncMock(specLLMClient) # 模拟一个只返回文本回复的LLM client.chat_completion_with_tools.return_value { content: 这是一个模拟回复。, tool_calls: [] } return client pytest.fixture def mock_prompt_provider(): provider AsyncMock(specPromptProvider) provider.get_messages.return_value [ {role: system, content: 测试系统提示}, {role: user, content: 用户输入} ] return provider pytest.fixture def mock_tool_registry(): registry Mock(specToolRegistry) registry.get_tools_schema.return_value [] # 模拟没有工具 return registry pytest.mark.asyncio async def test_agent_chat_without_tools(mock_llm_client, mock_prompt_provider, mock_tool_registry): 测试Agent在没有工具调用情况下的正常聊天流程。 # 1. 组装被测试对象注入所有Mock依赖 agent DIAgent( llm_clientmock_llm_client, prompt_providermock_prompt_provider, tool_registrymock_tool_registry ) # 2. 执行测试 response await agent.chat(你好) # 3. 验证行为 # 确保PromptProvider被正确调用 mock_prompt_provider.get_messages.assert_called_once() # 确保LLM Client被正确调用并且传入了正确的参数如消息和空工具列表 mock_llm_client.chat_completion_with_tools.assert_called_once() call_args mock_llm_client.chat_completion_with_tools.call_args assert call_args[1][tools] is None or call_args[1][tools] [] # 确保Tool Registry的get_tools_schema被调用 mock_tool_registry.get_tools_schema.assert_called_once() # 验证返回结果 assert response 这是一个模拟回复。 # 4. 验证工具执行方法没有被调用因为没有工具调用 assert not mock_tool_registry.execute_tool.called pytest.mark.asyncio async def test_agent_chat_with_tool_call(): 测试Agent处理工具调用的流程。 # 创建更复杂的Mock来模拟工具调用场景 mock_client AsyncMock(specLLMClient) mock_provider AsyncMock(specPromptProvider) mock_registry Mock(specToolRegistry) # 模拟第一轮LLM回复要求调用工具 mock_client.chat_completion_with_tools.side_effect [ { # 第一轮响应 content: 我来帮你查一下。, tool_calls: [{ id: call_123, type: function, function: {name: get_weather, arguments: {city: 北京}} }] }, { # 第二轮响应收到工具结果后 content: 北京今天晴天25度。, tool_calls: [] } ] # 模拟工具执行返回结果 mock_registry.execute_tool.return_value 北京晴天25摄氏度微风。 mock_registry.get_tools_schema.return_value [{type: function, function: {name: get_weather, description: ...}}] agent DIAgent(mock_client, mock_provider, mock_registry, max_tool_iterations5) response await agent.chat(北京天气怎么样) # 验证LLM被调用了两次 assert mock_client.chat_completion_with_tools.call_count 2 # 验证工具被执行了一次 mock_registry.execute_tool.assert_called_once_with(get_weather, {city: 北京}) # 验证最终回复包含了工具执行后的结果 assert 25度 in response通过Mock我们可以在完全隔离的环境中测试DIAgent的核心逻辑无需真实的API调用、网络连接或复杂的工具环境。测试运行速度极快且稳定可靠。你可以轻松覆盖各种边界情况比如网络超时、工具执行异常、LLM返回格式错误等。7. 进阶技巧与避坑指南在实际项目中应用这套模式我总结了一些关键的经验和容易踩的坑。7.1 配置管理集中化配置将所有依赖的配置API密钥、模型名称、Prompt模板、工具列表放在一个统一的配置管理系统中如config.yaml、环境变量、配置中心。容器根据配置动态组装对象。环境隔离利用配置系统为开发、测试、生产环境设置不同的参数例如测试环境使用Mock Client开发环境使用低配模型。安全存储API密钥等敏感信息务必使用环境变量或密钥管理服务绝不能硬编码在代码或配置文件中。7.2 生命周期管理单例与多例思考每个组件的生命周期。LLMClient通常可以是一个单例共享连接池。PromptProvider可能根据会话或用户不同而不同多例。ToolRegistry可能是单例但也可能需要支持动态更新。资源清理如果Client持有网络连接等资源确保容器或应用在关闭时能正确清理实现__aexit__或类似的清理钩子。7.3 性能与缓存Prompt缓存复杂的Prompt生成可能涉及数据库查询或网络请求。考虑对渲染后的Prompt进行缓存特别是系统提示词部分。工具Schema缓存get_tools_schema()的返回结果在工具未变更时是静态的可以缓存起来避免每次LLM调用都重新序列化。连接池对于HTTP Client确保使用连接池如aiohttp.ClientSession以提升性能。7.4 错误处理与韧性客户端抽象层统一错误在LLMClient实现中捕获所有底层SDK的异常并转换为自定义的、统一的异常类型如LLMClientError,RateLimitError,ContextLengthExceededError。这使上层业务逻辑的错误处理更清晰。重试与退避在网络调用和API调用中实现指数退避的重试机制。可以将这部分逻辑封装在Client内部也可以使用装饰器模式附加到Client上。降级策略当主要LLM服务不可用时是否有备选方案依赖注入使得实现一个“降级Client”如返回静态回复或切换到更稳定的模型变得非常容易。7.5 常见陷阱过度设计对于非常小的、一次性的脚本引入完整的DI框架可能杀鸡用牛刀。评估项目复杂度再决定。接口污染不要为了DI而DI。接口应该基于业务需求定义而不是为了解耦而生造出许多不必要的小接口。保持接口的简洁和稳定。循环依赖如果组件A依赖BB又依赖A会导致容器无法解析。设计时要避免循环依赖或者引入“懒加载”或“setter注入”来打破循环。Mock过于复杂如果Mock一个依赖需要写大量代码可能是这个依赖的接口设计得太复杂、职责太多。考虑重构接口使其更单一、更易于模拟。8. 总结与展望将依赖注入引入LLM应用开发本质上是在用成熟的软件工程方法论来管理AI系统固有的复杂性。通过解耦Client、Prompt和Tool Registry我们获得了一个高度模块化、可测试、可替换、可配置的系统。可维护性每个模块职责清晰修改一个部分不会波及其他。可测试性单元测试变得简单高效这是保障长期代码质量的基础。可扩展性想要支持新的LLM提供商实现一个新的LLMClient子类即可。想要动态Prompt换一个PromptProvider实现。工具需要权限管理装饰你的ToolRegistry或在execute_tool方法中添加逻辑。团队协作清晰的接口定义了团队之间的契约前端、后端、算法工程师可以并行开发。这套模式不仅适用于聊天Agent也可以扩展到更复杂的AI工作流、RAG系统、甚至是多智能体协作场景。每个智能体、每个检索器、每个评估器都可以被抽象和注入。当然没有银弹。依赖注入会引入一些前期设计的复杂性和少量的运行时开销。但对于任何计划长期迭代、需要稳定性和可维护性的LLM应用项目来说这笔投资绝对是值得的。它让我们的AI系统从“实验室原型”真正走向了“生产级工程”。