0.2 面向对象与工程实践

学习目标 本章教你如何使用面向对象编程（OOP）设计可维护的 AI Agent，掌握文件操作、异常处理、环境配置等工程实践，为构建生产级应用做准备。

📋 本章内容

✅ 掌握面向对象编程（类、继承、封装）
✅ 学会优雅的异常处理（try-except、自定义异常）
✅ 掌握文件与 JSON 操作（保存对话历史）
✅ 学会环境配置管理（.env、API Key）
✅ 掌握 Git/GitHub 协作流程
✅ 理解虚拟环境与依赖管理

📚 术语表

术语名称	LangGraph 定义和解读	Python 定义和说明	重要程度
Class	Agent系统的蓝图，封装状态管理、工具调用、记忆存储等功能模块。虽然LangGraph节点是函数，但类用于组织复杂逻辑	面向对象编程的核心，通过`class`关键字定义，包含属性和方法	⭐⭐⭐⭐⭐
init	构造函数，初始化Agent配置（model、temperature）、状态存储（history）等实例属性	类的初始化方法，创建实例时自动调用，`self`表示实例本身	⭐⭐⭐⭐⭐
self	实例引用，访问Agent的状态和方法，如`self.history`存储对话历史	类方法的第一个参数，指向当前实例，约定俗成命名为self	⭐⭐⭐⭐⭐
Exception	处理API调用失败、模型超时、参数错误等异常情况。生产环境必备的容错机制	异常基类，通过`try-except`捕获和处理错误，`raise`主动抛出异常	⭐⭐⭐⭐⭐
try-except	捕获LLM调用失败、网络超时等异常，实现重试、降级、日志记录等容错策略	异常处理语法，try块执行可能出错的代码，except块处理特定异常	⭐⭐⭐⭐⭐
JSON	序列化Agent状态、对话历史、配置参数，持久化到文件或数据库	轻量级数据交换格式，`json.dumps()`序列化，`json.loads()`反序列化	⭐⭐⭐⭐⭐
.env	存储API Key、数据库密码等敏感配置，避免硬编码和泄露。生产部署的安全实践	环境变量配置文件，通过`python-dotenv`库加载，格式为`KEY=VALUE`	⭐⭐⭐⭐⭐
Inheritance	继承基础Agent类，扩展专用功能（如WeatherAgent、BookingAgent）。代码复用的核心机制	子类继承父类的属性和方法，通过`class Child(Parent)`语法实现	⭐⭐⭐⭐
Property	用`@property`装饰器创建只读属性（如token计数），保护内部状态不被意外修改	将方法转换为属性访问，支持`@setter`定义赋值行为，增强封装性	⭐⭐⭐
with	上下文管理器，自动关闭文件、数据库连接等资源。避免资源泄露的最佳实践	`with`语句保证资源正确释放，常用于文件操作`with open(...) as f`	⭐⭐⭐⭐

1. 面向对象编程：设计你的 Agent

1.1 类与对象：Agent 的蓝图

在 LangGraph 中，虽然使用函数式编程定义节点，但理解 OOP 对于设计复杂 Agent 系统至关重要。

python

# 代码示例：创建一个简单的 ChatBot 类
class ChatBot:
    """基础聊天机器人"""

    def __init__(self, name, model="gpt-5", temperature=0.7):
        """初始化方法

        Args:
            name: 机器人名称
            model: 使用的模型
            temperature: 温度参数
        """
        self.name = name
        self.model = model
        self.temperature = temperature
        self.conversation_history = []
        self.message_count = 0

    def add_message(self, role, content):
        """添加消息到历史"""
        message = {
            "role": role,
            "content": content,
            "index": self.message_count
        }
        self.conversation_history.append(message)
        self.message_count += 1

    def chat(self, user_message):
        """处理用户消息"""
        # 添加用户消息
        self.add_message("user", user_message)

        # 生成回复（简化版）
        reply = f"[{self.name}] 收到: {user_message}"
        self.add_message("assistant", reply)

        return reply

    def get_history(self):
        """获取对话历史"""
        return self.conversation_history

    def clear_history(self):
        """清空历史"""
        self.conversation_history = []
        self.message_count = 0
        print(f"✅ 已清空 {self.name} 的对话历史")

    def __str__(self):
        """字符串表示"""
        return f"ChatBot(name={self.name}, model={self.model}, messages={self.message_count})"

    def __repr__(self):
        """开发者表示"""
        return f"ChatBot('{self.name}', '{self.model}', temp={self.temperature})"


# ===== 使用示例 =====

# 创建机器人实例
bot = ChatBot(name="小助手", model="gpt-5", temperature=0.7)
print(bot)

# 对话
bot.chat("你好")
bot.chat("今天天气怎么样？")

# 查看历史
print(f"\n对话历史（共 {bot.message_count} 条）:")
for msg in bot.get_history():
    role_icon = "👤" if msg["role"] == "user" else "🤖"
    print(f"  [{msg['index']}] {role_icon} {msg['content']}")

# 清空历史
bot.clear_history()
print(f"\n清空后: {bot}")

运行结果：

ChatBot(name=小助手, model=gpt-5, messages=0)

对话历史（共 4 条）:
  [0] 👤 你好
  [1] 🤖 [小助手] 收到: 你好
  [2] 👤 今天天气怎么样？
  [3] 🤖 [小助手] 收到: 今天天气怎么样？

✅ 已清空 小助手 的对话历史

清空后: ChatBot(name=小助手, model=5, messages=0)

1.2 继承：扩展 Agent 功能

python

# 代码示例：创建增强版 ChatBot
class AdvancedChatBot(ChatBot):
    """增强版聊天机器人，支持意图分类"""

    def __init__(self, name, model="gpt-5", temperature=0.7, enable_intent=True):
        # 调用父类构造函数
        super().__init__(name, model, temperature)
        self.enable_intent = enable_intent
        self.intent_history = []

    def classify_intent(self, message):
        """简单的意图分类"""
        message_lower = message.lower()

        if any(kw in message_lower for kw in ["天气", "温度", "下雨"]):
            return "weather"
        elif any(kw in message_lower for kw in ["预订", "订票", "预约"]):
            return "booking"
        elif any(kw in message_lower for kw in ["你好", "hi", "hello"]):
            return "greeting"
        else:
            return "general"

    def chat(self, user_message):
        """重写父类方法，添加意图分类"""
        # 意图分类
        if self.enable_intent:
            intent = self.classify_intent(user_message)
            self.intent_history.append(intent)
            print(f"🎯 识别意图: {intent}")

        # 调用父类方法
        reply = super().chat(user_message)
        return reply

    def get_intent_stats(self):
        """获取意图统计"""
        stats = {}
        for intent in self.intent_history:
            stats[intent] = stats.get(intent, 0) + 1
        return stats

    def __str__(self):
        """重写字符串表示"""
        base = super().__str__()
        return f"{base}, intents={len(self.intent_history)}"


# ===== 使用示例 =====

# 创建增强版机器人
advanced_bot = AdvancedChatBot(name="智能助手", enable_intent=True)

# 测试对话
test_messages = [
    "你好！",
    "今天北京天气怎么样？",
    "帮我预订明天的机票",
    "随便聊聊"
]

print("=== 对话测试 ===")
for msg in test_messages:
    print(f"\n👤 {msg}")
    reply = advanced_bot.chat(msg)

# 意图统计
print("\n\n=== 意图统计 ===")
stats = advanced_bot.get_intent_stats()
for intent, count in sorted(stats.items(), key=lambda x: x[1], reverse=True):
    print(f"  {intent}: {count} 次")

print(f"\n{advanced_bot}")

运行结果：

=== 对话测试 ===

👤 你好！
🎯 识别意图: greeting

👤 今天北京天气怎么样？
🎯 识别意图: weather

👤 帮我预订明天的机票
🎯 识别意图: booking

👤 随便聊聊
🎯 识别意图: general


=== 意图统计 ===
  greeting: 1 次
  weather: 1 次
  booking: 1 次
  general: 1 次

ChatBot(name=智能助手, model=gpt-5, messages=8), intents=4

1.3 封装与属性：保护 Agent 状态

python

# 代码示例：使用属性保护敏感信息
class SecureChatBot:
    """安全的聊天机器人，保护 API Key"""

    def __init__(self, name, api_key):
        self.name = name
        self._api_key = api_key  # 私有属性（约定）
        self._call_count = 0

    @property
    def api_key(self):
        """API Key 的 getter（隐藏部分内容）"""
        if self._api_key:
            return f"{self._api_key[:8]}...{self._api_key[-4:]}"
        return None

    @api_key.setter
    def api_key(self, value):
        """API Key 的 setter（验证格式）"""
        if not value or len(value) < 20:
            raise ValueError("API Key 格式无效")
        self._api_key = value
        print("✅ API Key 已更新")

    @property
    def call_count(self):
        """只读属性"""
        return self._call_count

    def make_api_call(self):
        """模拟 API 调用"""
        self._call_count += 1
        return f"使用 API Key {self.api_key} 调用成功（第 {self._call_count} 次）"

    def __str__(self):
        return f"SecureChatBot(name={self.name}, calls={self.call_count})"


# ===== 使用示例 =====

# 创建安全机器人
bot = SecureChatBot(
    name="安全助手",
    api_key="sk-1234567890abcdefghijklmnopqrstuvwxyz"
)

# 查看 API Key（自动隐藏）
print(f"API Key: {bot.api_key}")

# 进行 API 调用
print(bot.make_api_call())
print(bot.make_api_call())

# 尝试直接修改 call_count（失败）
try:
    bot.call_count = 100
except AttributeError as e:
    print(f"\n❌ 无法直接修改: {e}")

# 更新 API Key
bot.api_key = "sk-newkey1234567890abcdefghijklmnopqrstuvwxyz"
print(f"新 API Key: {bot.api_key}")

# 无效的 API Key
try:
    bot.api_key = "short"
except ValueError as e:
    print(f"❌ {e}")

print(f"\n{bot}")

运行结果：

API Key: sk-12345...wxyz
使用 API Key sk-12345...wxyz 调用成功（第 1 次）
使用 API Key sk-12345...wxyz 调用成功（第 2 次）

❌ 无法直接修改: can't set attribute

✅ API Key 已更新
新 API Key: sk-newke...wxyz
❌ API Key 格式无效

SecureChatBot(name=安全助手, calls=2)

2. 异常处理：优雅地处理错误

2.1 基础异常处理

python

# 代码示例：API 调用的容错处理
def call_external_api(endpoint, retries=3):
    """调用外部 API（模拟）

    Args:
        endpoint: API 端点
        retries: 重试次数

    Returns:
        API 响应或 None
    """
    import random

    for attempt in range(1, retries + 1):
        try:
            print(f"尝试 #{attempt}: 调用 {endpoint}")

            # 模拟随机失败
            if random.random() < 0.6:
                raise ConnectionError("网络连接失败")

            # 模拟成功
            response = {"status": "success", "data": "天气数据"}
            print(f"  ✅ 成功!")
            return response

        except ConnectionError as e:
            print(f"  ❌ 连接错误: {e}")
            if attempt == retries:
                print(f"  ⚠️ 达到最大重试次数")
                return None
        except Exception as e:
            print(f"  ❌ 未知错误: {e}")
            return None

    return None


# ===== 使用示例 =====

result = call_external_api("/api/weather", retries=3)
if result:
    print(f"\n最终结果: {result}")
else:
    print("\n最终结果: API 调用失败")

运行结果：

尝试 #1: 调用 /api/weather
  ❌ 连接错误: 网络连接失败
尝试 #2: 调用 /api/weather
  ✅ 成功!

最终结果: {'status': 'success', 'data': '天气数据'}

2.2 自定义异常：Agent 特定错误

python

# 代码示例：定义 Agent 专用异常
class AgentError(Exception):
    """Agent 基础异常"""
    pass

class InvalidStateError(AgentError):
    """状态无效异常"""
    pass

class ToolExecutionError(AgentError):
    """工具执行异常"""
    def __init__(self, tool_name, reason):
        self.tool_name = tool_name
        self.reason = reason
        super().__init__(f"工具 '{tool_name}' 执行失败: {reason}")

class MessageLimitError(AgentError):
    """消息长度超限异常"""
    def __init__(self, actual, limit):
        self.actual = actual
        self.limit = limit
        super().__init__(f"消息长度 {actual} 超过限制 {limit}")


def execute_tool(tool_name, params):
    """执行工具（模拟）"""
    valid_tools = ["weather", "calculator", "search"]

    if tool_name not in valid_tools:
        raise ToolExecutionError(tool_name, "工具不存在")

    if not params:
        raise ToolExecutionError(tool_name, "参数为空")

    return f"{tool_name} 执行成功: {params}"


def validate_message(message, max_length=100):
    """验证消息长度"""
    if len(message) > max_length:
        raise MessageLimitError(len(message), max_length)
    return True


# ===== 使用示例 =====

print("=== 工具执行测试 ===")

# 成功案例
try:
    result = execute_tool("weather", {"city": "北京"})
    print(f"✅ {result}")
except ToolExecutionError as e:
    print(f"❌ {e}")

# 工具不存在
try:
    result = execute_tool("unknown_tool", {"param": "value"})
except ToolExecutionError as e:
    print(f"❌ {e}")
    print(f"   工具名: {e.tool_name}")
    print(f"   原因: {e.reason}")

# 参数为空
try:
    result = execute_tool("calculator", None)
except ToolExecutionError as e:
    print(f"❌ {e}")

print("\n=== 消息验证测试 ===")

# 正常消息
try:
    validate_message("这是一条正常的消息")
    print("✅ 消息验证通过")
except MessageLimitError as e:
    print(f"❌ {e}")

# 超长消息
try:
    long_message = "x" * 150
    validate_message(long_message, max_length=100)
except MessageLimitError as e:
    print(f"❌ {e}")
    print(f"   实际长度: {e.actual}")
    print(f"   限制长度: {e.limit}")

运行结果：

=== 工具执行测试 ===
✅ weather 执行成功: {'city': '北京'}
❌ 工具 'unknown_tool' 执行失败: 工具不存在
   工具名: unknown_tool
   原因: 工具不存在
❌ 工具 'calculator' 执行失败: 参数为空

=== 消息验证测试 ===
✅ 消息验证通过
❌ 消息长度 150 超过限制 100
   实际长度: 150
   限制长度: 100

2.3 上下文管理器：资源管理

python

# 代码示例：自定义上下文管理器
class ConversationSession:
    """对话会话管理器"""

    def __init__(self, session_id, auto_save=True):
        self.session_id = session_id
        self.auto_save = auto_save
        self.messages = []
        self.start_time = None

    def __enter__(self):
        """进入上下文"""
        from datetime import datetime
        self.start_time = datetime.now()
        print(f"📂 开始会话: {self.session_id}")
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """退出上下文"""
        from datetime import datetime
        end_time = datetime.now()
        duration = (end_time - self.start_time).total_seconds()

        print(f"\n📂 结束会话: {self.session_id}")
        print(f"   持续时间: {duration:.2f} 秒")
        print(f"   消息数量: {len(self.messages)}")

        if self.auto_save:
            self.save()

        # 处理异常
        if exc_type is not None:
            print(f"   ⚠️ 发生错误: {exc_val}")
            return False  # 不抑制异常

        return True

    def add_message(self, role, content):
        """添加消息"""
        self.messages.append({"role": role, "content": content})

    def save(self):
        """保存会话"""
        print(f"   💾 保存会话到文件: {self.session_id}.json")


# ===== 使用示例 =====

# 正常使用
print("=== 正常会话 ===")
with ConversationSession("session_001") as session:
    session.add_message("user", "你好")
    session.add_message("assistant", "你好！有什么可以帮你的？")
    session.add_message("user", "今天天气怎么样？")

# 异常处理
print("\n\n=== 异常会话 ===")
try:
    with ConversationSession("session_002") as session:
        session.add_message("user", "消息1")
        raise ValueError("模拟错误")
        session.add_message("user", "消息2")  # 不会执行
except ValueError as e:
    print(f"✅ 异常已捕获: {e}")

运行结果：

=== 正常会话 ===
📂 开始会话: session_001

📂 结束会话: session_001
   持续时间: 0.00 秒
   消息数量: 3
   💾 保存会话到文件: session_001.json


=== 异常会话 ===
📂 开始会话: session_002

📂 结束会话: session_002
   持续时间: 0.00 秒
   消息数量: 1
   ⚠️ 发生错误: 模拟错误
   💾 保存会话到文件: session_002.json
✅ 异常已捕获: 模拟错误

3. 文件操作：持久化 Agent 数据

3.1 JSON 文件：保存对话历史

python

# 代码示例：对话历史的保存与加载
import json
from datetime import datetime

class ConversationManager:
    """对话管理器"""

    def __init__(self):
        self.conversations = {}

    def create_conversation(self, conv_id):
        """创建新对话"""
        self.conversations[conv_id] = {
            "id": conv_id,
            "created_at": datetime.now().isoformat(),
            "messages": [],
            "metadata": {}
        }
        return self.conversations[conv_id]

    def add_message(self, conv_id, role, content):
        """添加消息"""
        if conv_id not in self.conversations:
            self.create_conversation(conv_id)

        message = {
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat()
        }
        self.conversations[conv_id]["messages"].append(message)

    def save_to_file(self, conv_id, filename):
        """保存对话到 JSON 文件"""
        if conv_id not in self.conversations:
            raise ValueError(f"对话 {conv_id} 不存在")

        with open(filename, "w", encoding="utf-8") as f:
            json.dump(
                self.conversations[conv_id],
                f,
                ensure_ascii=False,
                indent=2
            )
        print(f"✅ 对话已保存到: {filename}")

    def load_from_file(self, filename):
        """从 JSON 文件加载对话"""
        with open(filename, "r", encoding="utf-8") as f:
            data = json.load(f)

        conv_id = data["id"]
        self.conversations[conv_id] = data
        print(f"✅ 已加载对话: {conv_id}")
        return conv_id

    def export_all(self, filename):
        """导出所有对话"""
        with open(filename, "w", encoding="utf-8") as f:
            json.dump(
                self.conversations,
                f,
                ensure_ascii=False,
                indent=2
            )
        print(f"✅ 所有对话已导出到: {filename}")


# ===== 使用示例 =====

manager = ConversationManager()

# 创建对话
manager.create_conversation("conv_001")
manager.add_message("conv_001", "user", "你好")
manager.add_message("conv_001", "assistant", "你好！有什么可以帮你的？")
manager.add_message("conv_001", "user", "今天天气怎么样？")

# 保存单个对话
manager.save_to_file("conv_001", "conversation_001.json")

# 创建第二个对话
manager.create_conversation("conv_002")
manager.add_message("conv_002", "user", "帮我预订机票")
manager.add_message("conv_002", "assistant", "好的，请问是哪个城市？")

# 导出所有对话
manager.export_all("all_conversations.json")

# 加载对话
print("\n=== 加载测试 ===")
new_manager = ConversationManager()
loaded_id = new_manager.load_from_file("conversation_001.json")

conv = new_manager.conversations[loaded_id]
print(f"对话ID: {conv['id']}")
print(f"创建时间: {conv['created_at']}")
print(f"消息数量: {len(conv['messages'])}")
print("\n消息内容:")
for msg in conv['messages']:
    role_icon = "👤" if msg["role"] == "user" else "🤖"
    print(f"  {role_icon} {msg['content']}")

运行结果：

✅ 对话已保存到: conversation_001.json
✅ 所有对话已导出到: all_conversations.json

=== 加载测试 ===
✅ 已加载对话: conv_001
对话ID: conv_001
创建时间: 2024-10-27T10:00:00.000000
消息数量: 3

消息内容:
  👤 你好
  🤖 你好！有什么可以帮你的？
  👤 今天天气怎么样？

3.2 文本文件：日志记录

python

# 代码示例：Agent 日志系统
from datetime import datetime

class AgentLogger:
    """Agent 日志记录器"""

    def __init__(self, log_file="agent.log"):
        self.log_file = log_file

    def _format_log(self, level, message):
        """格式化日志"""
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        return f"[{timestamp}] [{level}] {message}\n"

    def info(self, message):
        """记录信息"""
        self._write(self._format_log("INFO", message))

    def warning(self, message):
        """记录警告"""
        self._write(self._format_log("WARNING", message))

    def error(self, message):
        """记录错误"""
        self._write(self._format_log("ERROR", message))

    def _write(self, log_entry):
        """写入日志文件"""
        with open(self.log_file, "a", encoding="utf-8") as f:
            f.write(log_entry)
        print(log_entry.strip())

    def read_logs(self, lines=10):
        """读取最近的日志"""
        try:
            with open(self.log_file, "r", encoding="utf-8") as f:
                all_lines = f.readlines()
                return all_lines[-lines:] if lines else all_lines
        except FileNotFoundError:
            return []


# ===== 使用示例 =====

logger = AgentLogger("agent_demo.log")

# 记录不同级别的日志
logger.info("Agent 启动")
logger.info("加载配置文件")
logger.warning("API Key 未设置，使用默认值")
logger.info("开始处理用户消息")
logger.error("工具调用失败: connection timeout")
logger.info("Agent 停止")

# 读取日志
print("\n=== 最近 3 条日志 ===")
recent_logs = logger.read_logs(lines=3)
for log in recent_logs:
    print(log.strip())

运行结果：

[2024-10-27 10:00:00] [INFO] Agent 启动
[2024-10-27 10:00:00] [INFO] 加载配置文件
[2024-10-27 10:00:00] [WARNING] API Key 未设置，使用默认值
[2024-10-27 10:00:00] [INFO] 开始处理用户消息
[2024-10-27 10:00:00] [ERROR] 工具调用失败: connection timeout
[2024-10-27 10:00:00] [INFO] Agent 停止

=== 最近 3 条日志 ===
[2024-10-27 10:00:00] [ERROR] 工具调用失败: connection timeout
[2024-10-27 10:00:00] [INFO] Agent 停止

4. 环境配置：管理 API Key 和敏感信息

4.1 使用 .env 文件

python

# 代码示例：环境变量管理
import os

# 方法1：直接使用 os.environ（不推荐）
# os.environ["OPENAI_API_KEY"] = "sk-..."  # 硬编码，不安全

# 方法2：从 .env 文件加载（推荐）
def load_env_file(filename=".env"):
    """手动加载 .env 文件"""
    if not os.path.exists(filename):
        print(f"⚠️ {filename} 文件不存在")
        return

    with open(filename, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line and not line.startswith("#"):
                key, value = line.split("=", 1)
                os.environ[key.strip()] = value.strip()
    print(f"✅ 已加载环境变量从 {filename}")


# 创建示例 .env 文件
env_content = """# API Keys
OPENAI_API_KEY=sk-1234567890abcdef
ANTHROPIC_API_KEY=sk-ant-9876543210

# Configuration
MODEL_NAME=gpt-5
TEMPERATURE=0.7
MAX_TOKENS=1000
"""

with open("example.env", "w", encoding="utf-8") as f:
    f.write(env_content)

# 加载环境变量
load_env_file("example.env")

# 读取环境变量
openai_key = os.getenv("OPENAI_API_KEY", "NOT_SET")
model = os.getenv("MODEL_NAME", "gpt-5")
temp = float(os.getenv("TEMPERATURE", "0.7"))

print("\n=== 环境配置 ===")
print(f"OpenAI Key: {openai_key[:10]}...{openai_key[-4:]}")
print(f"Model: {model}")
print(f"Temperature: {temp}")

# 检查必需的环境变量
required_vars = ["OPENAI_API_KEY", "ANTHROPIC_API_KEY"]
missing = [var for var in required_vars if not os.getenv(var)]

if missing:
    print(f"\n⚠️ 缺少环境变量: {missing}")
else:
    print(f"\n✅ 所有必需环境变量已设置")

运行结果：

✅ 已加载环境变量从 example.env

=== 环境配置 ===
OpenAI Key: sk-1234567...cdef
Model: gpt-5
Temperature: 0.7

✅ 所有必需环境变量已设置

4.2 配置类：结构化配置管理

python

# 代码示例：使用类管理配置
import os
from typing import Optional

class AgentConfig:
    """Agent 配置管理"""

    def __init__(self, config_file: Optional[str] = None):
        # 从环境变量或默认值加载
        self.openai_api_key = os.getenv("OPENAI_API_KEY")
        self.model = os.getenv("MODEL_NAME", "gpt-5")
        self.temperature = float(os.getenv("TEMPERATURE", "0.7"))
        self.max_tokens = int(os.getenv("MAX_TOKENS", "1000"))
        self.max_retries = int(os.getenv("MAX_RETRIES", "3"))

        if config_file:
            self.load_from_file(config_file)

    def load_from_file(self, filename):
        """从文件加载配置"""
        import json
        try:
            with open(filename, "r", encoding="utf-8") as f:
                config = json.load(f)

            # 更新配置
            for key, value in config.items():
                if hasattr(self, key):
                    setattr(self, key, value)

            print(f"✅ 已从 {filename} 加载配置")
        except FileNotFoundError:
            print(f"⚠️ 配置文件 {filename} 不存在")

    def validate(self):
        """验证配置"""
        errors = []

        if not self.openai_api_key:
            errors.append("缺少 OPENAI_API_KEY")

        if not (0 <= self.temperature <= 2):
            errors.append(f"temperature 应在 0-2 之间，当前: {self.temperature}")

        if self.max_tokens < 1:
            errors.append(f"max_tokens 应大于 0，当前: {self.max_tokens}")

        if errors:
            raise ValueError(f"配置验证失败: {errors}")

        print("✅ 配置验证通过")
        return True

    def __repr__(self):
        """字符串表示"""
        key_preview = f"{self.openai_api_key[:8]}..." if self.openai_api_key else "NOT_SET"
        return f"""AgentConfig(
    openai_api_key={key_preview},
    model={self.model},
    temperature={self.temperature},
    max_tokens={self.max_tokens},
    max_retries={self.max_retries}
)"""


# ===== 使用示例 =====

# 创建配置
config = AgentConfig()
print(config)

# 验证配置
try:
    config.validate()
except ValueError as e:
    print(f"❌ {e}")

# 创建配置文件
config_data = {
    "model": "gpt-5",
    "temperature": 0.5,
    "max_tokens": 2000
}

with open("agent_config.json", "w", encoding="utf-8") as f:
    import json
    json.dump(config_data, f, indent=2)

# 从文件加载
config2 = AgentConfig("agent_config.json")
print(f"\n更新后的配置:\n{config2}")

运行结果：

AgentConfig(
    openai_api_key=sk-12345...,
    model=gpt-5,
    temperature=0.7,
    max_tokens=1000,
    max_retries=3
)
✅ 配置验证通过

✅ 已从 agent_config.json 加载配置

更新后的配置:
AgentConfig(
    openai_api_key=sk-12345...,
    model=gpt-5,
    temperature=0.5,
    max_tokens=2000,
    max_retries=3
)

5. Git/GitHub：版本控制与协作

5.1 Git 基础工作流

bash

# ===== 初始化 Git 仓库 =====

# 1. 配置 Git（首次使用）
git config --global user.name "你的名字"
git config --global user.email "your.email@example.com"

# 2. 初始化项目
mkdir my-agent-project
cd my-agent-project
git init

# 3. 创建项目文件
echo "# AI Agent Project" > README.md
echo "*.pyc" > .gitignore
echo "__pycache__/" >> .gitignore
echo ".env" >> .gitignore

# 4. 查看状态
git status

# 5. 添加文件
git add README.md .gitignore

# 6. 提交更改
git commit -m "Initial commit: Add README and gitignore"

# 7. 查看历史
git log --oneline

运行结果：

bash

Initialized empty Git repository in /path/to/my-agent-project/.git/

On branch main
No commits yet

Untracked files:
  README.md
  .gitignore

[main (root-commit) a1b2c3d] Initial commit: Add README and gitignore
 2 files changed, 5 insertions(+)

a1b2c3d Initial commit: Add README and gitignore

5.2 .gitignore：忽略敏感文件

bash

# 创建 .gitignore 文件
cat > .gitignore << 'EOF'
# Python
__pycache__/
*.py[cod]
*.so
.Python
*.egg-info/
dist/
build/

# 环境变量（重要！）
.env
.env.local
*.key
secrets.json

# 虚拟环境
venv/
env/
ENV/

# Jupyter Notebook
.ipynb_checkpoints/
*.ipynb

# IDE
.vscode/
.idea/
*.swp
*.swo

# 数据文件
*.db
*.sqlite
*.csv
*.pkl

# 日志
*.log
logs/

# macOS
.DS_Store

# 项目特定
agent_cache/
temp/
EOF

# 提交 .gitignore
git add .gitignore
git commit -m "Add comprehensive .gitignore"

5.3 GitHub 工作流

bash

# ===== 连接 GitHub =====

# 1. 在 GitHub 上创建新仓库（通过网页）
# 访问: https://github.com/new

# 2. 连接远程仓库
git remote add origin https://github.com/your-username/my-agent-project.git

# 3. 推送到 GitHub
git push -u origin main

# ===== 日常开发流程 =====

# 1. 创建新分支
git checkout -b feature/add-intent-classifier

# 2. 修改代码
echo "class IntentClassifier: pass" > classifier.py

# 3. 添加并提交
git add classifier.py
git commit -m "feat: Add intent classifier skeleton"

# 4. 推送分支
git push origin feature/add-intent-classifier

# 5. 在 GitHub 上创建 Pull Request

# 6. 合并后切回主分支
git checkout main
git pull origin main

# 7. 删除已合并的分支
git branch -d feature/add-intent-classifier

5.4 常用 Git 命令速查

命令	说明	示例
`git init`	初始化仓库	`git init`
`git clone`	克隆远程仓库	`git clone https://github.com/user/repo.git`
`git status`	查看状态	`git status`
`git add`	添加到暂存区	`git add .`
`git commit`	提交更改	`git commit -m "message"`
`git push`	推送到远程	`git push origin main`
`git pull`	拉取远程更改	`git pull`
`git log`	查看历史	`git log --oneline`
`git diff`	查看差异	`git diff`
`git branch`	查看/创建分支	`git branch feature-x`
`git checkout`	切换分支	`git checkout main`
`git merge`	合并分支	`git merge feature-x`
`git stash`	暂存更改	`git stash save "temp"`
`git reset`	重置更改	`git reset --hard HEAD`

6. 虚拟环境与依赖管理

6.1 创建虚拟环境

bash

# ===== 使用 venv（Python 内置）=====

# 1. 创建虚拟环境
python -m venv agent_env

# 2. 激活虚拟环境
# macOS/Linux:
source agent_env/bin/activate

# Windows:
agent_env\Scripts\activate

# 3. 验证环境
which python  # 应显示虚拟环境路径
python --version

# 4. 安装依赖
pip install langchain langgraph openai

# 5. 查看已安装包
pip list

# 6. 退出虚拟环境
deactivate

6.2 依赖管理：requirements.txt

bash

# ===== 创建 requirements.txt =====

# 1. 导出当前环境依赖
pip freeze > requirements.txt

# 2. 手动编辑 requirements.txt（推荐）
cat > requirements.txt << 'EOF'
# Core AI Libraries
langchain==0.1.0
langgraph==0.0.20
openai==1.10.0

# Data Processing
numpy==1.24.0
pandas==2.0.0

# Web Framework
fastapi==0.109.0
uvicorn==0.27.0

# Environment
python-dotenv==1.0.0

# Testing
pytest==8.0.0
pytest-asyncio==0.23.0
EOF

# 3. 安装依赖（新环境）
pip install -r requirements.txt

# 4. 升级所有包
pip install --upgrade -r requirements.txt

# 5. 仅安装生产依赖（创建 requirements-prod.txt）
cat > requirements-prod.txt << 'EOF'
langchain==0.1.0
langgraph==0.0.20
openai==1.10.0
fastapi==0.109.0
uvicorn==0.27.0
python-dotenv==1.0.0
EOF

pip install -r requirements-prod.txt

6.3 项目结构最佳实践

bash

# ===== 推荐的 Agent 项目结构 =====

my-agent-project/
├── .env                    # 环境变量（不提交到 git）
├── .env.example            # 环境变量模板
├── .gitignore              # Git 忽略文件
├── README.md               # 项目文档
├── requirements.txt        # 依赖列表
├── requirements-dev.txt    # 开发依赖
├── agent/                  # 主代码目录
│   ├── __init__.py
│   ├── config.py           # 配置管理
│   ├── graph.py            # LangGraph 定义
│   ├── nodes.py            # 节点函数
│   ├── tools.py            # 工具定义
│   └── utils.py            # 辅助函数
├── tests/                  # 测试目录
│   ├── __init__.py
│   ├── test_nodes.py
│   └── test_tools.py
├── data/                   # 数据文件
│   └── conversations/
├── logs/                   # 日志文件
└── scripts/                # 辅助脚本
    └── run_agent.py

7. 综合实战：完整的 Agent 工程项目

让我们整合所有知识，构建一个生产级的 Agent 系统。

python

# 代码示例：config.py - 配置管理
import os
from typing import Optional
from dotenv import load_dotenv

class AgentConfig:
    """Agent 配置管理"""

    def __init__(self, env_file: Optional[str] = ".env"):
        # 加载环境变量
        if env_file and os.path.exists(env_file):
            load_dotenv(env_file)

        # API 配置
        self.openai_api_key = os.getenv("OPENAI_API_KEY")
        self.model = os.getenv("MODEL", "5")
        self.temperature = float(os.getenv("TEMPERATURE", "0.7"))
        self.max_tokens = int(os.getenv("MAX_TOKENS", "1000"))

        # Agent 配置
        self.max_retries = int(os.getenv("MAX_RETRIES", "3"))
        self.timeout = int(os.getenv("TIMEOUT", "30"))

        # 日志配置
        self.log_level = os.getenv("LOG_LEVEL", "INFO")
        self.log_file = os.getenv("LOG_FILE", "agent.log")

    def validate(self):
        """验证配置"""
        if not self.openai_api_key:
            raise ValueError("必须设置 OPENAI_API_KEY")
        return True


# 代码示例：logger.py - 日志管理
from datetime import datetime
from typing import Literal

class AgentLogger:
    """Agent 日志记录器"""

    def __init__(self, log_file: str = "agent.log", level: str = "INFO"):
        self.log_file = log_file
        self.level = level
        self.levels = {"DEBUG": 0, "INFO": 1, "WARNING": 2, "ERROR": 3}

    def _should_log(self, level: str) -> bool:
        """判断是否应该记录"""
        return self.levels.get(level, 1) >= self.levels.get(self.level, 1)

    def _log(self, level: str, message: str):
        """内部日志方法"""
        if not self._should_log(level):
            return

        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        log_entry = f"[{timestamp}] [{level}] {message}\n"

        # 写入文件
        with open(self.log_file, "a", encoding="utf-8") as f:
            f.write(log_entry)

        # 打印到控制台
        print(log_entry.strip())

    def debug(self, message: str):
        self._log("DEBUG", message)

    def info(self, message: str):
        self._log("INFO", message)

    def warning(self, message: str):
        self._log("WARNING", message)

    def error(self, message: str):
        self._log("ERROR", message)


# 代码示例：exceptions.py - 自定义异常
class AgentError(Exception):
    """Agent 基础异常"""
    pass

class ConfigurationError(AgentError):
    """配置错误"""
    pass

class ToolExecutionError(AgentError):
    """工具执行错误"""
    pass


# 代码示例：agent.py - 主 Agent 类
from typing import List, Dict, Any

class ProductionAgent:
    """生产级 Agent"""

    def __init__(self, config: Optional[AgentConfig] = None):
        # 加载配置
        self.config = config or AgentConfig()
        self.config.validate()

        # 初始化日志
        self.logger = AgentLogger(
            log_file=self.config.log_file,
            level=self.config.log_level
        )

        # 状态管理
        self.state = {
            "messages": [],
            "context": {},
            "metadata": {
                "created_at": datetime.now().isoformat(),
                "call_count": 0
            }
        }

        self.logger.info("Agent 初始化完成")

    def process_message(self, message: str) -> Dict[str, Any]:
        """处理用户消息"""
        try:
            self.logger.info(f"处理消息: {message}")

            # 更新状态
            self.state["messages"].append({
                "role": "user",
                "content": message
            })
            self.state["metadata"]["call_count"] += 1

            # 简化的处理逻辑
            response = self._generate_response(message)

            # 添加响应
            self.state["messages"].append({
                "role": "assistant",
                "content": response
            })

            self.logger.info("消息处理完成")

            return {
                "status": "success",
                "response": response,
                "call_count": self.state["metadata"]["call_count"]
            }

        except Exception as e:
            self.logger.error(f"处理消息失败: {e}")
            raise AgentError(f"消息处理失败: {e}")

    def _generate_response(self, message: str) -> str:
        """生成响应（简化版）"""
        # 实际应用中这里会调用 LLM
        return f"收到你的消息: {message}"

    def save_conversation(self, filename: str):
        """保存对话历史"""
        try:
            import json
            with open(filename, "w", encoding="utf-8") as f:
                json.dump(self.state, f, ensure_ascii=False, indent=2)
            self.logger.info(f"对话已保存到: {filename}")
        except Exception as e:
            self.logger.error(f"保存对话失败: {e}")
            raise

    def __repr__(self):
        return f"ProductionAgent(calls={self.state['metadata']['call_count']})"


# ===== 使用示例 =====

# 创建 .env 文件
env_content = """OPENAI_API_KEY=sk-test1234567890
MODEL=gpt-5
TEMPERATURE=0.7
LOG_LEVEL=INFO
"""

with open("agent.env", "w") as f:
    f.write(env_content)

# 创建 Agent
print("=== 初始化 Agent ===")
agent = ProductionAgent(AgentConfig(env_file="agent.env"))

# 处理消息
messages = [
    "你好",
    "今天天气怎么样？",
    "帮我总结一下我们的对话"
]

print("\n=== 处理消息 ===")
for msg in messages:
    result = agent.process_message(msg)
    print(f"✅ 响应: {result['response']}")

# 保存对话
print("\n=== 保存对话 ===")
agent.save_conversation("production_conversation.json")

print(f"\n{agent}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207

运行结果：

=== 初始化 Agent ===
[2024-10-27 10:00:00] [INFO] Agent 初始化完成

=== 处理消息 ===
[2024-10-27 10:00:00] [INFO] 处理消息: 你好
[2024-10-27 10:00:00] [INFO] 消息处理完成
✅ 响应: 收到你的消息: 你好
[2024-10-27 10:00:00] [INFO] 处理消息: 今天天气怎么样？
[2024-10-27 10:00:00] [INFO] 消息处理完成
✅ 响应: 收到你的消息: 今天天气怎么样？
[2024-10-27 10:00:00] [INFO] 处理消息: 帮我总结一下我们的对话
[2024-10-27 10:00:00] [INFO] 消息处理完成
✅ 响应: 收到你的消息: 帮我总结一下我们的对话

=== 保存对话 ===
[2024-10-27 10:00:00] [INFO] 对话已保存到: production_conversation.json

ProductionAgent(calls=3)

8. 本章总结

✅ 你已经掌握

面向对象编程
- 类与对象：设计 Agent 蓝图
- 继承：扩展功能
- 封装：保护敏感信息
- 属性：@property 装饰器
异常处理
- try-except-finally
- 自定义异常
- 上下文管理器
文件操作
- JSON 文件：持久化对话
- 文本文件：日志记录
- with 语句：安全的资源管理
环境配置
- .env 文件管理
- 配置类设计
- API Key 保护
Git/GitHub
- 版本控制基础
- .gitignore 最佳实践
- 协作工作流
工程实践
- 虚拟环境
- requirements.txt
- 项目结构设计

📚 下一步

准备好了吗？接下来我们将学习：

0.3-AI开发工具链.md - 掌握 NumPy、Pandas、LangChain、Streamlit 等 AI 生态系统

💡 实战建议

创建自己的 Agent 项目，应用本章知识
为项目配置 Git 仓库并推送到 GitHub
使用 .env 文件管理 API Key
实现完整的日志系统

🎯 记住：优秀的工程实践是构建可维护、可扩展 AI Agent 的基础。投资时间学习这些技能，会让你的开发效率提升10倍！

0.2 面向对象与工程实践 ​

📋 本章内容 ​

📚 术语表 ​

1. 面向对象编程：设计你的 Agent ​

1.1 类与对象：Agent 的蓝图 ​

1.2 继承：扩展 Agent 功能 ​

1.3 封装与属性：保护 Agent 状态 ​

2. 异常处理：优雅地处理错误 ​

2.1 基础异常处理 ​

2.2 自定义异常：Agent 特定错误 ​

2.3 上下文管理器：资源管理 ​

3. 文件操作：持久化 Agent 数据 ​

3.1 JSON 文件：保存对话历史 ​

3.2 文本文件：日志记录 ​

4. 环境配置：管理 API Key 和敏感信息 ​

4.1 使用 .env 文件 ​

4.2 配置类：结构化配置管理 ​

5. Git/GitHub：版本控制与协作 ​

5.1 Git 基础工作流 ​

5.2 .gitignore：忽略敏感文件 ​

5.3 GitHub 工作流 ​

5.4 常用 Git 命令速查 ​

6. 虚拟环境与依赖管理 ​

6.1 创建虚拟环境 ​

6.2 依赖管理：requirements.txt ​

6.3 项目结构最佳实践 ​

7. 综合实战：完整的 Agent 工程项目 ​

8. 本章总结 ​

✅ 你已经掌握 ​

📚 下一步 ​

💡 实战建议 ​

0.2 面向对象与工程实践

📋 本章内容

📚 术语表

1. 面向对象编程：设计你的 Agent

1.1 类与对象：Agent 的蓝图

1.2 继承：扩展 Agent 功能

1.3 封装与属性：保护 Agent 状态

2. 异常处理：优雅地处理错误

2.1 基础异常处理

2.2 自定义异常：Agent 特定错误

2.3 上下文管理器：资源管理

3. 文件操作：持久化 Agent 数据

3.1 JSON 文件：保存对话历史

3.2 文本文件：日志记录

4. 环境配置：管理 API Key 和敏感信息

4.1 使用 .env 文件

4.2 配置类：结构化配置管理

5. Git/GitHub：版本控制与协作

5.1 Git 基础工作流

5.2 .gitignore：忽略敏感文件

5.3 GitHub 工作流

5.4 常用 Git 命令速查

6. 虚拟环境与依赖管理

6.1 创建虚拟环境

6.2 依赖管理：requirements.txt

6.3 项目结构最佳实践

7. 综合实战：完整的 Agent 工程项目

8. 本章总结

✅ 你已经掌握

📚 下一步

💡 实战建议