前言

用了不少 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 都有前缀缓存。缓存是按前缀命中的,例如:

                       ...[system][user1][assistant1][user2][assistant2]...

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

Logo

更多推荐