Mini-Agent 之学习组件源码

本文最后更新于82 天前，其中的信息可能已经过时，如有错误请发送邮件到 2915475627@qq.com

简介

项目架构如图：

flowchart TB
    A[config.py] --> B[schema/
数据结构]
    A --> C[llm/
LLMClient + 两个策略]
    A --> D[tools/
Tool 基类 + 6 种内置工具]

    B --> E[agent.py
Agent 运行时]
    C --> E
    D --> E

    E --> F[cli.py
用户入口]

源码前置已学习：

agent 运行时实现

CLI 命令行交互

本篇学习：

Config

Schema

LLM 抽象层

工具系统

skills

Config

config.py 就是带校验的 yaml 解析器，多级路径解析适配。

嵌套结构

  Config (顶层)
  ├── llm: LLMConfig                      # api_key, api_base, model, provider
  │   └── retry: RetryConfig              # enabled, max_retries, initial_delay ...
  ├── agent: AgentConfig                  # max_steps, workspace_dir, system_prompt_path
  └── tools: ToolsConfig                  # 哪些工具启用，skills/mcp 路径
      └── mcp: MCPConfig                  # connect_timeout, execute_timeout ...

  # 对应 config.yaml 的结构：
  #   api_key: xxx
  #   model: MiniMax-M2.5
  #   provider: anthropic
  #   retry:
  #     max_retries: 3
  #   max_steps: 50
  #   tools:
  #     enable_bash: true
  #     skills_dir: ./skills
  #     mcp:
  #       connect_timeout: 10.0

Pydantic Model 类型检查

使用 pydantic 好处：

直接传对象，缺属性用默认值
类型不对自动异常

给个示例：

from pydantic import BaseModel, Field

class LLMConfig(BaseModel):
    """LLM configuration"""

    api_key: str
    api_base: str = "https://api.minimax.io"
    model: str = "MiniMax-M2.5"
    provider: str = "anthropic"  # "anthropic" or "openai"
    retry: RetryConfig = Field(default_factory=RetryConfig)

三级优先路径解析

  # 第 177-206 行
  def find_config_file(filename):
      ① dev_config   = ./mini_agent/config/{filename}     # 开发模式
      ② user_config  = ~/.mini-agent/config/{filename}    # 用户目录
      ③ package_config = {安装包}/config/{filename}       # pip 安装后

      按 ① → ② → ③ 顺序找，找到就返回，都找不到返回 None

用户目录：Windows/Linux 用户目录

Schema

schema.py 存了通用类，可以类比前端的 /typing，Java 的 /types。

具体包含 LLMProvider，Message，LLMResponse，TokeUsage。这些类是系统内部统一格式，不依赖模型提供商，需要与其进行转换。

LLMProvide

字符串枚举类，可以这样赋值，provider=LLMProvider.ANTHROPIC

class LLMProvider(str, Enum):
    """LLM provider types."""

    ANTHROPIC = "anthropic"
    OPENAI = "openai"

为什么必须是LLMProvider(str, Enum)？

如果只有 str，就不是枚举类型限制。

如果只有 Enum，序列化会出问题，provider 的值会变成 ANTHROPIC。

Message

class Message(BaseModel):
    """Chat message."""

    role: str  # "system", "user", "assistant", "tool"
    content: str | list[dict[str, Any]]  # Can be string or list of content blocks
    thinking: str | None = None  # Extended thinking content for assistant messages
    tool_calls: list[ToolCall] | None = None
    tool_call_id: str | None = None
    name: str | None = None  # For tool role

一个类承载四种角色，通过 role 区分，可选字段按角色各取所需：

role	用到的字段
system	content
user	content
assistant	content, thinking?, tool_calls?
tool	content, tool_call_id, name

LLMResponse

不管 Anthropic 还是 OpenAI，generate() 都返回这个。Agent 用它判断：tool_calls 非空 → 执行工具继续；为空 → 任务完成

class LLMResponse(BaseModel):
    """LLM response."""

    content: str
    thinking: str | None = None  # Extended thinking blocks
    tool_calls: list[ToolCall] | None = None
    finish_reason: str
    usage: TokenUsage | None = None  # Token usage from API response

ToolCall，FunctionCall

就是 OpenAI/Anthropic 返回的 tool_call 的统一表示。两个 parser 各自把不同格式归一化成这个。

class FunctionCall(BaseModel):
    """Function call details."""

    name: str
    arguments: dict[str, Any]  # Function arguments as dict

class ToolCall(BaseModel):
    """Tool call structure."""

    id: str
    type: str  # "function"
    function: FunctionCall

TokeUsage

两个 parser 各自从 API response 里提取，填进这个统一结构。Agent 用 total_tokens 判断要不要触发上下文压缩。

class TokenUsage(BaseModel):
    """Token usage statistics from LLM API response."""

    prompt_tokens: int = 0
    completion_tokens: int = 0
    total_tokens: int = 0

LLM 抽象层

在 agent 运行时中已有详细解释，用 LLMClient 工厂提供统一调用，设计接口 LLMClientBase 规范不同实现。

classDiagram

%% Context
class LLMClient {
    &lt&ltContext>>
    - _client
}

%% Strategy Interface
class BaseLLMClient {
    &lt&ltStrategy>>
}

%% Concrete Strategies
class AnthropicClient {
    &lt&ltConcreteStrategy>>
}

class OpenAIClient {
   &lt&ltConcreteStrategy>>
}

%% Relationships
LLMClient --> BaseLLMClient : uses >
BaseLLMClient --> AnthropicClient : inherits
BaseLLMClient --> OpenAIClient : inherits

%% Agent injection
Agent --> LLMClient : inject

/llm 下文件不同职责如下

base.py

  class LLMClientBase(ABC):
      async def generate(messages, tools) -> LLMResponse  # 唯一入口
      def _convert_messages(messages) -> (system, api_msgs)  # 消息格式转换
      def _prepare_request(messages, tools) -> dict           # 组装请求

实现类需要实现这三个方法

llm_wrapper.py

根据参数实例不同客户端

  if provider == LLMProvider.ANTHROPIC:
      self._client = AnthropicClient(...)
  elif provider == LLMProvider.OPENAI:
      self._client = OpenAIClient(...)

anthropic_client.py / openai_client.py

两种实现的差异具体如下。

转换项	Anthropic	OpenAI
消息格式	`system` 为独立字段； `user` / `assistant` 交替； `tool_result` 使用 `role: user` 包裹	`system` 也作为 `messages` 数组成员； `tool` 使用独立 `role: tool`
工具格式	`input_schema`	`type: function` + `parameters`
思考内容	`type: thinking` 的 content block	`reasoning_details` 字段，需开启 `reasoning_split: true`

工具系统

基类与实现类

base.py

  class Tool:
      name: str              # 工具名，也是 LLM 调用的标识
      description: str        # 描述，注入 system prompt 让 LLM 知道何时调用
      parameters: dict        # JSON Schema，定义参数格式
      execute(**kwargs)       # 异步执行，返回 ToolResult

  class ToolResult:
      success: bool
      content: str            # 成功时的输出
      error: str | None       # 失败时的错误信息

实现类

文件	工具	功能
file_tools.py	`read_file` / `write_file` / `edit_file`	文件读写与编辑，相对路径基于 `workspace_dir` 解析
bash_tool.py	`bash` / `bash_output` / `bash_kill`	Shell 执行，支持前台/后台运行、超时 kill、输出轮询
note_tool.py	`record_note` / `recall_notes`	Agent 记忆系统，按 `category` 分类记录，持久化到 JSON 文件
skill_tool.py	`get_skill`	Skill 按需加载，Agent 通过此工具获取完整 skill 指令
mcp_loader.py	`MCPTool`	MCP 协议工具包装器，将外部 MCP Server 的工具映射为本地 Tool 对象

怎么实现 tool call？

涉及 /schema/schema.py 的三个类

  # 最底层：一个具体的函数调用
  class FunctionCall(BaseModel):
      name: str                    # 函数名，如 "read_file"
      arguments: dict[str, Any]    # 参数，如 {"path": "/foo.txt"}

  # 中间层：一次工具调用（LLM 一次可能同时调多个工具）
  class ToolCall(BaseModel):
      id: str                      # 唯一标识，用于把结果关联回来
      type: str                    # "function"
      function: FunctionCall       # 具体调哪个函数

  # 顶层：消息载体
  class Message(BaseModel):
      role: str                    # "assistant" 时可能带 tool_calls
      content: str
      tool_calls: list[ToolCall] | None   # assistant 消息里携带的批量工具调用
      tool_call_id: str | None            # tool 消息里用来关联结果的 id

完整往返流程

第 N 轮 LLM 返回

LLMResponse
  tool_calls = [
    ToolCall(
      id="toolu_001",
      type="function",
      function=FunctionCall(
        name="read_file",
        arguments={"path": "/foo.txt"}
      )
    ),
    ToolCall(
      id="toolu_002",
      function=FunctionCall(
        name="bash",
        arguments=...
      )
    )
  ]

存入 Message(role=”assistant”, tool_calls=[…])
追加到对话历史 self.messages

执行阶段

for tool_call in response.tool_calls:
  name = tool_call.function.name      # "read_file"
  args = tool_call.function.arguments # {"path": ...}
  tool = self.tools[name]             # 查表
  result = await tool.execute(**args) # 执行

回填 tool 消息

Message(
  role="tool",
  content="line 1|hello world",
  tool_call_id="toolu_001",  # 用 id 关联
  name="read_file"
)

添加到消息列表

第 N+1 轮 LLM 调用
- 下轮 generate() 时，整个 messages 历史一起发给 LLM
- LLM 看到 tool_call_id=”toolu_001″ 的结果，继续推理

skills

skill 系统的实现依靠 /tools/skill_loader.py。/skills 下存放所有下载的 skill。

Agent Skill 系统设计者与 Skill 实现者职责边界

Agent系统负责什么？

Agent 系统需要给出规范 skill 设计规范，skill 设计者负责遵守。还要从程序上实现对 skill 的解析和调用。

skill 的元数据规范
skill 的注册方式
skill 的调度方式
沙箱&隔离&安全
版本

本项目 /skills 有三个文档给出了 skill 规范。

skill 实现者负责什么？

按照系统要求规范实现 skill
写好 discription 供 LLM 理解准确
规范设计 schema
设计异常处理

渐进式披露设计

渐进式披露设计上分层披露 skill 给大模型，起到节约上下文和 token 的作用。

渐进式披露设计其实是两层披露 skill 信息给大模型，但是之间有一步是根据 skill 生成 prompt，拆开步骤更清晰。

第一步

cli.py 启动时把工具名字和描述加入系统提示词。

  if skill_loader:
      skills_metadata = skill_loader.get_skills_metadata_prompt()
      system_prompt = system_prompt.replace("{SKILLS_METADATA}", skills_metadata)

第二步

Agent 按需加载完整 skill，生成 prompt。

skill_tool.py 注册了一个 get_skill 工具。当 LLM 判断某个任务需要特定 skill 时，它会主动发起 tool call：

LLM: → tool_call: get_skill(skill_name=”pdf”)

GetSkillTool.execute() 从 SkillLoader.loaded_skills 中取出完整内容，调用 skill.to_prompt() 生成提示词返回给 LLM。

第三步

用 prompt 调用大模型，就是常规的调用模型。

简介

Config

嵌套结构

Pydantic Model 类型检查

三级优先路径解析

Schema

LLMProvide

Message

LLMResponse

ToolCall，FunctionCall

TokeUsage

LLM 抽象层

base.py

llm_wrapper.py

anthropic_client.py / openai_client.py

工具系统

基类与实现类

怎么实现 tool call？

skills

Agent Skill 系统设计者与 Skill 实现者职责边界

渐进式披露设计

第一步

第二步

第三步

发送评论 编辑评论

推荐文章

发送评论编辑评论