从 0 到 1 搭建你的第一个 AI Agent Harness Engineering：实践路线图

---

1. 标题 (Title)

• 从 0 到 1：构建你的首个 AI Agent 系统完整实践指南
• AI Agent 实战攻略：手把手教你驾驭工程化开发流程
• 不再迷茫！AI Agent Harness Engineering 入门到精通路线图
• 工程化视角：如何系统性地构建可扩展的 AI Agent 架构
• AI Agent 开发新范式：Harness Engineering 实践路线图详解

---

2. 引言 (Introduction)

痛点引入 (Hook)

你是否曾经被那些令人惊叹的 AI 演示所吸引？从能够自主完成复杂任务的 AI 助手，到可以进行多步推理解决问题的智能系统，AI Agent 的能力正在以惊人的速度发展。但是，当你满怀期待地想要自己动手构建一个 AI Agent 时，却发现自己面对的是一堆零散的概念、复杂的工具链和缺乏系统性指导的困境。

你可能尝试过一些快速开始的教程，复制粘贴了一些代码，确实让一个简单的 Agent 跑了起来。但是，当你想要扩展功能、优化性能、或者让它真正解决实际问题时，却发现之前的代码变得难以维护，各种问题层出不穷。你开始问自己：真正生产级别的 AI Agent 系统到底是怎么构建的？有没有一套系统的方法论可以遵循？

文章内容概述 (What)

本文正是为了解决你的这些困惑而写。我们将从一个全新的视角——Harness Engineering（驾驭工程）——来探讨 AI Agent 的开发。这不仅仅是写几行代码调用大语言模型那么简单，而是一套涵盖架构设计、开发流程、测试部署、监控优化的完整工程化实践。

在这篇文章中，我们将一起：

1. 建立对 AI Agent 和 Harness Engineering 的系统性认知
2. 设计一个清晰、可扩展的 Agent 架构
3. 一步步实现核心功能模块
4. 学习如何测试、调试和优化你的 Agent
5. 探讨如何将 Agent 部署到生产环境

读者收益 (Why)

读完本文，你将不仅仅是“跑通了一个 Demo”，而是能够：

• 从工程化的角度理解 AI Agent 系统的构成
• 掌握一套可复用的 AI Agent 开发方法论
• 构建出更加健壮、可扩展、易维护的 Agent 系统
• 为进一步深入学习和实践打下坚实的基础

无论你是 AI 领域的新手，还是有一定经验但希望系统化提升的开发者，这篇文章都将为你提供有价值的参考。

---

3. 准备工作 (Prerequisites)

在开始我们的实践之旅前，让我们确保你已经准备好了必要的知识和工具。

技术栈/知识：

1. Python 编程基础： 我们将使用 Python 作为主要开发语言，你需要熟悉 Python 的基本语法、面向对象编程概念，以及常用的标准库。
2. 基础的 Linux/命令行操作： 虽然不是必须，但熟悉基本的命令行操作会让你的开发过程更加顺畅。
3. 对大语言模型 (LLM) 的基本了解： 你不需要深入理解 LLM 的内部原理，但应该知道它们是什么，以及大致是如何工作的。
4. API 调用经验： 大多数 Agent 系统都会涉及调用各种 API（包括 LLM API），了解 RESTful API 的基本概念和调用方式会很有帮助。

环境/工具：

1. Python 3.8+： 确保你安装了较新版本的 Python。
2. 一个代码编辑器： 如 VS Code、PyCharm 等，选择你习惯的即可。
3. OpenAI API Key（或其他 LLM 提供商的 API Key）： 我们将使用 LLM 作为 Agent 的核心“大脑”，你需要准备一个可用的 API Key。
4. Git（可选但推荐）： 用于版本控制。

准备好这些，我们就可以开始我们的 AI Agent Harness Engineering 之旅了！

---

4. 核心内容：手把手实战 (Step-by-Step Tutorial)

概念铺垫：什么是 AI Agent Harness Engineering？

在开始写代码之前，让我们先花点时间理解一些核心概念。

什么是 AI Agent？

简单来说，AI Agent 是一个能够感知环境、做出决策并执行行动的智能系统。一个典型的 Agent 循环包括以下几个步骤：

1. 观察 (Observation)： 收集当前环境的信息。
2. 推理 (Reasoning)： 基于观察到的信息和已有的知识，思考下一步该做什么。
3. 行动 (Action)： 执行决策，如调用工具、输出文本等。
4. 反馈 (Feedback)： 获取行动的结果，并将其作为下一轮观察的输入。

什么是 Harness Engineering？

“Harness” 这个词有“驾驭、利用”的意思。在 AI Agent 的语境下，**Harness Engineering（驾驭工程）**指的是一套系统化的工程方法论，用于：

• 有效地“驾驭” LLM 的能力： 将 LLM 的强大能力转化为可控、可靠的功能。
• 构建完整的 Agent 系统： 不仅仅是调用 LLM，还包括工具集成、状态管理、错误处理等。
• 确保系统的可观测性和可维护性： 让 Agent 的行为可追踪、可调试、可优化。

如果说 LLM 是一匹强大的千里马，那么 Harness Engineering 就是教会你如何驯马、配鞍、驾驭，让它能够带你到达目的地。

---

步骤一：架构设计——搭建 Agent 的“骨架”

在开始写代码之前，先花时间设计架构是非常重要的。一个好的架构能让我们的系统更加清晰、易于扩展。

核心概念：Agent 的模块化构成

我们可以将一个 AI Agent 系统拆解为以下几个核心模块：

1. 核心控制器 (Core Controller)： Agent 的“大脑”，负责协调整个系统的运行，维护 Agent 的状态。
2. 大语言模型接口 (LLM Wrapper)： 负责与 LLM 交互，封装 API 调用细节，处理重试、错误等。
3. 工具管理 (Tool Management)： 管理 Agent 可以使用的工具（如搜索、计算、代码执行等），负责工具的注册、调用。
4. 提示词管理 (Prompt Management)： 管理系统提示词、动态构建提示词模板。
5. 记忆系统 (Memory System)： 负责存储和检索对话历史、重要信息，让 Agent 有“记忆”。
6. 输出解析 (Output Parsing)： 解析 LLM 的输出，将其转化为结构化的数据或指令。

概念结构与核心要素组成

为了让大家更直观地理解，我们可以用一个架构图来表示这些模块之间的关系：

在这个架构中，核心控制器是整个系统的中心，它协调各个模块的工作。当收到用户请求后，它会从记忆系统读取相关历史，从提示词管理模块获取提示词模板，构建完整的提示词，然后通过 LLM 接口发送给大语言模型。收到模型输出后，通过输出解析模块将其结构化，然后根据解析结果决定是直接返回给用户，还是调用相应的工具。工具执行的结果会再次被送入这个循环，直到任务完成。

---

步骤二：项目初始化——打下坚实的基础

现在我们对架构有了概念，接下来让我们开始实际的项目搭建。

环境安装

首先，创建一个新的项目目录，并初始化虚拟环境（这是一个好习惯，可以避免依赖冲突）：

mkdir my-first-agent
cd my-first-agent
python -m venv venv
source venv/bin/activate  # Windows 上使用 venv\Scripts\activate

接下来，我们安装一些必要的依赖库：

pip install openai python-dotenv pydantic

• openai: OpenAI 的官方 Python SDK，用于调用 GPT 模型。
• python-dotenv: 用于管理环境变量，避免把 API Key 等敏感信息直接写在代码里。
• pydantic: 一个强大的数据验证库，可以帮助我们确保数据格式的正确性。

创建一个 .env 文件，用于存放你的 API Key：

# .env
OPENAI_API_KEY=sk-your-actual-api-key-here

确保在你的 .gitignore 文件中加入了 .env，以免不小心将敏感信息提交到版本控制系统。

项目结构设计

一个好的项目结构能让代码更易于组织和维护。我们将按照模块化的思想来组织代码：

my-first-agent/
├── .env
├── .gitignore
├── main.py                # 程序入口
├── requirements.txt       # 依赖列表
├── agent/                 # Agent 核心代码
│   ├── __init__.py
│   ├── core.py            # 核心控制器
│   ├── llm.py             # LLM 接口封装
│   ├── memory.py          # 记忆系统
│   ├── prompts.py         # 提示词管理
│   ├── tools.py           # 工具管理
│   └── parser.py          # 输出解析
└── examples/              # 示例代码
    └── simple_chat.py

现在，让我们开始逐步实现这些模块。

---

步骤三：实现 LLM 接口封装——与“大脑”对话

LLM 是 Agent 的核心，我们首先要实现一个健壮的 LLM 接口封装模块。这个模块的职责是：

1. 加载配置和 API Key。
2. 封装 API 调用逻辑。
3. 处理常见的错误（如网络错误、限流、超时等）。
4. 提供简洁的接口供上层调用。

创建 agent/llm.py：

# agent/llm.py
import os
import openai
from dotenv import load_dotenv
from typing import List, Dict, Any, Optional

# 加载环境变量
load_dotenv()

class LLMClient:
    """
    大语言模型客户端封装类
    负责处理与 LLM 的所有交互
    """
    def __init__(self, model: str = "gpt-3.5-turbo", temperature: float = 0.7):
        """
        初始化 LLM 客户端
        
        Args:
            model: 使用的模型名称
            temperature: 温度参数，控制输出的随机性 (0-2)
        """
        self.api_key = os.getenv("OPENAI_API_KEY")
        if not self.api_key:
            raise ValueError("未找到 OPENAI_API_KEY 环境变量，请在 .env 文件中设置")
        
        openai.api_key = self.api_key
        self.model = model
        self.temperature = temperature
        
    def chat_completion(
        self, 
        messages: List[Dict[str, str]], 
        **kwargs
    ) -> str:
        """
        发送聊天完成请求
        
        Args:
            messages: 消息列表，格式为 [{"role": "user", "content": "..."}, ...]
            **kwargs: 其他传递给 OpenAI API 的参数
            
        Returns:
            LLM 的响应内容字符串
        """
        try:
            # 合并默认参数和传入参数
            params = {
                "model": self.model,
                "messages": messages,
                "temperature": self.temperature,
                **kwargs
            }
            
            response = openai.ChatCompletion.create(**params)
            return response.choices[0].message.content.strip()
            
        except openai.error.AuthenticationError:
            raise Exception("API Key 无效，请检查你的 OPENAI_API_KEY")
        except openai.error.RateLimitError:
            raise Exception("达到 API 调用频率限制，请稍后再试")
        except openai.error.APIConnectionError:
            raise Exception("连接 OpenAI API 失败，请检查网络连接")
        except openai.error.OpenAIError as e:
            raise Exception(f"OpenAI API 错误: {str(e)}")
        except Exception as e:
            raise Exception(f"调用 LLM 时发生未知错误: {str(e)}")

代码解析：

1. 配置加载： 我们使用 python-dotenv 从 .env 文件中加载 API Key，这样做比硬编码在代码里更安全、更灵活。
2. 类封装： 我们将 LLM 相关的功能封装在 LLMClient 类中，这样可以方便地管理配置（如模型类型、温度参数）。
3. 错误处理： 我们捕获了几种常见的 OpenAI API 错误，并给出了更友好的错误提示。这对于系统的健壮性非常重要。

现在，让我们创建一个简单的测试脚本来验证这个模块是否工作正常。创建 examples/simple_chat.py：

# examples/simple_chat.py
import sys
import os

# 将项目根目录添加到 Python 路径，以便能够导入 agent 模块
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from agent.llm import LLMClient

def main():
    print("正在初始化 LLM 客户端...")
    llm = LLMClient(temperature=0.7)
    
    print("LLM 客户端初始化成功！")
    print("------------------------")
    
    messages = [
        {"role": "system", "content": "你是一个友好的 AI 助手。"},
        {"role": "user", "content": "你好，请简单介绍一下你自己。"}
    ]
    
    print(f"用户: {messages[-1]['content']}")
    print("AI 正在思考...")
    
    try:
        response = llm.chat_completion(messages)
        print(f"AI: {response}")
    except Exception as e:
        print(f"发生错误: {e}")

if __name__ == "__main__":
    main()

你可以运行这个脚本测试一下：

python examples/simple_chat.py

如果一切顺利，你应该能看到 AI 的回复了！恭喜你，我们已经完成了第一步，建立了与 LLM 的连接。

---

步骤四：实现记忆系统——让 Agent 拥有“记忆”

如果没有记忆，Agent 每次对话都像是第一次见面，无法理解上下文。一个好的记忆系统对于 Agent 的对话连贯性至关重要。

让我们在 agent/memory.py 中实现一个简单但有效的记忆系统：

# agent/memory.py
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class MemoryItem:
    """
    表示一条记忆条目
    """
    role: str  # "user", "assistant", "system", "tool"
    content: str
    timestamp: datetime = field(default_factory=datetime.now)
    metadata: Dict[str, Any] = field(default_factory=dict)

class MemoryManager:
    """
    记忆管理器
    负责存储、检索和管理对话历史
    """
    def __init__(self, max_history_length: Optional[int] = None):
        """
        初始化记忆管理器
        
        Args:
            max_history_length: 最大记忆长度，超过这个长度会自动截断旧的记忆
                               如果为 None，则不限制长度
        """
        self.memories: List[MemoryItem] = []
        self.max_history_length = max_history_length
        
    def add(self, role: str, content: str, metadata: Optional[Dict[str, Any]] = None):
        """
        添加一条记忆
        
        Args:
            role: 角色
            content: 内容
            metadata: 可选的元数据
        """
        item = MemoryItem(
            role=role,
            content=content,
            metadata=metadata or {}
        )
        self.memories.append(item)
        
        # 如果超过最大长度，移除最旧的记忆
        if self.max_history_length and len(self.memories) > self.max_history_length:
            # 通常我们希望保留 system prompt，所以这里只截断非 system 的记忆
            # 这是一个简化的处理方式
            non_system_memories = [m for m in self.memories if m.role != "system"]
            system_memories = [m for m in self.memories if m.role == "system"]
            
            if len(non_system_memories) > self.max_history_length:
                excess = len(non_system_memories) - self.max_history_length
                self.memories = system_memories + non_system_memories[excess:]
    
    def get_recent(self, limit: Optional[int] = None) -> List[MemoryItem]:
        """
        获取最近的记忆
        
        Args:
            limit: 获取的数量限制
            
        Returns:
            记忆条目列表
        """
        if limit is None:
            return self.memories.copy()
        return self.memories[-limit:]
    
    def to_llm_messages(self) -> List[Dict[str, str]]:
        """
        将记忆转换为 LLM 可识别的消息格式
        
        Returns:
            消息列表
        """
        return [
            {"role": m.role, "content": m.content}
            for m in self.memories
        ]
    
    def clear(self):
        """清空所有记忆"""
        self.memories = []
        
    def __len__(self):
        return len(self.memories)

核心概念解析：

这里我们实现了一个基于列表的简单记忆系统。核心概念包括：

1. 记忆条目 (MemoryItem)： 使用 dataclass 定义，不仅包含角色和内容，还包含时间戳和元数据，为未来的扩展（如按时间检索、重要性标记）留出了空间。
2. 自动截断： 为了避免 Token 消耗过快，我们设置了 max_history_length，当记忆超过这个长度时，会自动丢弃最旧的记忆（保留 system prompt）。
3. 格式转换： 提供了 to_llm_messages() 方法，方便地将记忆转换为 OpenAI API 需要的格式。

现在，让我们更新一下测试脚本，看看加入记忆系统后的效果：

# examples/chat_with_memory.py
import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from agent.llm import LLMClient
from agent.memory import MemoryManager

def main():
    print("正在初始化 Agent 组件...")
    llm = LLMClient(temperature=0.7)
    memory = MemoryManager(max_history_length=10)
    
    # 添加系统提示词
    memory.add("system", "你是一个友好的 AI 助手，能够记住我们之前的对话。")
    
    print("初始化完成！来和我聊天吧（输入 'quit' 退出）")
    print("-" * 50)
    
    while True:
        user_input = input("你: ")
        
        if user_input.lower() in ['quit', 'exit', 'q']:
            print("再见！")
            break
            
        memory.add("user", user_input)
        
        print("AI: ", end="", flush=True)
        try:
            response = llm.chat_completion(memory.to_llm_messages())
            print(response)
            memory.add("assistant", response)
        except Exception as e:
            print(f"[错误: {e}]")

if __name__ == "__main__":
    main()

运行这个脚本，你可以进行多轮对话，Agent 现在能够“记住”之前的对话内容了！

---

步骤五：提示词管理——与 Agent 高效沟通

提示词 (Prompt) 是我们引导 LLM 行为的关键。随着系统变得复杂，我们的提示词也会变得越来越长、越来越复杂。如果把提示词直接硬编码在逻辑里，会导致代码难以维护。因此，我们需要一个专门的提示词管理模块。

创建 agent/prompts.py：

# agent/prompts.py
from typing import Dict, Any, Optional
from string import Template

class PromptTemplate:
    """
    提示词模板类
    支持变量替换
    """
    def __init__(self, template: str):
        self.template = template
        self._template_obj = Template(template)
        
    def format(self, **kwargs) -> str:
        """
        格式化提示词模板，替换变量
        
        Args:
            **kwargs: 模板变量的值
            
        Returns:
            格式化后的提示词
        """
        try:
            return self._template_obj.safe_substitute(**kwargs)
        except KeyError as e:
            raise ValueError(f"提示词模板缺少必要的变量: {e}")

class PromptManager:
    """
    提示词管理器
    用于管理和构建各种提示词
    """
    
    # 基础系统提示词
    BASE_SYSTEM_PROMPT = PromptTemplate("""
你是一个先进的 AI 助手，叫做 $agent_name。

你的目标是尽最大努力帮助用户完成任务。

关于你的能力：
- 你可以进行多轮对话
- 你会尽力理解用户的意图
- 如果你不知道答案，你会诚实地说出来，而不是编造信息

请始终保持友好和专业的态度。
""".strip())
    
    # 带有工具使用能力的系统提示词
    AGENT_WITH_TOOLS_PROMPT = PromptTemplate("""
你是一个先进的 AI 助手，叫做 $agent_name。

你可以使用以下工具来帮助完成任务：

$tools_description

使用工具的规则：
1.  当你需要使用工具时，请严格按照以下 JSON 格式输出：
    {
        "thinking": "你对当前问题的思考过程",
        "action": {
            "tool_name": "要使用的工具名称",
            "tool_input": "工具的输入参数"
        }
    }

2.  如果你认为不需要使用工具，可以直接回答用户，请使用以下 JSON 格式：
    {
        "thinking": "你对当前问题的思考过程",
        "action": {
            "tool_name": "final_answer",
            "tool_input": "你对用户的最终回答"
        }
    }

重要提示：
- 你的所有输出都必须是有效的 JSON 格式，不要包含其他额外的文本。
- "thinking" 字段用于记录你的思考过程，这有助于用户理解你的逻辑。
- 只有当你获取了足够的信息后，才使用 "final_answer" 工具。
""".strip())

    @classmethod
    def build_system_prompt(
        cls, 
        agent_name: str = "AI Assistant",
        tools_description: Optional[str] = None,
        custom_instructions: Optional[str] = None
    ) -> str:
        """
        构建系统提示词
        
        Args:
            agent_name: Agent 的名称
            tools_description: 工具描述（如果有工具的话）
            custom_instructions: 自定义指令（可选）
            
        Returns:
            完整的系统提示词
        """
        if tools_description:
            prompt = cls.AGENT_WITH_TOOLS_PROMPT.format(
                agent_name=agent_name,
                tools_description=tools_description
            )
        else:
            prompt = cls.BASE_SYSTEM_PROMPT.format(
                agent_name=agent_name)
                
        if custom_instructions:
            prompt += f"\n\n额外指令：\n{custom_instructions}"
            
        return prompt

设计思路：

1. 模板分离： 我们将不同场景下的提示词（基础对话、带工具使用）定义为不同的模板。
2. 变量注入： 使用 string.Template 来支持变量替换，比如 Agent 名称、工具描述等。
3. 构建方法： 提供了 build_system_prompt 这个便捷方法，根据参数动态选择和组合提示词。

这里我还想强调一下提示词工程中的一个重要概念：结构化输出。在上面的 AGENT_WITH_TOOLS_PROMPT 中，我们明确要求 LLM 输出 JSON 格式。这是 Agent 开发中的一个关键技术，它让我们能够可靠地解析 LLM 的输出，进而执行工具调用等操作。

---

步骤六：工具系统——赋予 Agent“动手”能力

LLM 虽然知识渊博，但也有其局限性：它无法获取实时信息（如今天的天气、最新的新闻），无法进行复杂的数学计算，无法直接操作外部系统。工具系统就是为了解决这些问题而存在的，它赋予了 Agent 与外部世界交互的能力。

让我们创建 agent/tools.py：

# agent/tools.py
from typing import Dict, Any, List, Callable, Optional
from dataclasses import dataclass, field
import json

@dataclass
class Tool:
    """
    工具定义类
    """
    name: str  # 工具名称
    description: str  # 工具描述，告诉 LLM 这个工具是做什么的
    func: Callable  # 实际执行的函数
    input_schema: Optional[Dict[str, Any]] = None  # 输入参数的 JSON Schema（可选）
    
    def __str__(self):
        return f"{self.name}: {self.description}"

class ToolRegistry:
    """
    工具注册表
    负责管理所有可用的工具
    """
    def __init__(self):
        self._tools: Dict[str, Tool] = {}
        
    def register_tool(self, tool: Tool):
        """
        注册一个工具
        """
        if tool.name in self._tools:
            print(f"警告: 工具 '{tool.name}' 已存在，将被覆盖")
        self._tools[tool.name] = tool
        
    def get_tool(self, name: str) -> Optional[Tool]:
        """
        获取工具
        """
        return self._tools.get(name)
        
    def list_tools(self) -> List[Tool]:
        """
        列出所有工具
        """
        return list(self._tools.values())
        
    def get_tools_description(self) -> str:
        """
        生成给 LLM 看的工具描述文本
        """
        if not self._tools:
            return "当前没有可用的工具。"
            
        descriptions = []
        for tool in self._tools.values():
            desc = f"- {tool.name}: {tool.description}"
            if tool.input_schema:
                # 如果有 schema，可以把参数信息也加进去
                desc += f"\n  参数: {json.dumps(tool.input_schema, ensure_ascii=False)}"
            descriptions.append(desc)
            
        return "\n".join(descriptions)
        
    def execute_tool(self, tool_name: str, tool_input: str) -> str:
        """
        执行工具
        
        Args:
            tool_name: 工具名称
            tool_input: 工具输入
            
        Returns:
            工具执行结果
        """
        tool = self.get_tool(tool_name)
        if not tool:
            return f"错误: 找不到工具 '{tool_name}'"
            
        try:
            # 这里我们做一个简化处理，直接把输入传给函数
            # 在实际生产中，你可能需要根据 input_schema 对输入进行验证和解析
            result = tool.func(tool_input)
            return str(result)
        except Exception as e:
            return f"执行工具 '{tool_name}' 时出错: {str(e)}"

# ==============================================================================
# 示例工具定义
# ==============================================================================

def create_sample_tools() -> ToolRegistry:
    """
    创建一些示例工具，用于演示
    """
    registry = ToolRegistry()
    
    # 1. 简单的计算器工具
    def calculator(input_expr: str) -> str:
        """一个简单的计算器，支持基本的数学运算"""
        try:
            # 注意：在生产环境中使用 eval 要非常小心！这里仅作演示
            result = eval(input_expr)
            return f"计算结果: {result}"
        except Exception as e:
            return f"计算错误: {str(e)}"
            
    registry.register_tool(Tool(
        name="calculator",
        description="一个简单的计算器，用于执行数学计算，输入是数学表达式字符串，如 '2 + 2'",
        func=calculator
    ))
    
    # 2. 获取当前时间的工具
    def get_current_time(_: str) -> str:
        """获取当前时间"""
        from datetime import datetime
        return f"当前时间是: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
        
    registry.register_tool(Tool(
        name="get_current_time",
        description="获取当前的日期和时间，不需要输入参数",
        func=get_current_time
    ))
    
    return registry

核心概念解析：

1. 工具抽象 (Tool)： 我们将工具抽象为包含名称、描述、执行函数等属性的数据类。
2. 注册表模式 (ToolRegistry)： 使用注册表模式来管理工具，这使得工具的添加、删除和查找变得非常方便。
3. 工具描述生成： get_tools_description() 方法会自动将所有注册的工具格式化为自然语言描述，这样我们就可以把它动态地插入到提示词中，告诉 LLM 有哪些工具可用。

在示例中，我实现了两个简单的工具：计算器和获取当前时间。这只是抛砖引玉，在实际应用中，你可以注册任何你需要的工具，比如：

• 网络搜索工具 (SerpAPI, Google Search API)
• 代码执行工具 (如 Python REPL)
• 数据库查询工具
• 特定业务 API 调用工具

---

步骤七：输出解析——理解 LLM 的“想法”

前面我们提到了结构化输出的重要性。现在，我们需要一个模块来专门负责解析 LLM 的输出，将其转化为我们的程序可以处理的数据结构。

创建 agent/parser.py：

# agent/parser.py
import json
from typing import Dict, Any, Optional, Tuple
from dataclasses import dataclass

@dataclass
class AgentAction:
    """
    表示 Agent 的一个行动决策
    """
    thinking: str  # Agent 的思考过程
    tool_name: str  # 要使用的工具名称
    tool_input: str  # 工具的输入
    
    @property
    def is_final_answer(self) -> bool:
        """判断是否是最终回答"""
        return self.tool_name == "final_answer"

class OutputParser:
    """
    输出解析器
    负责解析 LLM 的输出
    """
    
    @staticmethod
    def parse_json_output(output: str) -> Optional[Dict[str, Any]]:
        """
        尝试从输出中提取并解析 JSON
        
        LLM 有时候可能会在 JSON 前后添加一些额外的文本，
        这个方法会尝试找到并解析其中的 JSON 部分。
        """
        # 首先，尝试直接解析
        try:
            return json.loads(output)
        except json.JSONDecodeError:
            pass
            
        # 如果直接解析失败，尝试寻找 JSON 块
        # 寻找第一个 '{' 和最后一个 '}'
        start_idx = output.find('{')
        end_idx = output.rfind('}')
        
        if start_idx != -1 and end_idx != -1 and start_idx < end_idx:
            json_str = output[start_idx:end_idx+1]
            try:
                return json.loads(json_str)
            except json.JSONDecodeError:
                pass
                
        return None
        
    @staticmethod
    def parse_agent_action(output: str) -> AgentAction:
        """
        解析 LLM 输出为 AgentAction 对象
        
        Args:
            output: LLM 的原始输出
            
        Returns:
            AgentAction 对象
        """
        parsed_json = OutputParser.parse_json_output(output)
        
        if not parsed_json:
            # 如果解析失败，构造一个默认的 action，让它直接返回原始输出
            return AgentAction(
                thinking="未能解析输出，直接展示",
                tool_name="final_answer",
                tool_input=f"我产生了一个输出，但格式可能有问题：\n{output}"
            )
            
        # 尝试提取字段
        thinking = parsed_json.get("thinking", "无思考过程")
        
        # 处理 action 字段
        action = parsed_json.get("action", {})
        tool_name = action.get("tool_name", "final_answer")
        tool_input = action.get("tool_input", "")
        
        # 兼容性处理：有些时候 LLM 可能会直接把内容放在顶层
        if not tool_input and "answer" in parsed_json:
            tool_input = parsed_json["answer"]
            tool_name = "final_answer"
            
        return AgentAction(
            thinking=thinking,
            tool_name=tool_name,
            tool_input=tool_input
        )

为什么我们需要这个模块？

即使我们在提示词中千叮咛万嘱咐，LLM 偶尔还是会“不听话”：

• 它可能在 JSON 前后加一些 markdown 标记（如 json ... ）。
• 它可能输出格式不完全正确的 JSON。
• 它可能完全忘记了要输出 JSON。

OutputParser 模块的作用就是尽可能鲁棒地处理这些情况，尽量让 Agent 能够继续运行下去，而不是一遇到格式错误就崩溃。注意 parse_json_output 方法，它不仅尝试直接解析，还会尝试从文本中“扣”出 JSON 部分。

---

步骤八：核心控制器——将一切整合起来

现在，我们已经有了所有的组件模块：LLM 接口、记忆、提示词管理、工具系统、输出解析。是时候写核心控制器了，它将把这些模块串联起来，形成一个完整的 Agent 循环。

创建 agent/core.py，这将是我们最核心的文件：

# agent/core.py
from typing import Optional, Dict, Any
from .llm import LLMClient
from .memory import MemoryManager
from .prompts import PromptManager
from .tools import ToolRegistry, create_sample_tools
from .parser import OutputParser, AgentAction

class Agent:
    """
    AI Agent 核心类
    整合所有组件，实现完整的 Agent 循环
    """
    def __init__(
        self,
        agent_name: str = "My Agent",
        llm_client: Optional[LLMClient] = None,
        memory: Optional[MemoryManager] = None,
        tool_registry: Optional[ToolRegistry] = None,
        system_prompt: Optional[str] = None,
        max_iterations: int = 5
    ):
        """
        初始化 Agent
        
        Args:
            agent_name: Agent 的名称
            llm_client: LLM 客户端实例，如果为 None 会自动创建
            memory: 记忆管理器实例，如果为 None 会自动创建
            tool_registry: 工具注册表实例，如果为 None 会创建一个空的
            system_prompt: 自定义系统提示词，如果为 None 会自动生成
            max_iterations: 最大执行迭代次数，防止 Agent 陷入无限循环
        """
        self.agent_name = agent_name
        self.llm = llm_client or LLMClient()
        self.memory = memory or MemoryManager(max_history_length=20)
        self.tools = tool_registry or ToolRegistry()
        self.max_iterations = max_iterations
        
        # 构建系统提示词
        if not system_prompt:
            tools_desc = self.tools.get_tools_description() if self.tools.list_tools() else None
            system_prompt = PromptManager.build_system_prompt(
                agent_name=agent_name,
                tools_description=tools_desc
            )
            
        # 将系统提示词加入记忆
        self.memory.add("system", system_prompt)
        
        print(f"[Agent] {agent_name} 初始化完成！")
        if self.tools.list_tools():
            print(f"[Agent] 已加载工具: {[t.name for t in self.tools.list_tools()]}")
        
    def run(self, user_input: str, verbose: bool = True) -> str:
        """
        运行 Agent，处理用户输入
        
        Args:
            user_input: 用户的输入
            verbose: 是否打印详细的执行过程
            
        Returns:
            Agent 的最终回答
        """
        # 1. 将用户输入加入记忆
        self.memory.add("user", user_input)
        
        if verbose:
            print(f"\n{'='*60}")
            print(f"[用户输入] {user_input}")
            print(f"{'='*60}")
        
        final_answer = ""
        
        # Agent 主循环
        for i in range(self.max_iterations):
            if verbose:
                print(f"\n[迭代 {i+1}/{self.max_iterations}] 思考中...")
            
            # 2. 从记忆构建消息，调用 LLM
            try:
                llm_output = self.llm.chat_completion(self.memory.to_llm_messages())
            except Exception as e:
                error_msg = f"调用 LLM 失败: {str(e)}"
                print(f"[错误] {error_msg}")
                return error_msg
                
            # 3. 解析 LLM 输出
            action = OutputParser.parse_agent_action(llm_output)
            
            if verbose:
                print(f"[思考过程] {action.thinking}")
                print(f"[决策] 使用工具: {action.tool_name}")
            
            # 4. 判断是否是最终答案
            if action.is_final_answer:
                final_answer = action.tool_input
                if verbose:
                    print(f"\n[最终答案] {final_answer}")
                break
                
            # 5. 执行工具
            if verbose:
                print(f"[执行工具] 输入参数: {action.tool_input}")
                
            tool_result = self.tools.execute_tool(action.tool_name, action.tool_input)
            
            if verbose:
                print(f"[工具结果] {tool_result}")
            
            # 6. 将 LLM 的输出和工具结果加入记忆
            # 注意：这里我们将 LLM 的原始输出作为 assistant 消息加入记忆
            self.memory.add("assistant", llm_output)
            # 将工具结果作为一个特殊的 "tool" 角色消息加入记忆
            self.memory.add("tool", f"Tool '{action.tool_name}' returned: {tool_result}")
            
        else:
            # 如果循环正常结束（即不是通过 break 跳出的），说明达到了最大迭代次数
            final_answer = "抱歉，我尝试了多次，但仍未能完成任务。请尝试换一种方式提问。"
            print(f"[警告] 达到最大迭代次数 {self.max_iterations}")
            
        # 将最终答案加入记忆
        if not action.is_final_answer:
            # 如果是因为达到最大迭代次数退出的，直接把最终答案作为 assistant 消息
            self.memory.add("assistant", final_answer)
        # 如果是正常 break 的，其实之前已经加过 assistant 消息了，这里可以优化一下
        
        return final_answer

核心逻辑解析——Agent 循环：

这是整个系统的核心，让我们用流程图来更直观地理解这个循环：

这个循环就是著名的 ReAct (Reasoning + Acting) 模式的一个实现。Agent 不是一次性给出答案，而是不断地“思考-行动-观察”，直到任务完成。

现在，让我们创建 main.py 来把这一切都运行起来：

# main.py
from agent.core import Agent
from agent.tools import create_sample_tools

def main():
    print("="*60)
    print("欢迎使用你的第一个 AI Agent！")
    print("="*60)
    
    # 创建工具注册表并添加示例工具
    tool_registry = create_sample_tools()
    
    # 初始化 Agent
    agent = Agent(
        agent_name="测试助手小 A",
        tool_registry=tool_registry,
        max_iterations=5,
        llm_client=LLMClient(temperature=0) # 温度设低一点，让输出更稳定
    )
    
    print("\n你可以开始提问了（输入 'quit' 退出）")
    print("-" * 60)
    
    while True:
        try:
            user_input = input("\n你: ")
            
            if user_input.lower() in ['quit', 'exit', 'q']:
                print("再见！")
                break
                
            if not user_input.strip():
                continue
                
            # 运行 Agent
            response = agent.run(user_input, verbose=True)
            
            # 如果 verbose 是 True，其实内部已经打印过了，
            # 这里我们可以选择不再重复打印，或者只在非 verbose 模式下打印
            
        except KeyboardInterrupt:
            print("\n\n收到中断信号，退出程序。")
            break
        except Exception as e:
            print(f"\n[系统错误] 发生了未预期的错误: {e}")
            import traceback
            traceback.print_exc()

if __name__ == "__main__":
    from agent.llm import LLMClient # 这里为了演示方便，直接导入
    main()

是时候见证奇迹了！运行 python main.py，你可以尝试问它一些问题，比如：

1. “现在几点了？” （应该会调用 get_current_time 工具）
2. “帮我算一下 234 * 456 等于多少？” （应该会调用 calculator 工具）
3. “请先告诉我现在的时间，然后计算一下当前时间的小时数乘以 7 是多少。”（这是一个多步任务，Agent 应该会先调用时间工具，然后调用计算器工具）

如果你看到 Agent 能够自主地选择工具、执行工具、并根据结果继续推理，那么恭喜你！你已经成功构建了你的第一个 AI Agent 系统！

---

5. 进阶探讨 (Advanced Topics)

我们已经完成了一个基础但功能完整的 Agent 系统。但这只是开始，AI Agent 领域还有非常多值得深入探索的话题。

如何封装一个通用的、可复用的图表组件？（哦不，是 Agent 组件！）

开个玩笑，回到 AI Agent。我们刚才的架构其实已经有了一些可复用的影子，但我们可以更进一步。

面向接口编程： 你可能已经注意到，我们的 LLMClient 是和 Open