企业官网建设流程全解析

前言

用了不少 Agent，无论是 Claude Code 还是 Codex，用起来确实方便，但内置的各种 Skill、子 Agent 设置以及 MCP 协议使得代码不具可读性。想研究一下成熟的 Agent 是如何实现的，于是找了 Pi-Agent 作为学习对象。首先它开源，而且只保留了 Agent 核心循环必要的部分，其他组件由用户自行添加构建。我觉得相较于内置功能齐全的 Claude Code 等 Agent，Pi 这样的极简内核 + 扩展生态的架构方式更是未来趋势。

参考文献

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_key=os.getenv("OPENAI_API_KEY"), base_url="https://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, encoding="utf-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( model=MODEL, messages=self.messages, tool_choice="auto", tools=self.tools_schema, temperature=0.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_turns=5) 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.parameters

inspect.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 = None

2. 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_name='q', description='搜索关键词'), # DocstringParam(arg_name='limit', description='最多返回多少条'), # ]

原因2：模型幻觉——哪怕解决了漂移的问题，模型也可能输出错误的结果。

解决方法：以 schema 主导 + 签名兜底。模型按 schema 调；但在执行前用签名做一次防御性校验。

前置知识：

sig.bind():

bound = sig.bind(**clean) bound.apply_defaults()

sig.bind(参数)用和 Python 真正调用时一模一样的规则，检查"这组参数能不能合法绑定到这个签名上"，但不去真正调用函数。成功时返回一个BoundArguments 对象，其中将实际传入的参数绑定好。

def f(a, b=1): ... bound = sig.bind(a=10) # 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_key=os.getenv("OPENAI_API_KEY"), base_url="https://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, encoding="utf-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( model=MODEL, messages=self.messages, tool_choice="auto", tools=self.tools_schema, temperature=0.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_text：None → 解析成功，可以执行工具； 有内容 → 解析失败，把这个字符串当工具结果【回喂给模型】。（模型看到错误原因后通常会自己重试一次正确参数） """ # 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_reason=length），本轮整体作废，重试") 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_turns=5) agent.initiate() agent.run(user_input="先 list_files('.')，然后 read_file 读取当前目录下的 leetcode.py") agent.run(user_input="现在几点了")

从原本的170行到更新之后的430行，核心的 AgentLoop 很容易实现，但难的是如何保持代码的健壮性——让代码在出问题的情况下也能“优雅”处理。

Agent 就像和一个全知的幼儿园小朋友对话。一定要千叮咛万嘱咐各种边界，不然就很容易将任务以一种意想不到的方式进行。