Hcc的Blog

嵌入式 · AI · 折腾不止

0%

Agent 学习(3) 自建 Agent 框架

发表于 分类于 ，

LLM 根基实现

  一个 Agent 最重要的能力是能使用 API 调用 LLM，否则一切都是空谈。在之前的学习中我们已经实现了相关功能，现在新建 core 文件夹，并新建 LLM.py 文件，写入以下内容：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
import os
from openai import OpenAI
from dotenv import load_dotenv
from typing import List, Dict

# 加载 .env 文件中的环境变量
load_dotenv()

class LLM_Base:
    """
    该类用于调用任何兼容OpenAI接口的服务，并默认使用流式响应。
    """
    def __init__(self, model: str = None, apiKey: str = None, baseUrl: str = None, timeout: int = None):
        """
        初始化客户端。优先使用传入参数，如果未提供，则从环境变量加载。
        """
        self.model = model or os.getenv("LLM_MODEL_ID")
        apiKey = apiKey or os.getenv("LLM_API_KEY")
        baseUrl = baseUrl or os.getenv("LLM_BASE_URL")
        timeout = timeout or int(os.getenv("LLM_TIMEOUT", 60))
        
        if not all([self.model, apiKey, baseUrl]):
            raise ValueError("模型ID、API密钥和服务地址必须被提供或在.env文件中定义。")

        self.client = OpenAI(api_key=apiKey, base_url=baseUrl, timeout=timeout)

    def think(self, messages: List[Dict[str, str]], temperature: float = 0) -> str:
        """
        调用大语言模型进行思考，并返回其响应。
        """
        print(f"🧠 正在调用 {self.model} 模型...")
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,          # 传入对话消息历史
                temperature=temperature,    # 控制输出随机性，0表示贪心解码（最确定）
                stream=True,                # 流式传输
            )
            
            # 处理流式响应
            print("✅ 大语言模型响应成功:")
            collected_content = []
            for chunk in response:
                if not chunk.choices:
                    continue
                content = chunk.choices[0].delta.content or ""
                collected_content.append(content)
            return "".join(collected_content)

        except Exception as e:
            print(f"❌ 调用LLM API时发生错误: {e}")
            return None

框架接口实现

Message 类

  在 Agent 与 LLM 的交互中，对话历史是至关重要的上下文，为了规范地管理这些信息，我们设计了一个简易的 Message 类。在后续上下文工程的学习中还会对其进行进一步扩展。在 core 文件夹中新建 message.py 文件，并写入以下内容：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from typing import Optional, Dict, Any, Literal
from datetime import datetime
from pydantic import BaseModel

MessageRole = Literal["user", "assistant", "system", "tool"]

class Message(BaseModel):
    """
    消息类
    """
    content: str
    role: MessageRole
    timestamp: datetime = None
    metadata: Optional[Dict[str, Any]] = None

    def __init__(self, content: str, role: MessageRole, **kwargs):
        super().__init__(
            content=content,
            role=role,
            timestamp=kwargs.get('timestamp', datetime.now()),
            metadata=kwargs.get('metadata', {})
        )

    def to_dict(self) -> Dict[str, Any]:
        return {
            "role": self.role,
            "content": self.content
        }
    
    def __str__(self) -> str:
        return f"[{self.role}]{self.content}"

  首先，我们通过 typing.Literal 将 role 字段的取值严格限制为 “user” ,”assistant” , “system” , “tool” 四种，这直接对应 OpenAI API 的规范，保证了类型安全。除了 content 和 role 这两个核心字段外，我们还增加了 timestamp 和 metadata ，为日志记录和未来功能扩展预留了空间。最后， to_dict() 方法是其核心功能之一，负责将内部使用的 Message 对象转换为与 OpenAI API 兼容的字典格式，体现了“对内丰富，对外兼容”的设计原则。

Config 类

  Config 类的职责是将代码中硬编码配置参数集中起来并支持从环境变量中读取。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import os
from typing import Optional, Dict, Any
from pydantic import BaseModel

class Config(BaseModel):
    default_model: str = "deepseek-chat"
    default_provider: str = "deepseek"
    temperature: float = 0.7
    max_tokens: Optional[int] = None

    debug: bool = False
    log_level: str = "INFO"

    max_history_length: int = 100

    @classmethod
    def from_env(cls) -> "Config":
        """从环境变量创建配置"""
        return cls(
            debug=os.getenv("DEBUG", "false").lower() == "true",
            log_level=os.getenv("LOG_LEVEL", "INFO"),
            temperature=float(os.getenv("TEMPERATURE", "0.7")),
            max_tokens=int(os.getenv("MAX_TOKENS")) if os.getenv("MAX_TOKENS") else None
        )
    
    def to_dict(self) -> Dict[str, Any]:
        return self.dict()

  首先，我们将配置项按逻辑划分为 LLM配置、系统配置 等，使结构一目了然。其次，每个配置项都设有合理的默认值，保证了框架在零配置下也能工作。最核心的是 from_env() 类方法，它允许用户通过设置环境变量来覆盖默认配置，无需修改代码，这在部署到不同环境时尤其有用。

工具系统

  我们将从基础设施建设开始，逐步深入到自定义开发设计。本节的学习目标围绕以下三个核心方面展开：
  1. 统一的工具抽象与管理：建立标准化的Tool基类和ToolRegistry注册机制，为工具的开发、注册、发现和执行提供统一的基础设施。
  2. 实战驱动的工具开发：以数学计算工具为案例，展示如何设计和实现自定义工具，让读者掌握工具开发的完整流程。
  3. 高级整合与优化策略：通过多源搜索工具的设计，展示如何整合多个外部服务，实现智能后端选择、结果合并和容错处理，体现工具系统在复杂场景下的设计思维。

工具基类与注册机制

  在构建可扩展的工具系统时，我们需要首先建立一套标准化的基础设施。这套基础设施包括 Tool 基类、ToolRegistry 注册表，以及工具管理机制。

  我们先设计 Tool 的基类。它是整个工具系统的核心抽象，它定义了所有工具必须遵循的接口规范：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from abc import ABC, abstractmethod
from typing import Dict, Any, List

class Tool(ABC):
    """工具基类"""

    def __init(self, name: str, description: str):
        self.name = name
        self.description = description

    @abstractmethod
    def run(self, parameters: Dict[str: Any]) -> str:
        """执行工具"""
        pass

    @abstractmethod
    def get_parameters(self) -> list[ToolParameter]:
        """获取工具参数定义"""
        pass

  这个设计体现了面向对象设计的核心思想：通过统一的 run 方法接口，所有工具都能以一致的方式执行，接受字典参数并返回字符串结果，确保了框架的一致性。同时，工具具备了自描述能力，通过 get_parameters 方法能够清晰地告诉调用者自己需要什么参数，这种内省机制为自动化文档生成和参数验证提供了基础。而 name 和 description 等元数据的设计，则让工具系统具备了良好的可发现性和可理解性。

  代码中，ABC 表示抽象基类，该 Tool 类无法被实例化，若有继承 Tool 类的类，必须重定义 @abstractmethod 装饰词修饰的函数才可实例化。

  接下来，我们实现参数定义系统。为了实现复杂的参数验证和文档生成，我们设计一个 ToolParameter 类。

1
2
3
4
5
6
7
8
9
from pydantic import BaseModel
from typing import Any

class ToolParameter(BaseModel):
    name: str
    type: str
    description: str
    required: bool = True
    default: Any = None

  然后我们需要实现注册工具，即 ToolRegistry，这是工具系统的管理中枢，提供了工具的注册、发现、执行等核心功能，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
from Tools.base import Tool
from typing import Any, Callable

class ToolRegistry:
    """工具注册表"""

    def __init__(self):
        self._tools: dict[str, Tool] = {}
        self._functions: dict[str, dict[str, Any]] = {}

    def register_tool(self, tool: Tool):
        """注册Tool对象"""
        if tool.name in self._tools:
            print(f"警告，工具 {tool.name} 已存在，请重新检查")
            return
        self._tools[tool.name] = tool
        print(f"工具 {tool.name} 已注册。")

    def register_function(self, name: str, description: str, func: Callable[[str],str]):
        """
        直接注册函数作为工具(简便方式)
        
        Args:
            name: 工具名称
            description: 工具描述
            func: 工具函数
        """
        if name in self._functions:
            print(f"❌ 错误:工具 '{name}' 已存在，请重新检查。")
            return
        
        self._functions[name] = {
            "description": description,
            "func": func
        }
        print(f"工具 {name} 已注册")

  这里的注册表是一个字典类型，键为 Tool 的 name，值为 Tool 类型的变量。接下来实现工具的查询:

1
2
3
4
5
6
7
8
9
10
11
def get_tool_description(self) -> str:
    """获取所有可用工具的格式化描述"""
    description = []

    for tool in self._tools.values():
        description.append(f"- {tool.name}: {tool.description}")
    
    for name, info in self._functions.items():
        description.append(f"- {name}: {info['description']}")

    return "\n".join(description) if description else "暂无可用工具"

  接下来，我们实现一个最简单的计算工具，新建 calculate_tool.py 文件，写入如下代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
import ast, operator, math
from Tools.ToolRegistry import ToolRegistry
from Tools.base import Tool
from Tools.ToolParameter import ToolParameter

class CalculatorTool(Tool):
    name = "CalculatorTool"
    description = "这是一个简单的计算工具，可以实现加减乘除与开根号"
    def __init__(self):
        super().__init__(name="calculator",
            description="这是一个简单的计算工具，可以实现加减乘除与开根号",)
    
    def run(self, parameters):
        expression = parameters.get("expression") or parameters.get("input")
        if not expression:
            return "缺少 expression 参数"
        return self._my_calculate(expression)

    def get_parameters(self):
        return [
            ToolParameter(
                name="expression",
                type="string",
                description="需要计算的数学表达式",
            )
        ]

    def _my_calculate(self, expression: str) -> str:
        """简单的数学计算函数"""
        if not expression.strip():
            return "计算表达式不能为空"
        
        operators = {
            ast.Add: operator.add,
            ast.Sub: operator.sub,
            ast.Mult: operator.mul,
            ast.Div: operator.truediv,
        }

        # 支持的基本函数
        functions = {
            'sqrt': math.sqrt,
            'pi': math.pi,
        }
        try:
            node = ast.parse(expression, mode='eval')
            result = self._eval_node(node.body, operators, functions)
            return str(result)
        except:
            return "计算失败，请检查表达式格式"
        
    def _eval_node(self, node ,operators, functions):
        """简化的表达式求值"""
        if isinstance(node, ast.Constant):
            return node.value
        elif isinstance(node, ast.BinOp):
            left = self._eval_node(node.left, operators, functions)
            right = self._eval_node(node.right, operators, functions)
            op = operators.get(type(node.op))
            return op(left, right)
        elif isinstance(node, ast.Call):
            func_name = node.func.id
            if func_name in functions:
                args = [self._eval_node(arg, operators, functions) for arg in node.args]
                return functions[func_name](*args)
        elif isinstance(node, ast.Name):
            if node.id in functions:
                return functions[node.id]

Agent 范式的框架化实现

  SimpleAgent 是最基础的 Agent 实现，它展示了如何在框架基础上构建一个完整的对话智能体。我们将通过继承框架基类来重写 SimpleAgent。首先在 agents 文件中创建 MySimpleAgent.py，并写入如下内容：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
import re
from typing import Optional, Iterator
from core.config import Config
from core.LLM import LLM_Base
from core.message import Message
from Tools.ToolRegistry import ToolRegistry
from agents.simple_agent import SimpleAgent

class MySimpleAgent(simple_agent.SimpleAgent):
    def __init__(self, 
        name: str, 
        llm_client,
        system_prompt: Optional[str] = None,
        config: Optional[Config] = None,
        tool_registry: Optional['ToolRegistry'] = None,
        enable_tool_calling: bool = True
    ):
        super().__init__(name, llm_client, system_prompt, config)
        self.tool_registry = tool_registry
        self.enable_tool_calling = enable_tool_calling and tool_registry is not None
        print(f"✅ {name} 初始化完成，工具调用: {'启用' if self.enable_tool_calling else'禁用'}")
        

    def run(self, input_text: str,  max_tool_iterations:int = 3,**kwargs) -> str:
        """实现简单对话逻辑，支持可选工具调用"""
        print(f"{self.name} 正在处理: {input_text}")

        # 构建消息列表
        messages = []

        # 添加系统信息
        enhanced_system_prompt = self._get_enhanced_system_prompt()
        messages.append({"role":"system", "content": enhanced_system_prompt})

        # 添加历史消息
        for msg in self._history:
            messages.append({"role":msg.role, "content": msg.content})

        messages.append({"role":"user", "content":input_text})

        if not self.enable_tool_calling:
            response = self.llm_client.think(messages, **kwargs)
            self.add_history(Message(input_text, "user"))
            self.add_history(Message(response, "assistant"))
            print(f"{self.name} 响应完成")
            return response
        
        return self._run_with_tools(messages, input_text, max_tool_iterations, **kwargs)
        
    def _get_enhanced_system_prompt(self) -> str:
        base_prompt = self.system_prompt or "你是一个有用的 AI 助手"

        if not self.enable_tool_calling or not self.tool_registry:
            return base_prompt
        
        # 获取工具描述
        tools_description = self.tool_registry.get_tool_description()
        if not tools_description or tools_description == "暂无可用工具":
            return base_prompt
        
        tools_description = self.tool_registry.get_tool_description()
        if not tools_description or tools_description == "暂无可用工具":
            return base_prompt
        tools_section = "\n\n## 可用工具\n"
        tools_section += "你可以使用以下工具来帮助回答问题:\n"
        tools_section += tools_description + "\n"
        tools_section += "\n## 工具调用格式\n"
        tools_section += "当需要使用工具时，请使用以下格式:\n"
        tools_section += "`[TOOL_CALL:{tool_name}:{parameters}]`\n"
        tools_section += "例如:`[TOOL_CALL:search:Python编程]` 或`[TOOL_CALL:memory:recall=用户信息]`\n\n"
        tools_section += "工具调用结果会自动插入到对话中，然后你可以基于结果继续回答。\n"

        return base_prompt + tools_section

  在这里我们实现了系统提示词的构建：基础 + 工具列表（取决于是否启动工具用法）然后调用大模型进行回答。

  接下来我们实现工具的多轮调用，紧接着编写如下代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def _run_with_tools(self, messages: list, input_text: str, max_tool_iterations: int, **kwargs) -> str:
    """支持工具调用的运行逻辑"""
    current_iteration = 0
    final_response = ""

    while current_iteration < max_tool_iterations:
        response = self.llm_client.think(messages)

        tool_calls = self._parse_tool_calls(response)

        if tool_calls:
            print(f"检测到 {len(tool_calls)} 个工具调用")
            # 执行所有工具调用并收集结果
            tool_results = []
            clean_response = response

            for call in tool_calls:
                result = self._execute_tool_call(call['tool_name'], call['parameters'])
                print(result)
                tool_results.append(result)

                clean_response = clean_response.replace(call['original'], "")

            messages.append({"role": "assistant", "content": clean_response})
            # 添加工具结果
            tool_results_text = "\n\n".join(tool_results)
            messages.append({"role": "user", "content": f"工具执行结果:\n{tool_results_text}\n\n请基于这些结果给出完整的回答。"})
            current_iteration += 1
            continue
        # 没有工具调用，这是最终回答
        final_response = response
        break
    # 如果超过最大迭代次数，获取最后一次回答
    if current_iteration >= max_tool_iterations and not final_response:
        final_response = self.llm_client.think(messages, **kwargs)
    # 保存到历史记录
    self.add_history(Message(input_text, "user"))
    self.add_history(Message(final_response, "assistant"))
    print(f"✅ {self.name} 响应完成")
    return final_response

def _parse_tool_calls(self, text: str) -> list:
    """解析文本中的工具调用"""
    pattern = r'\[TOOL_CALL:([^:]+):([^\]]+)\]'
    matches = re.findall(pattern, text)

    tool_calls = []
    for tool_name, parameters in matches:
        tool_calls.append({
            'tool_name':tool_name,
            'parameters':parameters,
            'original': f'[TOOL_CALL:{tool_name}:{parameters}]'
        })
    
    return tool_calls

def _execute_tool_call(self, tool_name:str, parameters: str) -> str:
    """执行工具调用"""
    if not self.tool_registry:
        return f"错误：未配置工具注册表"
    
    try:
        if tool_name == 'CalculatorTool':
            result = self.tool_registry.execute_tool(tool_name, parameters)
        else:
            param_dict = self._parse_tool_parameters(tool_name, parameters)
            tool = self.tool_registry.get_tool(tool_name)
            if not tool:
                return f"❌ 错误:未找到工具 '{tool_name}'"
            result = tool.run(param_dict)
        return f"工具 {tool_name} 执行结果:\n{result}"
    except Exception as e:
        return f"工具调用失败:{str(e)}"
    

def _parse_tool_parameters(self, tool_name: str, parameters: str) -> dict:
    """智能解析工具参数"""
    param_dict = {}

    if '=' in parameters:
        if ',' in parameters:
            pairs = parameters.split(',')
            for pair in pairs:
                if '=' in pair:
                    key, value = pair.split('=', 1)
                    param_dict[key.strip()] = value.strip()
        else:
            key, value = parameters.split('=', 1)
            param_dict[key.strip()] = value.strip()
    else:
        if tool_name == 'search':
            param_dict = {'query': parameters}
        elif tool_name == 'memory':
            param_dict = {'action': 'search', 'query': parameters}
        else:
            param_dict = {'expression': parameters}
    return param_dict

  这里我们实现了工具的调用。接下来写一个测试程序，测试该 Agent 能否正常使用。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from dotenv import load_dotenv
from core.config import Config
from core.LLM import LLM_Base
from agents.MySimpleAgent import MySimpleAgent
from Tools.ToolRegistry import ToolRegistry
from Tools.calculate_tool import CalculatorTool
# 加载环境变量
load_dotenv()
# 创建LLM实例
llm_client = LLM_Base()
测试1:基础对话Agent（无工具）
print("=== 测试1:基础对话 ===")
basic_agent = MySimpleAgent(
    name="基础助手",
    llm_client=llm_client,
    system_prompt="你是一个友好的AI助手，请用简洁明了的方式回答问题。"
)
response1 = basic_agent.run("你好，请介绍一下自己")
print(f"基础对话响应: {response1}\n")

print("=== 测试2:工具调用 ===")
tool_registry = ToolRegistry()
calculator = CalculatorTool()
tool_registry.register_tool(calculator)
cal_agent = MySimpleAgent(
    name = "计算助手",
    llm_client=llm_client,
    system_prompt="你是一个友好的AI计算助手，请用简洁明了的方式回答问题。",
    tool_registry = tool_registry,
    enable_tool_calling=True
)

response1 = cal_agent.run("你好，请帮我使用工具计算 3 + 5 * 8")
print(f"计算对话响应: {response1}\n")

  运行 agent.py，终于成功运行，大模型成功给出回应，并调用工具计算式子。

总结

  在本次学习中，我们实现了一个最小 Agent 框架，在接下来的学习中我们会进一步学习 Agent 的记忆系统。