文章
7 · 面向对象：封装、继承与系统解耦
阅读数据加载中…
点赞数据加载中…
🔗 知识图谱导航：阅读本文前，建议先掌握/回顾 《6 · 闭包与装饰器：函数的终极形态》 中的核心概念；本文会在这个基础上继续推进。 承上启下：上一篇我们用装饰器扩展函数行为。现在进入更大的组织单元——类。OOP 的核心价值不是"写类"，而是系统解耦：让不同组件可以独立变化。
极客解析：先把数据流、控制流和模块边界跑通，再谈抽象；每段代码都围绕一个可执行 CLI 闭环展开。
痛点与架构
用大模型接口封装演示 OOP：你有 OpenAI、本地 Ollama、Mock 三种后端，上层代码不应该关心用的是哪个。
BaseAgent（抽象基类）
├── OpenAIAgent（子类：调用 OpenAI API）
├── OllamaAgent（子类：调用本地 Ollama）
└── MockAgent （子类：测试用 Mock）
AgentRouter（路由器）
└── 根据配置选择 Agent，上层代码无感知
| OOP 特性 | 体现 | 价值 |
|---|---|---|
| 封装 | __api_key 私有，@property 暴露接口 |
隐藏实现细节 |
| 继承 | OpenAIAgent(BaseAgent) |
复用公共逻辑 |
| 多态 | 所有 Agent 都有 .chat() 方法 |
上层代码统一调用 |
实战演练场
步步为营：核心逻辑自适应拆解
这一篇不要求你一口气吃下面向对象。我们把它拆成“数据盒子 -> 接口合同 -> 具体子类 -> 路由分发 -> 会话管理 -> CLI 总控台”。你先跑小块，再看完整架构，就不会被类、继承、多态一起砸懵。
Step 1：用 Message 把一条聊天消息装进标准盒子
痛点与机制：
Message 是最小的数据盒子，里面固定放 role、content、timestamp。新手可以把它理解成一张聊天记录卡片：谁说的、说了什么、什么时候说的，都有固定格子，不再是散乱字符串。
核心源码（逐字来自文末完整源码）：
@dataclass
class Message:
role: str # "user" | "assistant" | "system"
content: str
timestamp: str = field(default_factory=lambda: datetime.now().strftime("%H:%M:%S"))
def __str__(self) -> str:
icon = {"user": "👤", "assistant": "🤖", "system": "⚙️"}.get(self.role, "?")
return f"[{self.timestamp}] {icon} {self.role}: {self.content}"
可运行演示（补齐 Mock 数据与 print 反馈）：
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class Message:
role: str # "user" | "assistant" | "system"
content: str
timestamp: str = field(default_factory=lambda: datetime.now().strftime("%H:%M:%S"))
def __str__(self) -> str:
icon = {"user": "👤", "assistant": "🤖", "system": "⚙️"}.get(self.role, "?")
return f"[{self.timestamp}] {icon} {self.role}: {self.content}"
msg = Message(role="user", content="你好，什么是面向对象？")
print("对象原貌:", msg)
print("角色字段:", msg.role)
print("内容字段:", msg.content)
Step 2：用 AgentConfig 集中管理后端和模型参数
痛点与机制：
AgentConfig 把运行参数集中管理。不要让 backend、model、timeout 到处散落，否则后期改模型就像满屋子找遥控器；放进配置类后，上层只需要传一个对象。
核心源码（逐字来自文末完整源码）：
@dataclass
class AgentConfig:
backend: str
model: str
temperature: float = 0.7
max_tokens: int = 512
timeout: float = 30.0
可运行演示（补齐 Mock 数据与 print 反馈）：
from dataclasses import dataclass
# Step 2：配置对象像遥控器，把 backend/model/temperature 放在一个地方调。
@dataclass
class AgentConfig:
backend: str
model: str
temperature: float = 0.7
max_tokens: int = 512
timeout: float = 30.0
config = AgentConfig(backend="mock", model="mock-v1", temperature=0.3)
print("后端:", config.backend)
print("模型:", config.model)
print("温度:", config.temperature)
print("超时时间:", config.timeout)
Step 3：用 BaseAgent 定义所有智能体都必须遵守的接口合同
痛点与机制：
BaseAgent 不直接干活，它像一份岗位说明书：任何 Agent 子类都必须实现 chat() 和 stream_chat()。这样上层代码不用管底下是 Mock、Ollama 还是未来的 OpenAI，只认同一套接口。
核心源码（逐字来自文末完整源码）：
class BaseAgent(ABC):
"""所有 Agent 必须实现的接口契约。"""
def __init__(self, config: AgentConfig) -> None:
self._config = config
self._history: list[Message] = []
self._call_count: int = 0
@property
def config(self) -> AgentConfig:
"""只读属性：外部只能读配置，不能直接修改。"""
return self._config
@property
def call_count(self) -> int:
return self._call_count
@abstractmethod
def chat(self, user_input: str) -> str:
"""子类必须实现：发送消息并返回回复。"""
...
@abstractmethod
def stream_chat(self, user_input: str) -> Iterator[str]:
"""子类必须实现：流式返回回复（逐 token）。"""
...
def add_to_history(self, role: str, content: str) -> None:
self._history.append(Message(role=role, content=content))
def clear_history(self) -> None:
self._history.clear()
def print_history(self) -> None:
print(f"\n 对话历史（{len(self._history)} 条）:")
for msg in self._history:
print(f" {msg}")
def __repr__(self) -> str:
return f"{self.__class__.__name__}(model={self._config.model}, calls={self._call_count})"
可运行演示（补齐 Mock 数据与 print 反馈）：
import argparse
import time
import random
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Iterator
@dataclass
class Message:
role: str # "user" | "assistant" | "system"
content: str
timestamp: str = field(default_factory=lambda: datetime.now().strftime("%H:%M:%S"))
def __str__(self) -> str:
icon = {"user": "👤", "assistant": "🤖", "system": "⚙️"}.get(self.role, "?")
return f"[{self.timestamp}] {icon} {self.role}: {self.content}"
@dataclass
class AgentConfig:
backend: str
model: str
temperature: float = 0.7
max_tokens: int = 512
timeout: float = 30.0
# Step 3：抽象基类像合同，规定子类必须会 chat 和 stream_chat。
class BaseAgent(ABC):
"""所有 Agent 必须实现的接口契约。"""
def __init__(self, config: AgentConfig) -> None:
self._config = config
self._history: list[Message] = []
self._call_count: int = 0
@property
def config(self) -> AgentConfig:
"""只读属性：外部只能读配置，不能直接修改。"""
return self._config
@property
def call_count(self) -> int:
return self._call_count
@abstractmethod
def chat(self, user_input: str) -> str:
"""子类必须实现：发送消息并返回回复。"""
...
@abstractmethod
def stream_chat(self, user_input: str) -> Iterator[str]:
"""子类必须实现：流式返回回复（逐 token）。"""
...
def add_to_history(self, role: str, content: str) -> None:
self._history.append(Message(role=role, content=content))
def clear_history(self) -> None:
self._history.clear()
def print_history(self) -> None:
print(f"\n 对话历史（{len(self._history)} 条）:")
for msg in self._history:
print(f" {msg}")
def __repr__(self) -> str:
return f"{self.__class__.__name__}(model={self._config.model}, calls={self._call_count})"
class EchoAgent(BaseAgent):
def chat(self, user_input: str) -> str:
self._call_count += 1
self.add_to_history("user", user_input)
reply = f"Echo: {user_input}"
self.add_to_history("assistant", reply)
return reply
def stream_chat(self, user_input: str) -> Iterator[str]:
yield self.chat(user_input)
agent = EchoAgent(AgentConfig(backend="echo", model="echo-v1"))
print("回复:", agent.chat("测试抽象类"))
print("调用次数:", agent.call_count)
agent.print_history()
Step 4：用 MockAgent 做零网络依赖的可测试子类
痛点与机制：
MockAgent 的价值不是“假”，而是“稳定可测”。没有 API Key、没有网络、没有本地模型，也能把业务流程跑通。新手先用 MockAgent 学 OOP，比一开始就卡在外部服务上更稳。
核心源码（逐字来自文末完整源码）：
class MockAgent(BaseAgent):
"""Mock Agent：用于测试，无需真实 API。"""
RESPONSES: dict[str, str] = {
"你好": "你好！我是 MockAgent，随时为你服务。",
"python": "Python 是一门优雅的语言，推荐从类型提示开始学习。",
"default": "这是一个模拟回复，用于演示 OOP 多态特性。",
}
def chat(self, user_input: str) -> str:
self._call_count += 1
self.add_to_history("user", user_input)
time.sleep(0.05) # 模拟网络延迟
# 简单关键词匹配
for keyword, response in self.RESPONSES.items():
if keyword in user_input.lower():
reply = response
break
else:
reply = self.RESPONSES["default"]
self.add_to_history("assistant", reply)
return reply
def stream_chat(self, user_input: str) -> Iterator[str]:
"""流式输出：逐词返回。"""
full_reply = self.chat(user_input)
for word in full_reply.split():
time.sleep(0.05)
yield word + " "
可运行演示（补齐 Mock 数据与 print 反馈）：
import argparse
import time
import random
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Iterator
@dataclass
class Message:
role: str # "user" | "assistant" | "system"
content: str
timestamp: str = field(default_factory=lambda: datetime.now().strftime("%H:%M:%S"))
def __str__(self) -> str:
icon = {"user": "👤", "assistant": "🤖", "system": "⚙️"}.get(self.role, "?")
return f"[{self.timestamp}] {icon} {self.role}: {self.content}"
@dataclass
class AgentConfig:
backend: str
model: str
temperature: float = 0.7
max_tokens: int = 512
timeout: float = 30.0
class BaseAgent(ABC):
"""所有 Agent 必须实现的接口契约。"""
def __init__(self, config: AgentConfig) -> None:
self._config = config
self._history: list[Message] = []
self._call_count: int = 0
@property
def config(self) -> AgentConfig:
"""只读属性：外部只能读配置，不能直接修改。"""
return self._config
@property
def call_count(self) -> int:
return self._call_count
@abstractmethod
def chat(self, user_input: str) -> str:
"""子类必须实现：发送消息并返回回复。"""
...
@abstractmethod
def stream_chat(self, user_input: str) -> Iterator[str]:
"""子类必须实现：流式返回回复（逐 token）。"""
...
def add_to_history(self, role: str, content: str) -> None:
self._history.append(Message(role=role, content=content))
def clear_history(self) -> None:
self._history.clear()
def print_history(self) -> None:
print(f"\n 对话历史（{len(self._history)} 条）:")
for msg in self._history:
print(f" {msg}")
def __repr__(self) -> str:
return f"{self.__class__.__name__}(model={self._config.model}, calls={self._call_count})"
# Step 4：MockAgent 像训练用假人，不联网也能练完整流程。
class MockAgent(BaseAgent):
"""Mock Agent：用于测试，无需真实 API。"""
RESPONSES: dict[str, str] = {
"你好": "你好！我是 MockAgent，随时为你服务。",
"python": "Python 是一门优雅的语言，推荐从类型提示开始学习。",
"default": "这是一个模拟回复，用于演示 OOP 多态特性。",
}
def chat(self, user_input: str) -> str:
self._call_count += 1
self.add_to_history("user", user_input)
time.sleep(0.05) # 模拟网络延迟
# 简单关键词匹配
for keyword, response in self.RESPONSES.items():
if keyword in user_input.lower():
reply = response
break
else:
reply = self.RESPONSES["default"]
self.add_to_history("assistant", reply)
return reply
def stream_chat(self, user_input: str) -> Iterator[str]:
"""流式输出：逐词返回。"""
full_reply = self.chat(user_input)
for word in full_reply.split():
time.sleep(0.05)
yield word + " "
agent = MockAgent(AgentConfig(backend="mock", model="mock-v1"))
for question in ["你好", "讲讲 python", "随便问一句"]:
print("用户:", question)
print("助手:", agent.chat(question))
print("总调用次数:", agent.call_count)
agent.print_history()
Step 5：用 OllamaAgent 保留真实后端扩展位
痛点与机制：
OllamaAgent 展示的是“系统预留扩展口”。当前示例不真的发网络请求，而是告诉你真实项目会在这里接 requests.post(...)。这样文章既能闭环运行，又能让你看懂以后怎么接本地模型。
核心源码（逐字来自文末完整源码）：
class OllamaAgent(BaseAgent):
"""Ollama Agent：调用本地 Ollama 服务（演示版，实际需安装 ollama）。"""
def __init__(self, config: AgentConfig) -> None:
super().__init__(config)
self._base_url = "http://localhost:11434"
def chat(self, user_input: str) -> str:
self._call_count += 1
self.add_to_history("user", user_input)
# 实际调用：import requests; requests.post(f"{self._base_url}/api/chat", ...)
# 演示版：返回提示信息
reply = f"[Ollama/{self._config.model}] 需要本地运行 `ollama serve` 才能真实调用。"
self.add_to_history("assistant", reply)
return reply
def stream_chat(self, user_input: str) -> Iterator[str]:
yield self.chat(user_input)
可运行演示（补齐 Mock 数据与 print 反馈）：
import argparse
import time
import random
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Iterator
@dataclass
class Message:
role: str # "user" | "assistant" | "system"
content: str
timestamp: str = field(default_factory=lambda: datetime.now().strftime("%H:%M:%S"))
def __str__(self) -> str:
icon = {"user": "👤", "assistant": "🤖", "system": "⚙️"}.get(self.role, "?")
return f"[{self.timestamp}] {icon} {self.role}: {self.content}"
@dataclass
class AgentConfig:
backend: str
model: str
temperature: float = 0.7
max_tokens: int = 512
timeout: float = 30.0
class BaseAgent(ABC):
"""所有 Agent 必须实现的接口契约。"""
def __init__(self, config: AgentConfig) -> None:
self._config = config
self._history: list[Message] = []
self._call_count: int = 0
@property
def config(self) -> AgentConfig:
"""只读属性：外部只能读配置，不能直接修改。"""
return self._config
@property
def call_count(self) -> int:
return self._call_count
@abstractmethod
def chat(self, user_input: str) -> str:
"""子类必须实现：发送消息并返回回复。"""
...
@abstractmethod
def stream_chat(self, user_input: str) -> Iterator[str]:
"""子类必须实现：流式返回回复（逐 token）。"""
...
def add_to_history(self, role: str, content: str) -> None:
self._history.append(Message(role=role, content=content))
def clear_history(self) -> None:
self._history.clear()
def print_history(self) -> None:
print(f"\n 对话历史（{len(self._history)} 条）:")
for msg in self._history:
print(f" {msg}")
def __repr__(self) -> str:
return f"{self.__class__.__name__}(model={self._config.model}, calls={self._call_count})"
# Step 5：OllamaAgent 演示真实服务接入点，但这里仍保持零依赖闭环。
class OllamaAgent(BaseAgent):
"""Ollama Agent：调用本地 Ollama 服务（演示版，实际需安装 ollama）。"""
def __init__(self, config: AgentConfig) -> None:
super().__init__(config)
self._base_url = "http://localhost:11434"
def chat(self, user_input: str) -> str:
self._call_count += 1
self.add_to_history("user", user_input)
# 实际调用：import requests; requests.post(f"{self._base_url}/api/chat", ...)
# 演示版：返回提示信息
reply = f"[Ollama/{self._config.model}] 需要本地运行 `ollama serve` 才能真实调用。"
self.add_to_history("assistant", reply)
return reply
def stream_chat(self, user_input: str) -> Iterator[str]:
yield self.chat(user_input)
agent = OllamaAgent(AgentConfig(backend="ollama", model="llama3"))
print("Agent 表示:", agent)
print("助手回复:", agent.chat("你好"))
print("流式输出:", list(agent.stream_chat("你好")))
agent.print_history()
Step 6：用 AgentRouter 按 backend 创建不同子类，实现多态分发
痛点与机制：
AgentRouter 是多态最直观的体现。上层只说“我要 backend=mock”或者“我要 backend=ollama”，路由器负责创建对应对象。后续新增 OpenAIAgent，只要注册进去，上层调用方式不需要大改。
核心源码（逐字来自文末完整源码）：
class AgentRouter:
"""
根据配置选择 Agent 实现。
上层代码只依赖 BaseAgent 接口，不关心具体实现——这就是多态的价值。
"""
_registry: dict[str, type[BaseAgent]] = {
"mock": MockAgent,
"ollama": OllamaAgent,
}
@classmethod
def register(cls, name: str, agent_class: type[BaseAgent]) -> None:
"""开放注册：允许外部扩展新的 Agent 类型。"""
cls._registry[name] = agent_class
@classmethod
def create(cls, config: AgentConfig) -> BaseAgent:
agent_class = cls._registry.get(config.backend)
if not agent_class:
available = list(cls._registry.keys())
raise ValueError(f"未知 backend: {config.backend}，可用: {available}")
return agent_class(config)
可运行演示（补齐 Mock 数据与 print 反馈）：
import argparse
import time
import random
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Iterator
@dataclass
class Message:
role: str # "user" | "assistant" | "system"
content: str
timestamp: str = field(default_factory=lambda: datetime.now().strftime("%H:%M:%S"))
def __str__(self) -> str:
icon = {"user": "👤", "assistant": "🤖", "system": "⚙️"}.get(self.role, "?")
return f"[{self.timestamp}] {icon} {self.role}: {self.content}"
@dataclass
class AgentConfig:
backend: str
model: str
temperature: float = 0.7
max_tokens: int = 512
timeout: float = 30.0
class BaseAgent(ABC):
"""所有 Agent 必须实现的接口契约。"""
def __init__(self, config: AgentConfig) -> None:
self._config = config
self._history: list[Message] = []
self._call_count: int = 0
@property
def config(self) -> AgentConfig:
"""只读属性：外部只能读配置，不能直接修改。"""
return self._config
@property
def call_count(self) -> int:
return self._call_count
@abstractmethod
def chat(self, user_input: str) -> str:
"""子类必须实现：发送消息并返回回复。"""
...
@abstractmethod
def stream_chat(self, user_input: str) -> Iterator[str]:
"""子类必须实现：流式返回回复（逐 token）。"""
...
def add_to_history(self, role: str, content: str) -> None:
self._history.append(Message(role=role, content=content))
def clear_history(self) -> None:
self._history.clear()
def print_history(self) -> None:
print(f"\n 对话历史（{len(self._history)} 条）:")
for msg in self._history:
print(f" {msg}")
def __repr__(self) -> str:
return f"{self.__class__.__name__}(model={self._config.model}, calls={self._call_count})"
class MockAgent(BaseAgent):
"""Mock Agent：用于测试，无需真实 API。"""
RESPONSES: dict[str, str] = {
"你好": "你好！我是 MockAgent，随时为你服务。",
"python": "Python 是一门优雅的语言，推荐从类型提示开始学习。",
"default": "这是一个模拟回复，用于演示 OOP 多态特性。",
}
def chat(self, user_input: str) -> str:
self._call_count += 1
self.add_to_history("user", user_input)
time.sleep(0.05) # 模拟网络延迟
# 简单关键词匹配
for keyword, response in self.RESPONSES.items():
if keyword in user_input.lower():
reply = response
break
else:
reply = self.RESPONSES["default"]
self.add_to_history("assistant", reply)
return reply
def stream_chat(self, user_input: str) -> Iterator[str]:
"""流式输出：逐词返回。"""
full_reply = self.chat(user_input)
for word in full_reply.split():
time.sleep(0.05)
yield word + " "
class OllamaAgent(BaseAgent):
"""Ollama Agent：调用本地 Ollama 服务（演示版，实际需安装 ollama）。"""
def __init__(self, config: AgentConfig) -> None:
super().__init__(config)
self._base_url = "http://localhost:11434"
def chat(self, user_input: str) -> str:
self._call_count += 1
self.add_to_history("user", user_input)
# 实际调用：import requests; requests.post(f"{self._base_url}/api/chat", ...)
# 演示版：返回提示信息
reply = f"[Ollama/{self._config.model}] 需要本地运行 `ollama serve` 才能真实调用。"
self.add_to_history("assistant", reply)
return reply
def stream_chat(self, user_input: str) -> Iterator[str]:
yield self.chat(user_input)
# Step 6：路由器像前台分诊，根据 backend 把请求交给不同 Agent。
class AgentRouter:
"""
根据配置选择 Agent 实现。
上层代码只依赖 BaseAgent 接口，不关心具体实现——这就是多态的价值。
"""
_registry: dict[str, type[BaseAgent]] = {
"mock": MockAgent,
"ollama": OllamaAgent,
}
@classmethod
def register(cls, name: str, agent_class: type[BaseAgent]) -> None:
"""开放注册：允许外部扩展新的 Agent 类型。"""
cls._registry[name] = agent_class
@classmethod
def create(cls, config: AgentConfig) -> BaseAgent:
agent_class = cls._registry.get(config.backend)
if not agent_class:
available = list(cls._registry.keys())
raise ValueError(f"未知 backend: {config.backend}，可用: {available}")
return agent_class(config)
for backend in ["mock", "ollama"]:
config = AgentConfig(backend=backend, model="demo-model")
agent = AgentRouter.create(config)
print(f"{backend} ->", agent.__class__.__name__, "|", agent.chat("你好"))
Step 7：用 ConversationSession 管理多轮对话，并支持 len/bool/iter
痛点与机制：
ConversationSession 把对话历史从 Agent 里单独拎出来，像一个聊天窗口。__len__ 让你能用 len(session)，__bool__ 让你能判断有没有内容，__iter__ 让你能直接 for 循环遍历问答。
核心源码（逐字来自文末完整源码）：
class ConversationSession:
"""对话会话，演示魔法方法。"""
def __init__(self, agent: BaseAgent) -> None:
self._agent = agent
self._messages: list[tuple[str, str]] = []
def __len__(self) -> int:
return len(self._messages)
def __bool__(self) -> bool:
return len(self._messages) > 0
def __iter__(self) -> Iterator[tuple[str, str]]:
return iter(self._messages)
def __repr__(self) -> str:
return f"ConversationSession(agent={self._agent!r}, turns={len(self)})"
def ask(self, question: str) -> str:
reply = self._agent.chat(question)
self._messages.append((question, reply))
return reply
可运行演示（补齐 Mock 数据与 print 反馈）：
import argparse
import time
import random
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Iterator
@dataclass
class Message:
role: str # "user" | "assistant" | "system"
content: str
timestamp: str = field(default_factory=lambda: datetime.now().strftime("%H:%M:%S"))
def __str__(self) -> str:
icon = {"user": "👤", "assistant": "🤖", "system": "⚙️"}.get(self.role, "?")
return f"[{self.timestamp}] {icon} {self.role}: {self.content}"
@dataclass
class AgentConfig:
backend: str
model: str
temperature: float = 0.7
max_tokens: int = 512
timeout: float = 30.0
class BaseAgent(ABC):
"""所有 Agent 必须实现的接口契约。"""
def __init__(self, config: AgentConfig) -> None:
self._config = config
self._history: list[Message] = []
self._call_count: int = 0
@property
def config(self) -> AgentConfig:
"""只读属性：外部只能读配置，不能直接修改。"""
return self._config
@property
def call_count(self) -> int:
return self._call_count
@abstractmethod
def chat(self, user_input: str) -> str:
"""子类必须实现：发送消息并返回回复。"""
...
@abstractmethod
def stream_chat(self, user_input: str) -> Iterator[str]:
"""子类必须实现：流式返回回复（逐 token）。"""
...
def add_to_history(self, role: str, content: str) -> None:
self._history.append(Message(role=role, content=content))
def clear_history(self) -> None:
self._history.clear()
def print_history(self) -> None:
print(f"\n 对话历史（{len(self._history)} 条）:")
for msg in self._history:
print(f" {msg}")
def __repr__(self) -> str:
return f"{self.__class__.__name__}(model={self._config.model}, calls={self._call_count})"
class MockAgent(BaseAgent):
"""Mock Agent：用于测试，无需真实 API。"""
RESPONSES: dict[str, str] = {
"你好": "你好！我是 MockAgent，随时为你服务。",
"python": "Python 是一门优雅的语言，推荐从类型提示开始学习。",
"default": "这是一个模拟回复，用于演示 OOP 多态特性。",
}
def chat(self, user_input: str) -> str:
self._call_count += 1
self.add_to_history("user", user_input)
time.sleep(0.05) # 模拟网络延迟
# 简单关键词匹配
for keyword, response in self.RESPONSES.items():
if keyword in user_input.lower():
reply = response
break
else:
reply = self.RESPONSES["default"]
self.add_to_history("assistant", reply)
return reply
def stream_chat(self, user_input: str) -> Iterator[str]:
"""流式输出：逐词返回。"""
full_reply = self.chat(user_input)
for word in full_reply.split():
time.sleep(0.05)
yield word + " "
# Step 7：会话对象像聊天窗口，负责保存一轮一轮的问答。
class ConversationSession:
"""对话会话，演示魔法方法。"""
def __init__(self, agent: BaseAgent) -> None:
self._agent = agent
self._messages: list[tuple[str, str]] = []
def __len__(self) -> int:
return len(self._messages)
def __bool__(self) -> bool:
return len(self._messages) > 0
def __iter__(self) -> Iterator[tuple[str, str]]:
return iter(self._messages)
def __repr__(self) -> str:
return f"ConversationSession(agent={self._agent!r}, turns={len(self)})"
def ask(self, question: str) -> str:
reply = self._agent.chat(question)
self._messages.append((question, reply))
return reply
agent = MockAgent(AgentConfig(backend="mock", model="mock-v1"))
session = ConversationSession(agent)
print("刚创建时是否有内容:", bool(session), "长度:", len(session))
print("回复1:", session.ask("你好"))
print("回复2:", session.ask("介绍一下 Python"))
print("现在是否有内容:", bool(session), "长度:", len(session))
for question, reply in session:
print(f"问: {question} -> 答: {reply}")
print("会话表示:", session)
Step 8：用 argparse 串起配置、路由、会话和输出
痛点与机制：
main() 是脚本入口，像餐厅经理：先读命令行参数，再创建配置，再用路由器拿到 Agent，最后创建会话并跑演示问题。新手只要学会改 --backend 和 --model，就能控制整套流程。
核心源码（逐字来自文末完整源码）：
def main() -> None:
parser = argparse.ArgumentParser(description="大模型接口封装系统")
parser.add_argument("--backend", choices=["mock", "ollama"], default="mock")
parser.add_argument("--model", default="gpt-4o-mini")
parser.add_argument("--stream", action="store_true", help="流式输出模式")
args = parser.parse_args()
config = AgentConfig(backend=args.backend, model=args.model)
agent = AgentRouter.create(config)
print(f"\n 🤖 Agent 已就绪: {agent!r}")
print(f" 配置: backend={config.backend}, model={config.model}")
session = ConversationSession(agent)
questions = ["你好", "介绍一下 Python", "什么是面向对象编程"]
print(f"\n ── 对话演示 ──────────────────────────────")
for q in questions:
print(f"\n 👤 用户: {q}")
if args.stream:
print(" 🤖 助手: ", end="", flush=True)
for token in agent.stream_chat(q):
print(token, end="", flush=True)
print()
else:
reply = session.ask(q)
print(f" 🤖 助手: {reply}")
print(f"\n ── 会话统计 ──────────────────────────────")
print(f" {session!r}")
print(f" 总调用次数: {agent.call_count}")
print(f" 会话有内容: {bool(session)}")
agent.print_history()
可运行演示（补齐 Mock 数据与 print 反馈）：
import argparse
import time
import random
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Iterator
@dataclass
class Message:
role: str # "user" | "assistant" | "system"
content: str
timestamp: str = field(default_factory=lambda: datetime.now().strftime("%H:%M:%S"))
def __str__(self) -> str:
icon = {"user": "👤", "assistant": "🤖", "system": "⚙️"}.get(self.role, "?")
return f"[{self.timestamp}] {icon} {self.role}: {self.content}"
@dataclass
class AgentConfig:
backend: str
model: str
temperature: float = 0.7
max_tokens: int = 512
timeout: float = 30.0
class BaseAgent(ABC):
"""所有 Agent 必须实现的接口契约。"""
def __init__(self, config: AgentConfig) -> None:
self._config = config
self._history: list[Message] = []
self._call_count: int = 0
@property
def config(self) -> AgentConfig:
"""只读属性：外部只能读配置，不能直接修改。"""
return self._config
@property
def call_count(self) -> int:
return self._call_count
@abstractmethod
def chat(self, user_input: str) -> str:
"""子类必须实现：发送消息并返回回复。"""
...
@abstractmethod
def stream_chat(self, user_input: str) -> Iterator[str]:
"""子类必须实现：流式返回回复（逐 token）。"""
...
def add_to_history(self, role: str, content: str) -> None:
self._history.append(Message(role=role, content=content))
def clear_history(self) -> None:
self._history.clear()
def print_history(self) -> None:
print(f"\n 对话历史（{len(self._history)} 条）:")
for msg in self._history:
print(f" {msg}")
def __repr__(self) -> str:
return f"{self.__class__.__name__}(model={self._config.model}, calls={self._call_count})"
class MockAgent(BaseAgent):
"""Mock Agent：用于测试，无需真实 API。"""
RESPONSES: dict[str, str] = {
"你好": "你好！我是 MockAgent，随时为你服务。",
"python": "Python 是一门优雅的语言，推荐从类型提示开始学习。",
"default": "这是一个模拟回复，用于演示 OOP 多态特性。",
}
def chat(self, user_input: str) -> str:
self._call_count += 1
self.add_to_history("user", user_input)
time.sleep(0.05) # 模拟网络延迟
# 简单关键词匹配
for keyword, response in self.RESPONSES.items():
if keyword in user_input.lower():
reply = response
break
else:
reply = self.RESPONSES["default"]
self.add_to_history("assistant", reply)
return reply
def stream_chat(self, user_input: str) -> Iterator[str]:
"""流式输出：逐词返回。"""
full_reply = self.chat(user_input)
for word in full_reply.split():
time.sleep(0.05)
yield word + " "
class OllamaAgent(BaseAgent):
"""Ollama Agent：调用本地 Ollama 服务（演示版，实际需安装 ollama）。"""
def __init__(self, config: AgentConfig) -> None:
super().__init__(config)
self._base_url = "http://localhost:11434"
def chat(self, user_input: str) -> str:
self._call_count += 1
self.add_to_history("user", user_input)
# 实际调用：import requests; requests.post(f"{self._base_url}/api/chat", ...)
# 演示版：返回提示信息
reply = f"[Ollama/{self._config.model}] 需要本地运行 `ollama serve` 才能真实调用。"
self.add_to_history("assistant", reply)
return reply
def stream_chat(self, user_input: str) -> Iterator[str]:
yield self.chat(user_input)
class AgentRouter:
"""
根据配置选择 Agent 实现。
上层代码只依赖 BaseAgent 接口，不关心具体实现——这就是多态的价值。
"""
_registry: dict[str, type[BaseAgent]] = {
"mock": MockAgent,
"ollama": OllamaAgent,
}
@classmethod
def register(cls, name: str, agent_class: type[BaseAgent]) -> None:
"""开放注册：允许外部扩展新的 Agent 类型。"""
cls._registry[name] = agent_class
@classmethod
def create(cls, config: AgentConfig) -> BaseAgent:
agent_class = cls._registry.get(config.backend)
if not agent_class:
available = list(cls._registry.keys())
raise ValueError(f"未知 backend: {config.backend}，可用: {available}")
return agent_class(config)
class ConversationSession:
"""对话会话，演示魔法方法。"""
def __init__(self, agent: BaseAgent) -> None:
self._agent = agent
self._messages: list[tuple[str, str]] = []
def __len__(self) -> int:
return len(self._messages)
def __bool__(self) -> bool:
return len(self._messages) > 0
def __iter__(self) -> Iterator[tuple[str, str]]:
return iter(self._messages)
def __repr__(self) -> str:
return f"ConversationSession(agent={self._agent!r}, turns={len(self)})"
def ask(self, question: str) -> str:
reply = self._agent.chat(question)
self._messages.append((question, reply))
return reply
# Step 8：main 是总控台，把命令行参数变成完整对话流程。
def main() -> None:
parser = argparse.ArgumentParser(description="大模型接口封装系统")
parser.add_argument("--backend", choices=["mock", "ollama"], default="mock")
parser.add_argument("--model", default="gpt-4o-mini")
parser.add_argument("--stream", action="store_true", help="流式输出模式")
args = parser.parse_args()
config = AgentConfig(backend=args.backend, model=args.model)
agent = AgentRouter.create(config)
print(f"\n 🤖 Agent 已就绪: {agent!r}")
print(f" 配置: backend={config.backend}, model={config.model}")
session = ConversationSession(agent)
questions = ["你好", "介绍一下 Python", "什么是面向对象编程"]
print(f"\n ── 对话演示 ──────────────────────────────")
for q in questions:
print(f"\n 👤 用户: {q}")
if args.stream:
print(" 🤖 助手: ", end="", flush=True)
for token in agent.stream_chat(q):
print(token, end="", flush=True)
print()
else:
reply = session.ask(q)
print(f" 🤖 助手: {reply}")
print(f"\n ── 会话统计 ──────────────────────────────")
print(f" {session!r}")
print(f" 总调用次数: {agent.call_count}")
print(f" 会话有内容: {bool(session)}")
agent.print_history()
import sys
sys.argv = ["prog", "--backend", "mock", "--model", "mock-v1"]
main()
极客实战：完整源码与运行
现在，把上面的积木拼起来，将以下完整代码放进你的编辑器，运行它。先看整体闭环，再回头逐段改参数，你会更容易建立工程直觉。
# agent_system.py
"""
大模型接口封装系统 —— 演示 OOP 封装/继承/多态。
用法：
python3 agent_system.py
python3 agent_system.py --backend mock
python3 agent_system.py --backend ollama --model llama3
"""
import argparse
import time
import random
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Iterator
# ── 数据类：消息与响应 ────────────────────────────────────────
@dataclass
class Message:
role: str # "user" | "assistant" | "system"
content: str
timestamp: str = field(default_factory=lambda: datetime.now().strftime("%H:%M:%S"))
def __str__(self) -> str:
icon = {"user": "👤", "assistant": "🤖", "system": "⚙️"}.get(self.role, "?")
return f"[{self.timestamp}] {icon} {self.role}: {self.content}"
@dataclass
class AgentConfig:
backend: str
model: str
temperature: float = 0.7
max_tokens: int = 512
timeout: float = 30.0
# ── 抽象基类：定义接口契约 ────────────────────────────────────
class BaseAgent(ABC):
"""所有 Agent 必须实现的接口契约。"""
def __init__(self, config: AgentConfig) -> None:
self._config = config
self._history: list[Message] = []
self._call_count: int = 0
@property
def config(self) -> AgentConfig:
"""只读属性：外部只能读配置，不能直接修改。"""
return self._config
@property
def call_count(self) -> int:
return self._call_count
@abstractmethod
def chat(self, user_input: str) -> str:
"""子类必须实现：发送消息并返回回复。"""
...
@abstractmethod
def stream_chat(self, user_input: str) -> Iterator[str]:
"""子类必须实现：流式返回回复（逐 token）。"""
...
def add_to_history(self, role: str, content: str) -> None:
self._history.append(Message(role=role, content=content))
def clear_history(self) -> None:
self._history.clear()
def print_history(self) -> None:
print(f"\n 对话历史（{len(self._history)} 条）:")
for msg in self._history:
print(f" {msg}")
def __repr__(self) -> str:
return f"{self.__class__.__name__}(model={self._config.model}, calls={self._call_count})"
# ── 具体子类：Mock Agent（测试用，零网络依赖）────────────────
class MockAgent(BaseAgent):
"""Mock Agent：用于测试，无需真实 API。"""
RESPONSES: dict[str, str] = {
"你好": "你好！我是 MockAgent，随时为你服务。",
"python": "Python 是一门优雅的语言，推荐从类型提示开始学习。",
"default": "这是一个模拟回复，用于演示 OOP 多态特性。",
}
def chat(self, user_input: str) -> str:
self._call_count += 1
self.add_to_history("user", user_input)
time.sleep(0.05) # 模拟网络延迟
# 简单关键词匹配
for keyword, response in self.RESPONSES.items():
if keyword in user_input.lower():
reply = response
break
else:
reply = self.RESPONSES["default"]
self.add_to_history("assistant", reply)
return reply
def stream_chat(self, user_input: str) -> Iterator[str]:
"""流式输出：逐词返回。"""
full_reply = self.chat(user_input)
for word in full_reply.split():
time.sleep(0.05)
yield word + " "
# ── 具体子类：Ollama Agent（本地模型）────────────────────────
class OllamaAgent(BaseAgent):
"""Ollama Agent：调用本地 Ollama 服务（演示版，实际需安装 ollama）。"""
def __init__(self, config: AgentConfig) -> None:
super().__init__(config)
self._base_url = "http://localhost:11434"
def chat(self, user_input: str) -> str:
self._call_count += 1
self.add_to_history("user", user_input)
# 实际调用：import requests; requests.post(f"{self._base_url}/api/chat", ...)
# 演示版：返回提示信息
reply = f"[Ollama/{self._config.model}] 需要本地运行 `ollama serve` 才能真实调用。"
self.add_to_history("assistant", reply)
return reply
def stream_chat(self, user_input: str) -> Iterator[str]:
yield self.chat(user_input)
# ── 路由器：多态的体现 ────────────────────────────────────────
class AgentRouter:
"""
根据配置选择 Agent 实现。
上层代码只依赖 BaseAgent 接口，不关心具体实现——这就是多态的价值。
"""
_registry: dict[str, type[BaseAgent]] = {
"mock": MockAgent,
"ollama": OllamaAgent,
}
@classmethod
def register(cls, name: str, agent_class: type[BaseAgent]) -> None:
"""开放注册：允许外部扩展新的 Agent 类型。"""
cls._registry[name] = agent_class
@classmethod
def create(cls, config: AgentConfig) -> BaseAgent:
agent_class = cls._registry.get(config.backend)
if not agent_class:
available = list(cls._registry.keys())
raise ValueError(f"未知 backend: {config.backend}，可用: {available}")
return agent_class(config)
# ── 魔法方法演示 ──────────────────────────────────────────────
class ConversationSession:
"""对话会话，演示魔法方法。"""
def __init__(self, agent: BaseAgent) -> None:
self._agent = agent
self._messages: list[tuple[str, str]] = []
def __len__(self) -> int:
return len(self._messages)
def __bool__(self) -> bool:
return len(self._messages) > 0
def __iter__(self) -> Iterator[tuple[str, str]]:
return iter(self._messages)
def __repr__(self) -> str:
return f"ConversationSession(agent={self._agent!r}, turns={len(self)})"
def ask(self, question: str) -> str:
reply = self._agent.chat(question)
self._messages.append((question, reply))
return reply
# ── 主程序 ────────────────────────────────────────────────────
def main() -> None:
parser = argparse.ArgumentParser(description="大模型接口封装系统")
parser.add_argument("--backend", choices=["mock", "ollama"], default="mock")
parser.add_argument("--model", default="gpt-4o-mini")
parser.add_argument("--stream", action="store_true", help="流式输出模式")
args = parser.parse_args()
config = AgentConfig(backend=args.backend, model=args.model)
agent = AgentRouter.create(config)
print(f"\n 🤖 Agent 已就绪: {agent!r}")
print(f" 配置: backend={config.backend}, model={config.model}")
session = ConversationSession(agent)
questions = ["你好", "介绍一下 Python", "什么是面向对象编程"]
print(f"\n ── 对话演示 ──────────────────────────────")
for q in questions:
print(f"\n 👤 用户: {q}")
if args.stream:
print(" 🤖 助手: ", end="", flush=True)
for token in agent.stream_chat(q):
print(token, end="", flush=True)
print()
else:
reply = session.ask(q)
print(f" 🤖 助手: {reply}")
print(f"\n ── 会话统计 ──────────────────────────────")
print(f" {session!r}")
print(f" 总调用次数: {agent.call_count}")
print(f" 会话有内容: {bool(session)}")
agent.print_history()
if __name__ == "__main__":
main()
终端预期输出：
$ python3 agent_system.py --backend mock
🤖 Agent 已就绪: MockAgent(model=gpt-4o-mini, calls=0)
配置: backend=mock, model=gpt-4o-mini
── 对话演示 ──────────────────────────────
👤 用户: 你好
🤖 助手: 你好！我是 MockAgent，随时为你服务。
👤 用户: 介绍一下 Python
🤖 助手: Python 是一门优雅的语言，推荐从类型提示开始学习。
── 会话统计 ──────────────────────────────
ConversationSession(agent=MockAgent(model=gpt-4o-mini, calls=3), turns=3)
总调用次数: 3
会话有内容: True
NexDo Time ⚡
5 分钟极客微操：实现一个 OpenAIAgent(BaseAgent) 子类，用 requests 调用真实的 OpenAI API（https://api.openai.com/v1/chat/completions），API Key 从环境变量 OPENAI_API_KEY 读取，然后用 AgentRouter.register("openai", OpenAIAgent) 注册它。
Don’t wait for next time, do it in the next moment.