Learn Pi-Agent (1)

前言用了不少 Agent无论是 Claude Code 还是 Codex用起来确实方便但内置的各种 Skill、子 Agent 设置以及 MCP 协议使得代码不具可读性。想研究一下成熟的 Agent 是如何实现的于是找了 Pi-Agent 作为学习对象。首先它开源而且只保留了 Agent 核心循环必要的部分其他组件由用户自行添加构建。我觉得相较于内置功能齐全的 Claude Code 等 AgentPi 这样的极简内核 扩展生态的架构方式更是未来趋势。参考文献Pi Agent 双轨教程 · 实战上手 源码学习Pi Agent 教程 | 菜鸟教程直接问“本人“”呗和 Pi 自己唠Pi Agent 源码其中有大量的例子和实例1.重温基础知识1.为何要将 assistant 存入 memory自己做了一下测试在某些情况下忽略 assistant 其实不会出现硬编码错误。例如memory { {role: user, content 给我三个整数”}, {role: assistant, content 2,4,6”}, {role: user, content 将第二个数乘以2返回给我”}, }真把 assistant 那条去掉其实也不会发生什么。模型会将两条 user 的消息合并起来然后返回结果。但在实际使用时仍要将 assistant 加入 memory 的意义在于1. tool_call 的硬编码错误当 assistant 返回结果包含 tool_call 时会出现硬编码错误。此时下一条 memory 中role tool其中 tool_call_id 需要与 assistant 中匹配。之前遇到过很多次了2. 语义层面上的静默失败不报错不抛异常但整个答案是建立在幻觉上的。例如a. 自我矛盾 —— 上一轮说建议 A 方案这条丢了下一轮可能推荐 B。而文档、代码已经按 A 走了前后不一致。b. 自我纠正丢失 —— 上一轮发现前面搞错了应该是 X。丢了这条下一轮可能又回到错误。c.安全边界失忆—— 上一轮拒绝了某个请求这个我不能做因为…。换个说法再问一遍模型不知道它拒绝过可能就做了。3. 反直觉的成本陷阱直觉认为不带历史应该成本更低但由于现代 API 都有前缀缓存。缓存是按前缀命中的例如将 assistant1 删除后面所有内容前缀全部变化缓存失效按照未命中缓存的原价计算价格。2. 关于 finish_reason模型返回 choice 格式大致如下在正常情况下若回复有 tool_call则 finish_reason “tool_calls”。但要注意 finish_reason length 的情况此时回复达到 max_tokens被截断。回复中的 tool_calls 可能残缺此时需要全部作废重新生成。伪代码如下if tool_calls: # ← 唯一入口 if finish_reason length: raise / 作废让模型重发 # ← 参数可能是残的 else: 执行3. 重写之前写过的 Agent 并完善import json from dotenv import load_dotenv from openai import OpenAI import os from typing import List, Dict, Callable load_dotenv() client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlhttps://api.deepseek.com/v1, ) MODEL deepseek-chat class Agent: def __init__(self, max_turns): self.max_turns max_turns self.messages [] self.tools_schema: List[Dict] [] self.tools: Dict[str, Callable] {} self.turn_count 0 def initiate(self): def get_time(): from datetime import datetime return datetime.now().isoformat() def list_files(dir: str): try: return \n.join(os.listdir(dir)) except Exception as e: return f错误: {e} def read_file(path: str): try: with open(path, encodingutf-8) as f: return f.read() except Exception as e: return f错误: {e} self.tools_schema [ { type: function, function: { name: list_files, description: 列出一个目录下的文件名, parameters: { type: object, properties: { dir: { type: string, description: 相对或绝对路径的目录, }, }, required: [dir] } } }, { type: function, function: { name: read_file, description: 读取一个文本文件的全部内容, parameters: { type: object, properties: { path: { type: string, description: 文件路径, }, }, required: [path] } } }, { type: function, function: { name: get_time, description: 获取当前时间, parameters: { type: object, properties: {}, required: [] } } } ] self.tools { list_files: list_files, read_file: read_file, get_time: get_time, } def _execute_tool(self, name: str, args: dict) - str: if name in self.tools: result self.tools[name](**args) else: result f❌ 未知工具: {name} return result def _think(self) - dict: response client.chat.completions.create( modelMODEL, messagesself.messages, tool_choiceauto, toolsself.tools_schema, temperature0.3, ) return response.choices[0].message.model_dump() def _act(self, assistant_msg: dict) - bool: tool_calls assistant_msg.get(tool_calls) if not tool_calls: return False self.messages.append(assistant_msg) for tool_call in tool_calls: print(tool_call[function][arguments]) tool_name tool_call[function][name] tool_call_id tool_call[id] tool_args json.loads(tool_call[function][arguments]) print(f 调用工具: {tool_name}) result self._execute_tool(tool_name, tool_args) print(f 结果: {result[:100]}...) self.messages.append({ role: tool, tool_call_id: tool_call_id, content: result, }) return True def run(self, user_input: str) - str: self.turn_count 0 print(f 用户: {user_input}) system_prompt 你是一个人工智能助手能够帮助人回答问题 self.messages [ {role: system, content: system_prompt}, {role: user, content: user_input} ] while self.turn_count self.max_turns: self.turn_count 1 print(f\n--- 第 {self.turn_count} 轮 ---) assistant_msg self._think() if assistant_msg.get(content): print(f 思考: {assistant_msg[content]}...) need_continue self._act(assistant_msg) if not need_continue: final_answer assistant_msg.get(content, ) self.messages.append({role: assistant, content: final_answer}) print(f\n✅ 最终回答:\n{final_answer}) return final_answer return ⚠️ 达到最大轮次 if __name__ __main__: agent Agent(max_turns5) agent.initiate() agent.run(user_input先 list_files(.)然后 read_file 读取当前目录下的 leetcode.py) agent.run(现在几点了)如上是之前自己写的简单 agent 指出的问题如下1. 没有考虑当 finish_reason length 时的截断此时 message 中的任何部分都可能截断受损应直接拒绝2. _act 函数中 json.loads(arguments) 没有任何保护首先要明白 json.loads() 的作用把一段JSON 文本解析成一个 Python 值其参数需求类型是 str/bytes/bytearray且内容能完整解析成一个 JSON 值。一个合法的 JSON 文本至少要有一个值。JSON 语法里值value的合法形态是value object | array | string | number | true | false | null模型返回的一定是 str可以通过类型检查。但以下几种情况都可能有不同的异常抛出。a. 空字符串可能抛出 JSONDecodeError。在 get_time() 这样的无参函数中不同的厂商、网关给的 arguments 不一致例如 {}、 空、 含一个空格。第一项能过而二三项不是合法JSON抛出JSONDecodeError。b. 返回值从 str 解析后可能不是 dict模型可能返回 []抛出 TypeError。tool_schema 并不能强制模型输出想要的结果只是建议不是代码上的 assert)。所以在实际运行过程中可能输出 [] 能够通过 json.loads()。但 **args 要求 tool_args 是 dict如果不是抛出 TypeError。3. self.tools[name](**args) 可能参数不匹配模型返回结果中可能多给、少给或者给错参数。给错参数可以由函数内部的 try / except 解决。少给参数不是很常见因为 tool_schema 中有 required 项。但 shema 非强制所以还是需要处理。问题原因1.参数定义有两份函数签名以及 schema 彼此之间可能漂移2.模型幻觉。原因1例如 list_files() 函数既有函数签名又有 schema。两者之间没有任何机制保持一致。解决方法函数签名为真相源。schema 不手写从函数签名自动生成。用 inspect.sigature 类型注解 docstring反射出 JSON Schema。前置知识1. inspect 库Python 标准库中的一个模块主要用于在运行时获取对象模块、类、方法、函数、代码对象等的内部信息。def search( query: str, limit: int 10, *, tags: Optional[list[str]] None, exact: bool False, ) - list[dict]: 搜索文章。 Args: query: 搜索关键词 limit: 最多返回多少条 tags: 按标签过滤 exact: 是否精确匹配 return [] sig inspect.signature(search) params sig.parametersinspect.signature(fn) 是 Python 自带的签名探测器。sig 有两个核心属性sig.parameters 和 sig.return_annotation用得少。params 为一个有序字典键是参数名值是 Parameter 对象。Parameter 对象可进一步调用 name、kind、annotation 以及 default 。for name, p in sig.parameters.items(): print(name) print( kind , p.kind) print( annotation, p.annotation) print( default , p.default) print() # 结果 query kind POSITIONAL_OR_KEYWORD annotation class str default class inspect._empty limit kind POSITIONAL_OR_KEYWORD annotation class int default 10 tags kind KEYWORD_ONLY annotation typing.Optional[list[str]] default None2. get_type_hints()from typing import get_type_hints hints get_type_hints(fn)作用是获取函数的类型注解。返回一个dict键是所有参数以及 return 值是已解析成真正类型对象的注解。同样以 search 函数为例 from typing import get_type_hints hints get_type_hints(search) hints { query: str, limit: int, tags: typing.Optional[list[str]], exact: bool, return: list[dict], }相较于 p.annotation其结果可能为字符串而不是真正的参数类型。hints 返回的是解析后的类型对象。3. parse(fn.__doc__ or )fn.__doc__是 Python 函数对象的__doc__ 属性保存了函数的docstring文档字符串。def search(q: str, limit: int 10): 搜索文章。 在数据库里按关键词检索。 Args: q: 搜索关键词 limit: 最多返回多少条 print(search.__doc__) # 打印结果就是 绿色文字所有 doc parse(search.__doc__ or ) # parse() 自动解析结构化字段 doc.short_description # 搜索文章。 doc.long_description # 在数据库里按关键词检索。 doc.params # [ # DocstringParam(arg_nameq, description搜索关键词), # DocstringParam(arg_namelimit, description最多返回多少条), # ]原因2模型幻觉——哪怕解决了漂移的问题模型也可能输出错误的结果。解决方法以 schema 主导 签名兜底。模型按 schema 调但在执行前用签名做一次防御性校验。前置知识sig.bind():bound sig.bind(**clean) bound.apply_defaults()sig.bind(参数)用和 Python 真正调用时一模一样的规则检查这组参数能不能合法绑定到这个签名上但不去真正调用函数。成功时返回一个BoundArguments 对象其中将实际传入的参数绑定好。def f(a, b1): ... bound sig.bind(a10) # bound.arguments {a: 10} ← 注意 b 还没出现bound.apply_defaults() 绑定好默认值。补完之后bound.args 和 bound.kwargs 就是可以直接喂给函数的完整参数。fn(*bound.args, **bound.kwargs) # 一次完美的、参数齐全的调用 # bound.args元组位置参数部分 # bound.kwargs字典关键字参数部分4. read_file() 函数中没有长度上限文件内容原封不动存进 messages下一轮请求容易上下文超限5. _think() 函数中model_dump() 把整个对象回传这会把 refusal、annotations、audio、function_call 这些字段一起塞回 messages。OpenAI/DeepSeek 通常忽略未知字段但严格的兼容网关会 400 报错。$.新代码import inspect import json from docstring_parser import parse from dotenv import load_dotenv from openai import OpenAI import os from typing import List, Dict, Callable, get_type_hints, get_origin, get_args, Union, Literal load_dotenv() client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlhttps://api.deepseek.com/v1, ) MODEL deepseek-chat PY_TO_JSON { str: string, int: integer, float: number, bool: boolean, list: array, dict: object, } MAX_CHARS 20000 class Agent: def __init__(self, max_turns): self.max_turns max_turns self.messages [] self.tools_schema: List[Dict] [] self.tools: Dict[str, Callable] {} self.turn_count 0 # 问题 3 原因 1 Part 1 def annotation_to_schema(self, ann): if ann is inspect.Parameter.empty: return {type: string} if ann in PY_TO_JSON: return {type: PY_TO_JSON[ann]} origin get_origin(ann) args get_args(ann) # Optional[str] Union[str, None] if origin is Union: non_none [a for a in args if a is not type(None)] if len(non_none) 1: s self.annotation_to_schema(non_none[0]) return {type: [s[type], null] if type in s else s} # 复杂情况在此简化 if origin is list: return {type: array, items: self.annotation_to_schema(args[0])} if origin is Literal: return {enum: list(args)} return {type: string} # 兜底 # 问题 3 原因 1 Part 2 def build_schema_from_func(self, fn: Callable) - dict: sig inspect.signature(fn) hints get_type_hints(fn) doc parse(fn.__doc__ or ) param_docs {p.arg_name: p.description for p in doc.params} properties {} required [] for name, p in sig.parameters.items(): # 只能接受按关键字传的参数和 _execute_tool 函数白名单一致 if p.kind not in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY): continue ann hints.get(name, inspect.Parameter.empty) # 获得解析后的参数类型 prop self.annotation_to_schema(ann) # 将参数类型转为 JSON Schema 类 if name in param_docs and param_docs[name]: prop[description] param_docs[name] if p.default is not inspect.Parameter.empty: prop[default] p.default else: required.append(name) properties[name] prop return { type: function, function: { name: fn.__name__, description: (doc.short_description or ).strip(), parameters: { type: object, properties: properties, required: required, }, }, } def initiate(self): def get_time(): 获取当前时间 from datetime import datetime return datetime.now().isoformat() def list_files(dir: str): 列出一个目录下的文件名 Args: dir: 相对或绝对路径的目录 try: return \n.join(os.listdir(dir)) except Exception as e: return f错误: {e} def read_file(path: str): 读取一个文本文件的全部内容 Args: path: 文件路径 # 问题 4 修复 try: with open(path, encodingutf-8) as f: content f.read() except Exception as e: return f错误: {e} if len(content) MAX_CHARS: return content[:MAX_CHARS] f\n\n[已截断原文共 {len(content)} 字符只显示前 {MAX_CHARS}] return content # 问题 3 原因 1 Part 3 删除手动写的 tools_schema # self.tools_schema [ # { # type: function, # function: { # name: list_files, # description: 列出一个目录下的文件名, # parameters: { # type: object, # properties: { # dir: { # type: string, # description: 相对或绝对路径的目录, # }, # }, # required: [dir] # } # } # }, # { # type: function, # function: { # name: read_file, # description: 读取一个文本文件的全部内容, # parameters: { # type: object, # properties: { # path: { # type: string, # description: 文件路径, # }, # }, # required: [path] # } # } # }, # { # type: function, # function: { # name: get_time, # description: 获取当前时间, # parameters: { # type: object, # properties: {}, # required: [] # } # } # } # ] self.tools { list_files: list_files, read_file: read_file, get_time: get_time, } # 问题 3 原因 1 Part 4 tools_schema 由函数签名生成 self.tools_schema [self.build_schema_from_func(f) for f in self.tools.values()] # 问题 3 原因 2 Part 1 def _execute_tool(self, name: str, args: dict) - str: 函数执行带安全检查 :param name: 模型输出的调用函数的名字 :param args: 模型输出的调用函数的参数可能出现问题 :return: 调用函数结果 fn self.tools.get(name) if fn is None: return f❌ 未知工具{name}。可用工具{, .join(self.tools)} sig inspect.signature(fn) params sig.parameters # 显式拒绝 Agent 不支持的参数类型POSITIONAL_ONLY 和 VAR_POSITIONAL for n, p in params.items(): if p.kind in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.VAR_POSITIONAL): return (f❌ 工具 {name} 的参数 {n} 类型为 {p.kind.description} f当前框架不支持按 JSON 关键字传参请修改工具签名。) # 有 **kwargs任何参数都合法 has_var_kw False for p in params.values(): if p.kind is inspect.Parameter.VAR_KEYWORD: has_var_kw True break # 算出这个工具接受哪些参数名存进 accepted 集合 # 只考虑具名参数不管 **kwargs accepted { n for n, p in sig.parameters.items() if p.kind in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY) } # 多给的参数丢弃但要记录 ignored sorted(set(args) - accepted) if not has_var_kw else [] # 除去多给的参数只保留工具认得的键值对 if has_var_kw: clean dict(args) else: clean {k: v for k, v in args.items() if k in accepted} # 缺少的必须参数 # 遍历所有参数找出没有默认值、能按关键字传、且模型没给的参数名收进一个列表 missing [] for n, p in sig.parameters.items(): if p.default is inspect.Parameter.empty: if p.kind in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY): if n not in clean: missing.append(n) if missing: hint f。你多给了 {ignored}已忽略 if ignored else return ( f❌ 缺少必填参数{missing}。 f本工具接受参数{sorted(accepted)}{hint}。请重新调用。 ) # 执行 # try: # result fn(**clean) # except Exception as e: # return f❌ 工具执行失败{type(e).__name__}: {e} # fn(**clean)抛出的 TypeError 有两个来源 # 1. 参数仍然错误 # 2. 参数正确函数内部有错。例如出现了 1 “2” 的情况 # Exception 无法捕捉两者区别 try: bound sig.bind(**clean) bound.apply_defaults() except TypeError as e: return f❌ 参数与签名不匹配{e}。本工具接受参数{sorted(accepted)} try: result fn(*bound.args, **bound.kwargs) except Exception as e: return f❌ 工具内部执行失败非参数问题请勿重试参数{type(e).__name__}: {e} # 如果有被忽略的模型要告诉模型不然它不知道自己给错了 note f[已忽略未知参数{ignored}]\n if ignored else return note str(result) # 问题 1 修复 Part 1 增加 finish_reason # 问题 5 修复 修改函数输出 def _think(self) - tuple: response client.chat.completions.create( modelMODEL, messagesself.messages, tool_choiceauto, toolsself.tools_schema, temperature0.3, ) choice response.choices[0] # 原 return response.choices[0].message.model_dump() # 问题 1 修复return choice.message.model_dump(), choice.finish_reason # 问题 5 修复 msg choice.message tool_calls None if msg.tool_calls: tool_calls [ { id: tc.id, type: function, function: { name: tc.function.name, arguments: tc.function.arguments, }, } for tc in msg.tool_calls ] assistant_msg { role: assistant, content: msg.content, tool_calls: tool_calls, } # 边界content 和 tool_calls 不能同时为 None否则回传 API 会 400 if assistant_msg[content] is None and assistant_msg[tool_calls] is None: raise ValueError( f模型返回了既无 content 也无 tool_calls 的消息finish_reason{choice.finish_reason} ) return assistant_msg, choice.finish_reason # 问题 2 修复 Part 1 : 安全解析 arguments def _parse_tool_args(self, raw: str): :param raw: 待处理参数 :return: tool_args error_textNone → 解析成功可以执行工具 有内容 → 解析失败把这个字符串当工具结果【回喂给模型】。模型看到错误原因后通常会自己重试一次正确参数 # 1. None, 空字符串纯空白 - 当成无参数而不是崩溃 if not raw or not raw.strip(): return {}, None # 2. 不是合法JSON try: parsed json.loads(raw) except json.JSONDecodeError as e: return {}, ( f参数不是合法 JSON{e.msg}位置 {e.pos}。 f请只输出一个 JSON 对象例如 {{dir: .}} f不要使用单引号、注释、尾逗号或裸换行。 f你收到的是{str(raw)[:200]} ) # 3. 是合法 JSON但不是 object if not isinstance(parsed, dict): return {}, ( f参数必须是 JSON 对象object收到的是 {type(parsed).__name__}。 f请输出形如 {{dir: .}} 的对象。你收到的是{str(raw)[:200]} ) return parsed, None def _act(self, assistant_msg: dict) - bool: tool_calls assistant_msg.get(tool_calls) if not tool_calls: return False self.messages.append(assistant_msg) for tool_call in tool_calls: print(tool_call[function][arguments]) tool_name tool_call[function][name] tool_call_id tool_call[id] # tool_args json.loads(tool_call[function][arguments]) # 问题 2 修复 Part 2 安全解析并将错误喂给模型 raw_args tool_call[function].get(arguments) tool_args, error_text self._parse_tool_args(raw_args) if error_text: # 解析失败将错误原因当做工具结果喂给模型 print(f ⚠️ 参数解析失败: {error_text[:80]}...) result f❌ {error_text} else: print(f 调用工具: {tool_name} args{tool_args}) result self._execute_tool(tool_name, tool_args) print(f 结果: {result}) self.messages.append({ role: tool, tool_call_id: tool_call_id, content: result, }) return True def run(self, user_input: str) - str: self.turn_count 0 print(f 用户: {user_input}) system_prompt 你是一个人工智能助手能够帮助人回答问题 self.messages [ {role: system, content: system_prompt}, {role: user, content: user_input} ] while self.turn_count self.max_turns: self.turn_count 1 print(f\n--- 第 {self.turn_count} 轮 ---) # 问题 2 修复 Part 3 _think() 返回二元组 # assistant_msg self._think() assistant_msg, finish_reason self._think() # 问题 1 修复 if finish_reason length: print( ⚠️ 输出被截断finish_reasonlength本轮整体作废重试) continue if assistant_msg.get(content): print(f 思考: {assistant_msg[content]}...) need_continue self._act(assistant_msg) if not need_continue: final_answer assistant_msg.get(content, ) self.messages.append({role: assistant, content: final_answer}) print(f\n✅ 最终回答:\n{final_answer}) return final_answer return ⚠️ 达到最大轮次 if __name__ __main__: agent Agent(max_turns5) agent.initiate() agent.run(user_input先 list_files(.)然后 read_file 读取当前目录下的 leetcode.py) agent.run(user_input现在几点了)从原本的170行到更新之后的430行核心的 AgentLoop 很容易实现但难的是如何保持代码的健壮性——让代码在出问题的情况下也能“优雅”处理。Agent 就像和一个全知的幼儿园小朋友对话。一定要千叮咛万嘱咐各种边界不然就很容易将任务以一种意想不到的方式进行。