“我”：Java/Go后端开发者、有点时间想自己琢磨，想入门Agent但不想堆砌框架、希望理解底层原理的研发
上一篇链接：Java/Go后端手撸原生Agent（第二篇）

前言

前两篇文章我们从零搭建了原生ReAct智能体，并完成了Pydantic结构化JSON输出改造，告别了脆弱的文本分割解析。

但跑多工具场景时暴露了三个工程级问题：

1. 工具描述双份维护：工具参数说明既在代码里定义、又在System Prompt里手写，加一个新工具要改两处，违背后端单一事实源（SSOT）原则；
2. 状态枚举是死代码：上一篇引入了RUNNING/FINISHED枚举，但只是赋值后return，循环仍然靠for i in range(max_loop)隐式驱动，"思考→调用工具→再思考"两个阶段的行为没有被状态显式管控；
3. 参数零校验：工具run方法里直接params["expr"]硬取，LLM少传一个参数直接KeyError崩溃，没有统一的参数校验层。

本文完成三大升级，兑现上一篇结尾的拓展1和拓展3：

1. 每个工具自带Pydantic参数模型，自动生成OpenAI Function Schema和Prompt自然语言描述，工具注册即用、零侵入接入；
2. 三态状态机（THINKING/TOOL_EXECUTING/FINISHED）真正驱动while循环，状态显式管控流程分支；
3. 新增FileReadTool文件读取工具（带路径沙箱、行号范围、白名单），完成计算器+文件读取双工具实战；
4. 新增第二层死循环防护：代码层重复调用检测门禁，不靠Prompt劝。

前置说明

1. 完全复用上两篇基础文件：env_loader.py、llm_client.py（JSON Mode版本）、agent/memory.py、agent/schema.py、agent/structured_parser.py；
2. 核心改造：tools/base_tool.py（工具抽象基类升级）、tools/calculator.py（适配新契约）、main.py（三态状态机+动态Prompt+重复调用检测）；
3. 新增文件：tools/file_reader.py（文件读取工具）；
4. 删除上一篇未真正使用的RUNNING/FINISHED半成品枚举。

---

一、问题复现与根因分析

1.1 痛点1：工具信息双份维护（违反SSOT）

上一篇的写法，工具描述硬编码在System Prompt里：

SYSTEM_PROMPT = """
可用工具：
calculator：数学计算器，参数expr为数学表达式，例{"expr":"(100+20)*5"}
"""

而CalcTool类里自己也有name和desc两个属性。工具信息存在两个地方：加新工具既要写类、又要改Prompt字符串，稍有遗漏LLM就不知道新工具存在。后端开发一眼就能看出这是典型的"接口定义和文档不同步"问题——等价于Java接口上写了@ApiOperation但Swagger扫描不到、或者Go结构体tag和手写API文档对不上。

1.2 痛点2：假状态机，真for循环

上一篇的代码看起来有状态枚举：

class AgentTaskState(Enum):
    RUNNING = "running"
    FINISHED = "finished"

# 在FinishResponse分支里：
task_state = AgentTaskState.FINISHED
return parse_res.final_answer

但task_state赋值后立刻return，没有任何代码读取这个变量。循环仍然是for i in range(max_loop)固定次数驱动，"调LLM"和"执行工具"两个阶段混在同一个for循环体里靠if isinstance分支区分——这不是状态机，是枚举装饰。真正的状态机必须满足：当前状态决定本轮要做什么，非法转移要报错或拦截。

1.3 痛点3：参数裸dict硬取，零校验

def run(self, params: dict) -> str:
    expr = params["expr"]  # LLM少传expr直接KeyError

没有参数名、类型、必填性校验，工具直接消费裸字典，等价于Controller层直接接收Map而不是绑定POJO——后端工程里这是Code Review直接打回的写法。

1.4 痛点4：多工具场景死循环（比单工具更严重）

单工具计算器场景下，Prompt约束"拿到结果就final_answer"还能勉强工作。但加入文件读取工具后，需要"先读文件→再调用计算器→再回答"的多轮链路，LLM经常在拿到计算器结果后忘记自己已经算过，重复调用同一工具直到耗尽max_loop。单纯靠Prompt写"禁止重复调用"是防不住的，必须在代码层加门禁。

---

二、改造1：工具基类升级——Pydantic Schema自动生成

2.1 设计思路（后端视角）

1. 每个工具自带参数模型：等价于Java中每个API对应一个Request DTO，用Pydantic BaseModel定义参数名、类型、描述、默认值、数值范围；
2. 模板方法模式：基类实现execute(params)模板方法，内部完成参数校验→调用子类run(validatedArgs)，子类只关心业务逻辑，不用重复写校验代码；
3. 自动Schema生成：Pydantic v2内置model_json_schema()，直接生成标准JSON Schema，一键对齐OpenAI Function Call规范；
4. 自动Prompt描述：遍历参数Schema的properties，拼接成自然语言描述注入System Prompt，工具描述从此只在工具类里定义一次。

类比后端：args_schema = Request DTO类，execute = DispatcherServlet参数绑定+校验，run = Controller方法，to_openai_tool_schema = Swagger/OpenAPI自动生成接口文档。

2.2 重写工具基类 tools/base_tool.py

from abc import ABC, abstractmethod
from pydantic import BaseModel, ValidationError


class BaseTool(ABC):

    @property
    @abstractmethod
    def name(self) -> str:
        """工具唯一标识名称"""
        pass

    @property
    @abstractmethod
    def desc(self) -> str:
        """工具功能描述，给LLM看"""
        pass

    @property
    @abstractmethod
    def args_schema(self) -> type[BaseModel]:
        """参数Pydantic模型类（注意是类不是实例，不加括号）"""
        pass

    @abstractmethod
    def run(self, args: BaseModel) -> str:
        """
        工具业务逻辑（子类实现）
        :param args: 已通过Pydantic校验的参数模型实例
        :return: 工具执行结果字符串
        """
        pass

    def execute(self, params: dict) -> str:
        """
        模板方法：参数校验 → 调用run
        外部调用入口，子类不要重写
        """
        try:
            validated = self.args_schema.model_validate(params)
        except ValidationError as e:
            return f"工具{self.name}参数校验失败：{e}"
        return self.run(validated)

    def to_openai_tool_schema(self) -> dict:
        """
        生成OpenAI标准Function Call Schema
        为后续接入原生tools接口做准备
        """
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.desc,
                "parameters": self.args_schema.model_json_schema(),
            },
        }

    def to_prompt_description(self) -> str:
        """
        生成适合嵌入System Prompt的自然语言工具描述
        JSON Mode阶段使用，自动拼接参数名、类型、必填标记、描述
        """
        schema = self.args_schema.model_json_schema()
        props = schema.get("properties", {})
        required = schema.get("required", [])
        parts = []
        for field_name, field_info in props.items():
            req_mark = "*" if field_name in required else ""
            desc = field_info.get("description", "")
            type_info = field_info.get("type", "")
            parts.append(f'{field_name}{req_mark}({type_info}): {desc}')
        params_str = "; ".join(parts)
        return f"- {self.name}: {self.desc} | 参数：{params_str}"

关键设计：

• args_schema返回类型是type[BaseModel]（类本身，不是实例），等价Java的Class<ReqDTO>，用于在execute里调用model_validate；
• execute是模板方法（Template Method Pattern）：参数校验逻辑所有工具共用，业务逻辑下沉到子类run；
• model_validate是Pydantic v2的强校验入口，类型错误、缺少必填字段、数值范围越界都会抛ValidationError，被统一捕获后返回友好错误；
• to_openai_tool_schema为下一篇接入原生Function Call打基础；
• to_prompt_description自动生成形如- calculator: 数学计算器 | 参数：expr*(string): 数学表达式的描述，Prompt里不再手写。

2.3 改造计算器工具 tools/calculator.py

按照新契约重写，作为新工具的标准模板：

from pydantic import BaseModel, Field
from tools.base_tool import BaseTool


class CalcArgs(BaseModel):
    """计算器参数模型（等价Java Request DTO）"""
    expr: str = Field(description="数学表达式，支持加减乘除和括号，例如 (100+20)*5")


class CalcTool(BaseTool):

    @property
    def name(self) -> str:
        return "calculator"

    @property
    def desc(self) -> str:
        return "数学计算器，输入数学表达式返回计算结果"

    @property
    def args_schema(self) -> type[BaseModel]:
        return CalcArgs

    def run(self, args: CalcArgs) -> str:
        # 直接从校验后的模型实例取参，无需params["expr"]硬取字典
        res = eval(args.expr)
        return f"计算结果: {args.expr} = {res}"

对比旧版三个变化：新增CalcArgs模型类、实现args_schema属性、run参数从dict改为CalcArgs，用args.expr属性访问。加新工具照抄这个结构即可。

---

三、改造2：三态状态机真正驱动主循环

3.1 为什么需要三个状态而不是两个

上一篇的RUNNING/FINISHED两态设计中，"RUNNING"过于笼统——"正在等LLM思考"和"正在执行工具"是两个完全不同的阶段：

阶段	行为	下一个合法转移
THINKING（等待LLM输出）	组装messages→调用LLM→解析JSON	→ FINISHED（拿到final_answer）/→ TOOL_EXECUTING（拿到tool_call）
TOOL_EXECUTING（执行工具）	根据工具名查找工具→execute校验+执行→存observation	→ THINKING（回到LLM思考）
FINISHED（任务完成）	循环退出，返回结果	终态，无转移

如果把THINKING和TOOL_EXECUTING合并成一个RUNNING，LLM调用和工具执行就混在一个代码块里，状态无法携带上下文（"LLM决定调用哪个工具、传什么参数"这两个数据必须跨状态保留），也无法拦截非法转移。

3.2 状态携带数据：pending变量

状态机不是只有状态名，状态转移需要携带上下文数据。THINKING解析出ToolAction后，要把tool_name和tool_params带到TOOL_EXECUTING状态去执行，通过两个pending_*变量实现：

pending_tool_name = None
pending_tool_params = None

等价于Go里channel传递、Java里状态上下文对象。

3.3 三态状态机主循环 main.py

import json
from enum import Enum
from agent.memory import ShortMemory
from agent.schema import FinishResponse, ToolAction
from agent.structured_parser import StructuredParser
from llm_client import chat_completion
from tools.base_tool import BaseTool
from tools.calculator import CalcTool
from tools.file_reader import FileReadTool  # 新建的文件读取工具，见第四节


class AgentTaskState(Enum):
    THINKING = "thinking"         # 等待/刚收到LLM输出，需要解析
    TOOL_EXECUTING = "tool_executing"  # LLM要求调用工具，正在执行
    FINISHED = "finished"         # 任务完成，循环退出


# 工具注册：加新工具只需要在list里加一个实例，其他地方自动适配
tool_list: list[BaseTool] = [CalcTool(), FileReadTool()]
tool_map = {t.name: t for t in tool_list}


SYSTEM_PROMPT_TEMPLATE = """你是支持工具调用的智能助手，必须仅输出纯JSON，禁止额外文字、Markdown、换行注释。

## 可用工具
{tools_description}

## 严格执行规则
1. 需要获取信息时调用对应工具；
2. 收到工具观测结果后，判断是否已有足够信息回答用户：
   - 信息不足 → 调用其他工具（**禁止用完全相同的参数重复调用同一个工具**）；
   - 信息充足 → 必须直接输出final_answer，禁止再调用任何工具；
3. 两种输出格式严格二选一：
- 需要调用工具时：{{"thought":"推理过程","action":"工具名称","params":{{...}}}}
- 任务完成无需再调用工具：{{"final_answer":"把结果整理成自然语言回答用户"}}
"""


def build_system_prompt(tools: list[BaseTool]) -> str:
    """动态生成System Prompt，工具描述自动从工具类提取"""
    descriptions = "\n".join(t.to_prompt_description() for t in tools)
    return SYSTEM_PROMPT_TEMPLATE.format(tools_description=descriptions)


def run_agent(user_query: str):
    memory = ShortMemory()
    memory.add_user(user_query)
    max_loop = 10

    # 初始状态：THINKING
    state = AgentTaskState.THINKING
    loop_count = 0
    final_answer = None
    pending_tool_name = None
    pending_tool_params = None
    executed_calls: set[tuple[str, str]] = set()  # 重复调用检测：第五小节详述

    system_prompt = build_system_prompt(tool_list)

    while state != AgentTaskState.FINISHED and loop_count < max_loop:
        loop_count += 1

        if state == AgentTaskState.THINKING:
            # THINKING状态：组装消息→调LLM→解析→决定下一状态
            messages = [{"role": "system", "content": system_prompt}]
            messages.extend(memory.get_raw_dict_list())

            print(f"\n=== 第{loop_count}轮 THINKING ===")
            for msg in messages:
                print(msg)

            llm_raw_json = chat_completion(messages, json_mode=True)
            parse_res = StructuredParser.parse_json(llm_raw_json)

            if parse_res is None:
                final_answer = f"模型输出格式解析失败，原始内容：{llm_raw_json}"
                state = AgentTaskState.FINISHED
                break

            if isinstance(parse_res, FinishResponse):
                memory.add_assistant(parse_res.final_answer)
                final_answer = parse_res.final_answer
                state = AgentTaskState.FINISHED
                break

            if isinstance(parse_res, ToolAction):
                print(f"【推理思考】{parse_res.thought}")
                # 把工具名和参数存到pending变量，交给TOOL_EXECUTING状态消费
                pending_tool_name = parse_res.action
                pending_tool_params = parse_res.params
                state = AgentTaskState.TOOL_EXECUTING
                continue

        if state == AgentTaskState.TOOL_EXECUTING:
            # TOOL_EXECUTING状态：找工具→校验参数→执行→存结果→回THINKING
            call_key = (pending_tool_name, json.dumps(pending_tool_params, sort_keys=True, ensure_ascii=False))

            if call_key in executed_calls:
                # 重复调用门禁：不执行工具，注入纠偏提示
                obs = (
                    f"[系统纠偏] 你已经用完全相同的参数调用过{pending_tool_name}工具，"
                    f"结果已在上方消息中。禁止无限循环！请直接基于已有结果输出final_answer。"
                )
                print(f"【重复调用拦截】{pending_tool_name} {pending_tool_params}")
            else:
                print(f"【工具调用】name={pending_tool_name}, params={pending_tool_params}")
                tool = tool_map.get(pending_tool_name)
                if not tool:
                    obs = f"异常：不存在工具{pending_tool_name}"
                else:
                    obs = tool.execute(pending_tool_params)  # 走基类模板方法（含参数校验）
                print(f"【工具返回结果】{obs}")
                executed_calls.add(call_key)

            memory.add_observation(f"[{pending_tool_name}] 返回结果:\n{obs}")
            # 清空pending，状态切回THINKING
            pending_tool_name = None
            pending_tool_params = None
            state = AgentTaskState.THINKING
            continue

    if final_answer is None:
        final_answer = f"达到最大循环次数{max_loop}，任务未完成"

    return final_answer


if __name__ == "__main__":
    # 双工具测试：读文件+解释+计算
    answer = run_agent(
        "帮我读一下 tools/base_tool.py 的内容，"
        "然后解释execute和to_openai_tool_schema方法做了什么，"
        "这两个方法的行数加起来乘以4再除以2等于多少（必须调用calculator计算）"
    )
    print("\n最终回答:", answer)

核心设计要点：

1. while循环替代for循环：while state != FINISHED，状态驱动而不是轮次驱动，loop_count只是安全阀；
2. 每个if块是互斥状态分支：THINKING块里不会有工具执行代码，TOOL_EXECUTING块里不会调LLM，职责边界清晰；
3. continue驱动状态转移：每个状态处理完要么break（FINISHED）要么continue进入下一循环，新循环开头根据state值进入对应分支；
4. pending变量跨状态传数据：THINKING写入pending→TOOL_EXECUTING读取并清空→回THINKING，等价状态模式里的Context对象；
5. 动态Prompt：build_system_prompt从tool_list自动生成工具描述，加新工具改tool_list即可。

---

四、新增文件读取工具 tools/file_reader.py

基础设施搭好后，加新工具就是照抄CalcTool的模板——定义参数模型、实现四个成员、注册到tool_list，不需要改Prompt、不需要改Parser、不需要改主循环。这就是Schema自动生成的价值。

from pathlib import Path
from pydantic import BaseModel, Field
from tools.base_tool import BaseTool


# 工作区根目录：限制Agent只能读这个目录下的文件（路径沙箱）
WORKSPACE_ROOT = Path(__file__).resolve().parent.parent


class FileReadArgs(BaseModel):
    """文件读取参数模型"""
    path: str = Field(description="要读取的文件绝对路径")
    offset: int = Field(default=0, ge=0, description="起始行号，从0开始，默认0")
    limit: int = Field(default=200, gt=0, le=500, description="读取行数上限，默认200，最大500")


class FileReadTool(BaseTool):

    @property
    def name(self) -> str:
        return "read_file"

    @property
    def desc(self) -> str:
        return "读取本地文件内容，按行范围返回带行号的文本，适合阅读源代码"

    @property
    def args_schema(self) -> type[BaseModel]:
        return FileReadArgs

    def run(self, args: FileReadArgs) -> str:
        target = Path(args.path).resolve()

        # 路径沙箱校验：禁止访问工作区外文件（防 ../../etc/passwd）
        try:
            target.relative_to(WORKSPACE_ROOT)
        except ValueError:
            return f"错误：路径{args.path}不在工作目录内，禁止访问工作区外文件"

        if not target.is_file():
            return f"错误：文件{args.path}不存在或不是普通文件"

        try:
            lines = target.read_text(encoding="utf-8").splitlines()
        except Exception as e:
            return f"读取文件失败：{e}"

        total = len(lines)
        start = args.offset
        end = min(start + args.limit, total)
        selected = lines[start:end]

        # 返回带行号的内容，方便LLM引用具体行
        numbered = [f"{i+1:4d} | {line}" for i, line in enumerate(selected, start=start)]
        header = f"[文件: {target}] 共{total}行，显示{start+1}-{end}行："
        return header + "\n" + "\n".join(numbered)

三个安全/工程设计：

• 路径沙箱：target.relative_to(WORKSPACE_ROOT)校验，路径解析为绝对路径后必须在工作区内，防止../../etc/passwd越权；
• Pydantic数值范围：offset: ge=0、limit: gt=0, le=500，参数层面就拦住负数和超大行数，LLM传limit=999999会被execute直接挡在校验阶段；
• 带行号返回： 1 | from abc import ABC，方便LLM后续引用"第28行的execute方法"。

---

五、多工具死循环第二层防护：代码层重复调用检测

5.1 为什么Prompt防不住

Prompt规则写了"禁止重复调用"，但LLM是概率性的——多轮上下文长、工具结果多时，注意力被稀释，还是会"忘记"自己刚调过。单靠自然语言约束等价于在代码里写注释提醒"这里不要传null"但不写if判断——迟早出问题。

5.2 代码门禁实现

main.py的TOOL_EXECUTING状态入口，执行工具前先检查：

executed_calls: set[tuple[str, str]] = set()

# 在TOOL_EXECUTING分支里：
call_key = (pending_tool_name, json.dumps(pending_tool_params, sort_keys=True, ensure_ascii=False))

if call_key in executed_calls:
    # 不执行工具，注入强纠偏observation
    obs = "[系统纠偏] 你已经用完全相同的参数调用过..."
else:
    obs = tool.execute(pending_tool_params)
    executed_calls.add(call_key)

设计要点：

• call_key用(工具名, 参数JSON规范化字符串)作为去重键，sort_keys=True保证{"a":1,"b":2}和{"b":2,"a":1}视为同一组参数；
• 检测到重复时不执行工具（节省API调用和计算），而是注入一条[系统纠偏]消息，比Prompt里的规劝有效得多——这条消息就在当前上下文里，模型"看得到"自己被拦截了；
• 这是学习阶段的简洁实现，生产框架（LangGraph等）会进一步做state hash检测、A→B→A→B序列模式识别、反思节点等多层防护，但核心思想一致：不靠模型自觉，靠代码拦截。

---

六、运行效果

双工具场景完整链路：

=== 第1轮 THINKING ===
{'role': 'system', 'content': '...可用工具:\n- calculator: 数学计算器...| 参数：expr*(string): ...\n- read_file: 读取本地文件...| 参数：path*(string): ...; offset(integer): ...; limit(integer): ...'}
{'role': 'user', 'content': '帮我读一下 tools/base_tool.py 的内容，然后解释execute和to_openai_tool_schema方法...'}
【推理思考】用户要求先读取文件内容，再解释方法并做计算，第一步需要读取文件。

=== 第2轮 TOOL_EXECUTING ===
【工具调用】name=read_file, params={'path': 'tools/base_tool.py'}
【工具返回结果】[文件: .../tools/base_tool.py] 共56行，显示1-56行
   1 | from abc import ABC, abstractmethod
   ...
  28 |     def execute(self, params: dict) -> str:
  ...
  35 |     def to_openai_tool_schema(self) -> dict:
  ...

=== 第3轮 THINKING ===
【推理思考】文件内容已在上下文中，execute方法在28-33行共6行，to_openai_tool_schema在35-43行共9行，合计15行，需要调用计算器计算(6+9)*4/2。

=== 第4轮 TOOL_EXECUTING ===
【工具调用】name=calculator, params={'expr': '(6+9)*4/2'}
【工具返回结果】计算结果: (6+9)*4/2 = 30.0

=== 第5轮 THINKING ===
最终回答: 已读取文件内容...execute方法（第28-33行）是模板方法...to_openai_tool_schema方法（第35-43行）生成OpenAI标准Function Schema...两个方法共15行，乘以4再除以2结果是30。

共5轮，两个工具各调用1次，无重复、无死循环，最终自然语言回答整合了文件内容解释和数值计算结果。

---

七、核心改造总结

维度	上一篇（改造前）	本文（改造后）
工具参数定义	裸dict，run里硬取params["expr"]	Pydantic模型，args: CalcArgs属性访问
参数校验	无，KeyError直接崩溃	基类execute统一Pydantic校验，返回友好错误
Prompt工具列表	手写硬编码，加工具必须改Prompt	to_prompt_description()自动生成，加工具=注册类
OpenAI Schema	无	to_openai_tool_schema()一键生成，为原生Function Call准备
循环驱动	for i in range固定次数，if分支隐式跳转	while state != FINISHED状态驱动，THINKING/TOOL_EXECUTING职责分离
状态枚举	RUNNING/FINISHED死代码	三态显式转移，pending变量跨状态传数据
死循环防护	仅靠Prompt规则	max_loop兜底 + executed_calls代码层门禁+纠偏消息
加新工具成本	改工具类+改Prompt+改Parser（潜在）	新建一个类+tool_list注册，零侵入

---

八、后续拓展

1. 接入OpenAI原生Function Call：本文已预留to_openai_tool_schema()，下一篇将LLM客户端切换到tools参数+tool_calls响应解析，彻底告别System Prompt里的工具描述文字，协议层约束替代文本约束；
2. Memory Role修正：工具结果目前仍用role=system，导致系统指令被工具数据污染，下一步改为role=tool+tool_call_id，对齐OpenAI标准消息协议；
3. 接入Chroma向量数据库：实现长期记忆RAG，跨会话代码检索；
4. 封装AgentEngine类：拆分日志、异常、指标、Hook模块，面向对象工程化重构。

九、Java/Go后端快速语法映射

• type[BaseModel]（类作为值传递）= Java Class<ReqDTO> / Go reflect.Type
• Pydantic model_validate = Jackson反序列化+@Valid校验 / Go json.Unmarshal+validator库
• @property 抽象属性 = 接口中定义getter方法
• 模板方法execute= 抽象类中固定流程方法+子类实现抽象钩子方法
• Enum状态 + while + continue状态转移 = State Pattern / 状态机引擎
• set[tuple[str, str]]去重 = Java HashSet<Pair<String,String>> / Go map[string]struct{}

---

下一篇：Java/Go后端手撸原生Agent（第四篇）