从零构建Agent框架：LLM与Tool服务的模块化实现指南

2025年11月22日 互联网

从零构建Agent框架：LLM与Tool服务的模块化实现指南

一、引言：Agent框架的服务层设计意义

在Agent框架的架构中，服务层（Service Layer）是连接智能体核心逻辑与外部能力的桥梁。它通过抽象化LLM（大语言模型）和Tool（工具）的调用接口，实现以下核心价值：

  1. 解耦性：将模型推理与工具执行分离，降低系统耦合度
  2. 可扩展性：支持动态添加新模型或工具而无需修改核心逻辑
  3. 复用性：统一的服务接口使不同Agent可共享相同能力

本篇将聚焦LLM Service和Tool Service的实现，通过Python代码展示如何构建这两个关键模块，并讨论设计模式的选择。

二、LLM Service实现：模型调用的抽象层

1. 基础接口设计

  1. from abc import ABC, abstractmethod
  2. from typing import Dict, Any, Optional
  4. class LLMService(ABC):
  5.     """大语言模型服务抽象基类"""
  7.     @abstractmethod
  8.     async def complete(
  9.         self,
  10.         prompt: str,
  11.         system_prompt: Optional[str] = None,
  12.         temperature: float = 0.7,
  13.         max_tokens: int = 2000,
  14.         **kwargs
  15.     ) -> Dict[str, Any]:
  16.         """完成文本生成"""
  17.         pass
  19.     @abstractmethod
  20.     def get_model_info(self) -> Dict[str, Any]:
  21.         """获取模型信息"""
  22.         pass

2. 具体实现示例（OpenAI兼容版）

  1. import aiohttp
  2. from .base import LLMService
  4. class OpenAICompatibleLLM(LLMService):
  5.     def __init__(self, api_key: str, base_url: str = "https://api.openai.com/v1"):
  6.         self.api_key = api_key
  7.         self.base_url = base_url
  8.         self.model = "gpt-3.5-turbo"  # 默认模型
  10.     async def complete(self, prompt: str, **kwargs) -> Dict:
  11.         url = f"{self.base_url}/chat/completions"
  12.         headers = {"Authorization": f"Bearer {self.api_key}"}
  14.         messages = [{"role": "user", "content": prompt}]
  15.         if kwargs.get("system_prompt"):
  16.             messages.insert(0, {"role": "system", "content": kwargs["system_prompt"]})
  18.         payload = {
  19.             "model": self.model,
  20.             "messages": messages,
  21.             "temperature": kwargs.get("temperature", 0.7),
  22.             "max_tokens": kwargs.get("max_tokens", 2000)
  23.         }
  25.         async with aiohttp.ClientSession() as session:
  26.             async with session.post(url, json=payload, headers=headers) as resp:
  27.                 return await resp.json()
  29.     def get_model_info(self) -> Dict:
  30.         return {"provider": "openai", "model": self.model}

3. 设计要点解析

  1. 异步支持：使用async/await处理网络请求，避免阻塞事件循环
  2. 参数标准化：统一温度、最大token等参数的命名和默认值
  3. 消息格式化：自动处理系统提示与用户提示的拼接
  4. 错误处理：实际实现中应添加重试机制和异常捕获

三、Tool Service实现：工具调用的标准化方案

1. 工具注册与发现机制

  1. from typing import Callable, Dict, List, Coroutine, Any
  3. class ToolRegistry:
  4.     """工具注册中心"""
  6.     def __init__(self):
  7.         self._tools: Dict[str, Dict] = {}
  9.     def register(
  10.         self,
  11.         name: str,
  12.         func: Callable[..., Coroutine[Any, Any, Any]],
  13.         description: str,
  14.         required_params: List[str] = None
  15.     ) -> None:
  16.         """注册工具"""
  17.         self._tools[name] = {
  18.             "func": func,
  19.             "description": description,
  20.             "required_params": required_params or []
  21.         }
  23.     def get_tool(self, name: str) -> Dict:
  24.         """获取工具信息"""
  25.         return self._tools.get(name)
  27.     async def execute(self, name: str, **kwargs) -> Any:
  28.         """执行工具"""
  29.         tool = self._tools.get(name)
  30.         if not tool:
  31.             raise ValueError(f"Tool {name} not found")
  33.         # 参数校验逻辑...
  34.         return await tool["func"](**kwargs)

2. 具体工具实现示例（搜索引擎）

  1. from .registry import ToolRegistry
  2. import aiohttp
  4. async def search_web(query: str, num_results: int = 5) -> Dict:
  5.     """模拟搜索引擎工具"""
  6.     # 实际实现应调用真实搜索引擎API
  7.     return {
  8.         "results": [
  9.             {"title": f"Result {i}", "url": f"https://example.com/{i}", "snippet": "Sample snippet"}
  10.             for i in range(num_results)
  11.         ]
  12.     }
  14. # 注册工具
  15. registry = ToolRegistry()
  16. registry.register(
  17.     name="web_search",
  18.     func=search_web,
  19.     description="Perform web search",
  20.     required_params=["query"]
  21. )

3. 工具调用链实现

  1. class ToolChain:
  2.     """工具调用链管理器"""
  4.     def __init__(self, registry: ToolRegistry):
  5.         self.registry = registry
  6.         self.chain = []
  8.     def add_tool(self, tool_name: str, input_mapping: Dict[str, str]) -> None:
  9.         """添加工具到调用链"""
  10.         self.chain.append({
  11.             "tool": tool_name,
  12.             "input_mapping": input_mapping  # 输入参数映射
  13.         })
  15.     async def execute(self, initial_input: Dict) -> Any:
  16.         """执行工具链"""
  17.         current_input = initial_input
  18.         for step in self.chain:
  19.             tool = self.registry.get_tool(step["tool"])
  20.             if not tool:
  21.                 raise ValueError(f"Tool {step['tool']} not found")
  23.             # 构建工具调用参数
  24.             tool_args = {
  25.                 k: current_input.get(v) 
  26.                 for k, v in step["input_mapping"].items()
  27.             }
  29.             # 执行工具并更新输入
  30.             current_input = await self.registry.execute(step["tool"], **tool_args)
  32.         return current_input

四、服务集成与最佳实践

1. 服务依赖注入模式

  1. class AgentContext:
  2.     """Agent上下文，管理服务依赖"""
  4.     def __init__(self):
  5.         self.llm_service: Optional[LLMService] = None
  6.         self.tool_registry: Optional[ToolRegistry] = None
  8.     def set_llm_service(self, service: LLMService) -> None:
  9.         self.llm_service = service
  11.     def set_tool_registry(self, registry: ToolRegistry) -> None:
  12.         self.tool_registry = registry

2. 服务配置管理建议

  1. 环境变量配置：使用.env文件管理API密钥等敏感信息
  2. 服务发现：支持从配置文件动态加载服务实现
  3. 熔断机制：为LLM服务添加请求限流和失败重试
  4. 缓存层：对频繁调用的工具结果进行缓存

3. 测试策略

  1. import pytest
  2. from unittest.mock import AsyncMock
  3. from my_agent.services import OpenAICompatibleLLM
  5. @pytest.mark.asyncio
  6. async def test_llm_completion():
  7.     # 创建模拟响应
  8.     mock_response = {
  9.         "choices": [{"message": {"content": "Test response"}}]
  10.     }
  12.     # 创建带模拟客户端的LLM服务
  13.     llm = OpenAICompatibleLLM("fake-key")
  14.     llm.client = AsyncMock()
  15.     llm.client.post.return_value.__aenter__.return_value.json = AsyncMock(return_value=mock_response)
  17.     # 测试调用
  18.     result = await llm.complete("Test prompt")
  19.     assert result["choices"][0]["message"]["content"] == "Test response"

五、扩展性设计：插件架构

1. 插件接口定义

  1. from typing import Protocol
  3. class AgentPlugin(Protocol):
  4.     """Agent插件协议"""
  6.     async def setup(self, context: AgentContext) -> None:
  7.         """插件初始化"""
  8.         pass
  10.     async def teardown(self) -> None:
  11.         """插件清理"""
  12.         pass

2. 插件加载示例

  1. import importlib
  2. from pathlib import Path
  3. from typing import List
  5. class PluginManager:
  6.     """插件管理器"""
  8.     def __init__(self, plugin_dirs: List[str]):
  9.         self.plugin_dirs = plugin_dirs
  10.         self.plugins: List[AgentPlugin] = []
  12.     async def load_plugins(self, context: AgentContext) -> None:
  13.         """加载所有插件"""
  14.         for dir_path in self.plugin_dirs:
  15.             for entry in Path(dir_path).glob("*.py"):
  16.                 module_name = entry.stem
  17.                 spec = importlib.util.spec_from_file_location(
  18.                     module_name,
  19.                     str(entry)
  20.                 )
  21.                 if spec is None:
  22.                     continue
  24.                 module = importlib.util.module_from_spec(spec)
  25.                 spec.loader.exec_module(module)
  27.                 if hasattr(module, "PLUGIN_CLASS"):
  28.                     plugin_class = getattr(module, "PLUGIN_CLASS")
  29.                     if isinstance(plugin_class, type) and issubclass(plugin_class, AgentPlugin):
  30.                         plugin = plugin_class()
  31.                         await plugin.setup(context)
  32.                         self.plugins.append(plugin)

六、总结与展望

本篇详细阐述了Agent框架中LLM Service和Tool Service的实现方案，核心要点包括：

  1. 通过抽象基类定义标准化接口
  2. 采用异步编程模式提升并发性能
  3. 实现工具注册与发现机制
  4. 设计可扩展的插件架构

未来方向可考虑：

  1. 添加服务监控与日志系统
  2. 实现多模型路由策略
  3. 开发可视化工具配置界面
  4. 增加服务版本管理功能

通过模块化的服务设计，开发者可以构建出既稳定又灵活的Agent系统，为后续添加规划、记忆等高级功能奠定坚实基础。

最新文章