1. 从“胶水代码”到“工程系统”为什么你的LLM应用难以测试和迭代最近在重构一个内部使用的AI客服系统时我遇到了一个典型问题想从GPT-4切换到Claude 3或者想把提示词从英文改成更地道的中文结果发现改动一处处处报错。Client初始化散落在十几个业务类里提示词模板和工具调用逻辑硬编码在核心流程中想写个单元测试比重新开发还难。这让我意识到很多LLM应用在快速原型阶段后就陷入了“胶水代码”的泥潭——功能能跑但架构是一团乱麻。这不仅仅是换模型的问题。当你需要A/B测试不同提示词对最终回答质量的影响。动态切换LLM供应商以应对服务降级或成本优化。对工具调用Tool Calling逻辑进行单元测试而不需要每次调用真实的LLM烧钱且慢。团队协作时清晰地区分“基础设施层”怎么调用模型、“策略层”用什么提示词和工具和“业务层”处理什么业务逻辑。你会发现如果Client、Prompt和Tool Registry工具注册表这三者紧紧耦合在一起上述任何需求都意味着伤筋动骨的重构。标题里提到的“依赖注入”Dependency Injection, DI正是解决这类耦合问题的经典软件工程范式。它不是什么新潮概念但在LLM应用开发中其价值被严重低估了。很多人觉得“我的脚本能跑通就行”直到需要维护和扩展时才追悔莫及。本文的目标就是分享如何将DI思想落地到LLM应用中通过解耦上述三个核心依赖构建出真正可测试、可替换、易维护的AI系统。我们会从问题出发一步步拆解设计并用具体的代码示例以Python为主展示如何实现。无论你用的是LangChain、LlamaIndex这类框架还是直接调用SDK这套工程实践都能让你的项目底座更稳固。2. 核心痛点剖析紧耦合是如何“绑架”你的LLM应用的在深入解决方案之前我们先具体化一下“紧耦合”带来的痛苦。假设我们有一个简单的查询天气的Agent最初的、常见的“快糙猛”实现可能是这样的# 紧耦合的典型示例 - 难以维护和测试的代码 from openai import OpenAI import requests class WeatherAgent: def __init__(self, api_key): # 依赖1: Client 被硬编码在初始化中 self.client OpenAI(api_keyapi_key) # 依赖2: Prompt 模板是类内部的字符串 self.prompt_template 你是一个天气助手。用户问{user_query} 请调用获取天气的工具来回答用户。 # 依赖3: Tool 的实现直接写在类方法里 self.tools [ { type: function, function: { name: get_current_weather, description: 获取指定城市的当前天气, parameters: {...} } } ] def get_weather_tool(self, location): # 硬编码的工具实现可能调用某个特定天气API response requests.get(fhttps://api.weatherapi.com/v1/current.json?keyYOUR_KEYq{location}) return response.json() def run(self, user_query): # 拼接Prompt prompt self.prompt_template.format(user_queryuser_query) # 调用LLM response self.client.chat.completions.create( modelgpt-4, messages[{role: user, content: prompt}], toolsself.tools, tool_choiceauto ) # 解析并执行工具调用 # ... 复杂的解析逻辑 tool_call response.choices[0].message.tool_calls[0] if tool_call.function.name get_current_weather: args json.loads(tool_call.function.arguments) result self.get_weather_tool(args[location]) # 再次调用LLM总结结果... return final_answer这段代码跑起来没问题但它隐藏了至少四个致命缺陷缺陷一Client 不可替换。整个WeatherAgent类的生命周期与OpenAI客户端绑定。如果你想换成anthropic.Client或者本地部署的llama.cpp你需要修改__init__和run方法中所有调用self.client的地方。这违反了“开闭原则”对扩展开放对修改关闭。缺陷二Prompt 难以管理和优化。提示词以字符串形式散落在代码中。当你有几十个不同的提示词模板需要统一调整格式、添加系统指令或进行版本化管理时你会陷入“字符串地狱”。更别提进行A/B测试了你需要复制整个类或者写一堆if-else。缺陷三Tool 的逻辑与 Agent 核心流程耦合。get_weather_tool方法直接写在WeatherAgent里。这意味着你无法单独测试这个工具函数比如模拟不同API返回。其他Agent想复用这个工具只能复制代码。工具的实现细节如请求的URL、认证方式变更会直接影响Agent类。缺陷四几乎无法进行单元测试。如何测试run方法你需要准备一个真实的OpenAI API Key产生费用。实际调用外部天气API不稳定且慢。测试用例变得缓慢、脆弱且不可重复。这些缺陷在项目初期可能不明显但随着复杂度提升它们会像债务一样累积最终导致项目难以迭代团队协作效率低下。解决之道就是引入“依赖注入”的思想将创建依赖Client, Prompt, Tools的责任从使用它们的类如WeatherAgent中剥离出去。3. 依赖注入DI在LLM上下文中的核心思想依赖注入不是什么银弹它只是一种设计模式核心思想是“控制反转”IoC。简单说一个类不应该自己创建它所需要的依赖对象而应该由外部“注入”给它。在LLM应用场景下我们可以将核心依赖抽象为三类LLM Client负责与底层大模型交互的客户端。例如OpenAI、Anthropic、OllamaClient或是封装了Azure OpenAI的客户端。Prompt Provider/Template负责提供和管理提示词模板的组件。它可能从文件、数据库或配置中心加载模板并负责变量的渲染。Tool Registry/Executor负责注册、管理和执行“工具”函数的组件。它知道有哪些工具可用并能根据名称调用对应的函数。一个设计良好的LLM Agent或Chain应该像下面这样声明它的依赖class WellDesignedAgent: def __init__(self, llm_client, prompt_provider, tool_registry): self.llm_client llm_client # 依赖被注入 self.prompt_provider prompt_provider self.tool_registry tool_registry def run(self, input): # 使用注入的依赖而不关心它们如何被创建 prompt self.prompt_provider.get_prompt(weather_assistant, queryinput) # ... 其余逻辑这样的好处立竿见影可替换性想要换模型只需在创建WellDesignedAgent的地方传入另一个llm_client实例即可Agent内部代码一行都不用改。可测试性在单元测试中你可以传入模拟对象Mock。给llm_client传入一个MockClient让它直接返回你预设的响应避免真实API调用。给tool_registry传入一个MockRegistry验证Agent是否用正确的参数调用了正确的工具。关注点分离Agent只关心业务流程先获取提示词再调用LLM再解析工具调用...而Client、Prompt、Tools的创建、配置和管理则由更上层的模块通常是应用的主入口或一个专门的工厂类负责。理解了核心思想后我们接下来看看如何具体地解耦每一个依赖。4. 实践一解耦 LLM Client —— 定义抽象接口与多实现第一步是为LLM Client定义一个抽象的接口在Python中通常使用Protocol或ABC抽象基类。这个接口只声明我们关心的方法例如chat_completion。from typing import Protocol, List, Dict, Any, Optional import json class LLMClientProtocol(Protocol): LLM客户端的抽象协议。任何符合此协议的对象都可以被注入。 def chat_completion( self, messages: List[Dict[str, str]], model: Optional[str] None, tools: Optional[List[Dict]] None, tool_choice: Optional[str] None, **kwargs ) - Dict[str, Any]: 发起聊天补全请求。 返回的字典应至少包含与OpenAI API类似的 choices 结构。 ... # 具体实现OpenAI客户端适配器 class OpenAIClient: def __init__(self, api_key: str, base_url: Optional[str] None, default_model: str gpt-4): import openai self.client openai.OpenAI(api_keyapi_key, base_urlbase_url) self.default_model default_model def chat_completion(self, messages, modelNone, toolsNone, tool_choiceNone, **kwargs): model model or self.default_model response self.client.chat.completions.create( modelmodel, messagesmessages, toolstools, tool_choicetool_choice, **kwargs ) # 将Pydantic对象转换为字典统一接口 return json.loads(response.model_dump_json()) # 具体实现用于测试的Mock客户端 class MockLLMClient: def __init__(self, fixed_response: Dict[str, Any]): self.fixed_response fixed_response def chat_completion(self, messages, modelNone, toolsNone, tool_choiceNone, **kwargs): # 直接返回预设的响应用于单元测试 return self.fixed_response # 具体实现Claude客户端适配器 class AnthropicClient: def __init__(self, api_key: str, default_model: str claude-3-sonnet-20240229): import anthropic self.client anthropic.Anthropic(api_keyapi_key) self.default_model default_model def chat_completion(self, messages, modelNone, toolsNone, tool_choiceNone, **kwargs): # 注意Anthropic API的格式与OpenAI不同需要做适配转换 # 这里是简化的示例实际转换更复杂 converted_messages self._convert_messages(messages) response self.client.messages.create( modelmodel or self.default_model, messagesconverted_messages, # Anthropic的工具调用参数也不同 **self._adapt_kwargs(kwargs) ) return self._format_response(response)关键设计点LLMClientProtocol是契约。它不关心底层是OpenAI、Anthropic还是本地模型只要实现了chat_completion方法并返回约定格式的数据就可以被我们的Agent使用。这实现了“依赖倒置”—— 高层模块Agent依赖抽象Protocol而非具体实现。依赖注入的时机现在你的Agent在构造时接收一个LLMClientProtocol类型的参数。class MyAgent: def __init__(self, llm_client: LLMClientProtocol, ...): # 依赖抽象 self.llm_client llm_client def process(self, input_text): messages [{role: user, content: input_text}] # Agent内部只调用协议定义的方法 response self.llm_client.chat_completion(messagesmessages) return response在应用入口如main.py或工厂类中你决定具体注入哪个实现# 生产环境注入真实的OpenAI客户端 from config import OPENAI_API_KEY llm_client OpenAIClient(api_keyOPENAI_API_KEY) agent MyAgent(llm_clientllm_client, ...) # 测试环境注入Mock客户端 mock_response { choices: [{ message: { content: 这是模拟的回复, tool_calls: [...] } }] } test_client MockLLMClient(fixed_responsemock_response) test_agent MyAgent(llm_clienttest_client, ...) # 现在可以快速、无成本地运行单元测试了通过这种方式Client的切换成本降至最低测试也变得极其简单。5. 实践二解耦 Prompt —— 模板化、外部化与动态渲染提示词是LLM应用的“灵魂”但它不应该成为代码的“枷锁”。解耦Prompt的目标是将提示词从代码中移出使其成为可配置、可管理的数据。5.1 设计 PromptTemplate 抽象首先定义一个提示词模板的抽象它负责两件事1. 存储模板内容2. 根据输入变量渲染出最终的提示词。from string import Template from typing import Dict, Any class PromptTemplate: def __init__(self, template_str: str): # 使用Python的string.Template或自定义逻辑 self.template Template(template_str) def render(self, **kwargs) - str: 使用关键字参数渲染模板。 try: return self.template.substitute(**kwargs) except KeyError as e: raise ValueError(fMissing variable {e} in prompt template) from e # 示例从配置文件或数据库加载模板 weather_prompt_str 你是一个专业的天气助手。 用户的问题是${user_query} 请根据上下文调用合适的工具来回答。 如果用户问题不明确请礼貌地请求澄清。 上下文${context} weather_template PromptTemplate(weather_prompt_str) # 渲染 final_prompt weather_template.render( user_query北京今天天气怎么样, context用户位于中国北京。 )5.2 构建 PromptProvider 集中管理单一的模板不够我们需要一个中心化的管理器来存放所有模板并可能根据不同的场景或版本提供不同的模板。class PromptProvider: def __init__(self): self._templates: Dict[str, PromptTemplate] {} def register_template(self, name: str, template: PromptTemplate): self._templates[name] template def get_template(self, name: str) - PromptTemplate: if name not in self._templates: raise KeyError(fPrompt template {name} not found.) return self._templates[name] def render(self, name: str, **kwargs) - str: template self.get_template(name) return template.render(**kwargs) # 初始化Provider并注册模板 prompt_provider PromptProvider() prompt_provider.register_template(weather_assistant, PromptTemplate(weather_prompt_str)) prompt_provider.register_template(customer_service, PromptTemplate(customer_service_prompt_str))5.3 与Agent集成现在Agent不再硬编码提示词字符串而是依赖PromptProvider。class MyAgent: def __init__(self, llm_client: LLMClientProtocol, prompt_provider: PromptProvider, ...): self.llm_client llm_client self.prompt_provider prompt_provider def answer_weather(self, user_query: str, user_context: Dict): # 从Provider获取并渲染提示词 prompt_text self.prompt_provider.render( weather_assistant, user_queryuser_query, contextjson.dumps(user_context, ensure_asciiFalse) ) messages [{role: user, content: prompt_text}] response self.llm_client.chat_completion(messagesmessages) return self._parse_response(response)这样做带来的巨大优势动态切换你可以轻松实现A/B测试。创建两个不同的PromptTemplate比如一个详细版一个简洁版注册为weather_assistant_v1和weather_assistant_v2然后在运行时根据用户ID或实验配置决定使用哪一个。外部化管理模板内容可以存储在YAML、JSON文件或数据库中。你甚至可以开发一个简单的管理界面让产品经理或运营同学在不重启服务的情况下修改和发布提示词。版本控制PromptProvider可以扩展以支持模板版本方便回滚和对比。易于测试在单元测试中你可以注入一个MockPromptProvider让它返回固定的提示词从而隔离测试Agent的业务逻辑。实操心得提示词模板中尽量使用明确的变量名并做好参数校验。我曾因为变量名拼写错误${userQuery}vs${user_query}导致渲染失败问题很难排查。现在我会在PromptTemplate.render方法中加入严格的参数检查和清晰的错误信息。6. 实践三解耦 Tool Registry —— 注册、发现与执行分离工具调用Function Calling/Tool Calling是构建复杂Agent的核心。紧耦合的工具实现会让Agent变得臃肿且难以测试。我们的目标是将工具的定义、注册和执行与Agent的核心流程分离。6.1 定义工具接口首先定义一个工具函数的通用接口。一个工具本质上是一个可调用的对象它有名字、描述、参数模式并返回一个结果。from typing import Callable, Dict, Any from pydantic import BaseModel class ToolDefinition(BaseModel): 工具的定义用于向LLM描述工具。 name: str description: str parameters: Dict[str, Any] # 可以使用JSON Schema class RegisteredTool: 注册的工具对象包含定义和具体的执行函数。 def __init__(self, definition: ToolDefinition, func: Callable): self.definition definition self.func func def execute(self, **kwargs) - Any: 执行工具函数。 return self.func(**kwargs)6.2 构建中心化的 ToolRegistryToolRegistry是一个中心化的仓库负责所有工具的注册、查找和格式转换将工具定义转换为LLM API所需的格式。class ToolRegistry: def __init__(self): self._tools: Dict[str, RegisteredTool] {} def register(self, tool: RegisteredTool): if tool.definition.name in self._tools: raise ValueError(fTool {tool.definition.name} is already registered.) self._tools[tool.definition.name] tool def get_tool(self, name: str) - RegisteredTool: tool self._tools.get(name) if not tool: raise KeyError(fTool {name} not found in registry.) return tool def get_tools_for_llm(self) - List[Dict]: 将注册的工具转换为LLM API所需的列表格式。 return [tool.definition.model_dump() for tool in self._tools.values()] def execute(self, tool_name: str, **kwargs) - Any: 根据工具名和参数执行工具。 tool self.get_tool(tool_name) return tool.execute(**kwargs)6.3 定义并注册工具现在工具函数可以定义在任何地方只需在应用启动时向ToolRegistry注册。# 工具函数定义在独立的模块中例如 tools/weather_tool.py def get_current_weather(location: str, unit: str celsius) - str: 模拟获取天气的工具。在生产中这里会调用真实API。 # 模拟API调用 print(f[Tool Call] 正在获取 {location} 的天气单位{unit}) # 返回模拟数据 return f{location}的天气是晴朗22度。 # 创建工具定义 weather_tool_definition ToolDefinition( nameget_current_weather, description获取指定城市的当前天气信息, parameters{ type: object, properties: { location: {type: string, description: 城市名称例如北京San Francisco}, unit: {type: string, enum: [celsius, fahrenheit], description: 温度单位} }, required: [location] } ) # 在主程序或工厂中注册 registry ToolRegistry() registry.register(RegisteredTool(definitionweather_tool_definition, funcget_current_weather)) # 可以注册更多工具 # registry.register(calculator_tool) # registry.register(search_tool)6.4 在Agent中集成ToolRegistryAgent现在只持有ToolRegistry的引用而不关心工具的具体实现。class MyAgent: def __init__(self, llm_client: LLMClientProtocol, prompt_provider: PromptProvider, tool_registry: ToolRegistry): self.llm_client llm_client self.prompt_provider prompt_provider self.tool_registry tool_registry def run(self, user_input: str): # 1. 准备提示词和消息 prompt self.prompt_provider.render(agent_with_tools, queryuser_input) messages [{role: user, content: prompt}] # 2. 从Registry获取工具列表并调用LLM available_tools self.tool_registry.get_tools_for_llm() response self.llm_client.chat_completion( messagesmessages, toolsavailable_tools, tool_choiceauto ) # 3. 解析LLM响应处理工具调用 message response[choices][0][message] if hasattr(message, tool_calls) and message.tool_calls: for tool_call in message.tool_calls: tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments) # 关键步骤通过Registry执行工具Agent不关心工具如何实现 tool_result self.tool_registry.execute(tool_name, **tool_args) # 将结果追加到消息历史准备下一次LLM调用... messages.append({ role: tool, content: str(tool_result), tool_call_id: tool_call.id }) # 可能需要进行多轮工具调用... # 4. 返回最终结果 final_answer message.content return final_answer解耦带来的测试便利性 现在我们可以对ToolRegistry和Agent进行独立的、彻底的单元测试。# 测试 ToolRegistry def test_tool_registry_execution(): registry ToolRegistry() # 注册一个模拟工具 def mock_tool(x): return x*2 registry.register(RegisteredTool(definitionToolDefinition(namemock, ...), funcmock_tool)) result registry.execute(mock, x5) assert result 10 # 测试 Agent (使用Mock) def test_agent_tool_calling_logic(): # 创建Mock依赖 mock_client MockLLMClient(fixed_response{ choices: [{ message: { tool_calls: [{ id: call_123, function: {name: get_weather, arguments: {location: Beijing}} }] } }] }) mock_prompt_provider MockPromptProvider() mock_registry MockToolRegistry() agent MyAgent(mock_client, mock_prompt_provider, mock_registry) agent.run(test) # 验证Agent是否正确地调用了Registry的execute方法 mock_registry.execute.assert_called_once_with(get_weather, locationBeijing)通过将工具逻辑外置Agent的核心流程变得清晰且可测试工具函数本身也可以被独立复用和测试。7. 组装与依赖注入容器的选择我们已经成功地将三大依赖解耦成了独立的、可注入的组件。最后一步就是如何优雅地将它们“组装”起来。最简单的方式是在应用入口如main.py手动创建和注入# main.py - 手动装配 def create_production_agent(): # 1. 创建并配置Client llm_client OpenAIClient(api_keyos.getenv(OPENAI_API_KEY)) # 2. 创建并加载Prompt Provider prompt_provider PromptProvider() with open(prompts/weather.yaml, r) as f: prompts yaml.safe_load(f) for name, template_str in prompts.items(): prompt_provider.register_template(name, PromptTemplate(template_str)) # 3. 创建并注册Tool Registry tool_registry ToolRegistry() from my_tools import weather_tool, calculator_tool tool_registry.register(weather_tool) tool_registry.register(calculator_tool) # 4. 注入依赖创建Agent agent MyAgent( llm_clientllm_client, prompt_providerprompt_provider, tool_registrytool_registry ) return agent if __name__ __main__: agent create_production_agent() result agent.run(今天上海热吗) print(result)对于更复杂的应用依赖关系图可能非常庞大。这时可以考虑使用依赖注入容器DI Container例如 Python 中的dependency-injector或injector库。它们可以自动管理组件的生命周期和依赖关系。# 使用 dependency-injector 的示例简化 from dependency_injector import containers, providers class Container(containers.DeclarativeContainer): config providers.Configuration() llm_client providers.Singleton( OpenAIClient, api_keyconfig.openai.api_key ) prompt_provider providers.Singleton( PromptProvider ) tool_registry providers.Singleton( ToolRegistry ) agent providers.Factory( MyAgent, llm_clientllm_client, prompt_providerprompt_provider, tool_registrytool_registry ) # 使用容器 container Container() container.config.openai.api_key.from_env(OPENAI_API_KEY) agent container.agent()DI容器在大型项目中能显著提升配置管理的整洁度但在中小型项目中手动装配通常更简单直接。选择哪种方式取决于项目的复杂度和团队偏好。8. 总结与进阶思考从解耦到可观测性通过以上三步——定义Client抽象接口、外部化管理Prompt、中心化注册Tool——我们成功地将一个紧耦合的LLM应用重构为高度模块化、可测试的系统。这套模式带来的好处是长期的团队协作清晰前端工程师可以专注于UI算法工程师可以优化Prompt后端工程师可以维护Tool的实现大家通过清晰的接口契约协作。技术栈升级无忧明天有一个新的LLM API发布只需实现一个新的LLMClientProtocol适配器然后在创建Agent时换掉它。核心业务代码纹丝不动。测试覆盖率提升每个组件都可以被单独测试。你可以用Mock对象模拟LLM的任意输出来测试Agent复杂的工具调用链条和错误处理逻辑而无需花费一分钱API调用费。进阶一步可观测性Observability当系统解耦后添加可观测性也变得异常简单。你可以在LLMClientProtocol的包装层、ToolRegistry.execute方法等处轻松加入日志、指标Metrics和追踪Trace。例如创建一个InstrumentedLLMClient装饰器class InstrumentedLLMClient(LLMClientProtocol): def __init__(self, wrapped_client: LLMClientProtocol): self._client wrapped_client def chat_completion(self, messages, modelNone, toolsNone, **kwargs): start_time time.time() try: response self._client.chat_completion(messages, model, tools, **kwargs) # 记录成功指标 record_metric(llm_calls_success, 1) record_latency(llm_call_latency, time.time() - start_time) return response except Exception as e: # 记录失败指标 record_metric(llm_calls_failure, 1) raise e然后在生产环境的装配过程中将真实的Client包装一下即可base_client OpenAIClient(api_keyapi_key) instrumented_client InstrumentedLLMClient(wrapped_clientbase_client) agent MyAgent(llm_clientinstrumented_client, ...)这样你就能无侵入地获得所有LLM调用的耗时、成功率等关键指标为性能优化和故障排查提供数据支持。重构的初期可能会觉得增加了些许复杂度但这是为了换取长期的灵活性与可维护性。当你需要应对快速变化的AI模型生态、复杂的业务需求以及严格的代码质量要求时一个基于依赖注入的清晰架构将是你的最强后盾。