【Java/Go后端手撸原生Agent（第三篇）：Pydantic自动生成工具Schema + 三态状态机 + 文件读取工具】

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

前言

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

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

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

前置说明

一、问题复现与根因分析

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 设计思路（后端视角）

类比后端：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和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)

核心设计要点：

四、新增文件读取工具 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次，无重复、无死循环，最终自然语言回答整合了文件内容解释和数值计算结果。

七、核心改造总结

八、后续拓展

九、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 #golang #后端 #Agent #python #Pydantic #状态机