欢迎来到「Python教程从零基础到实战」系列的第23期！
在前面的系列中，我们学习了Python基础语法、数据结构、面向对象编程、装饰器与生成器、爬虫、数据分析、Web后端开发、AI实战入门（LangChain）、前端入门、项目实战、Docker容器化部署、数据库进阶、异步编程、日志系统、自动化运维、单元测试、微服务、CICD与DevOps、并发多线程、包管理发布、以及性能优化和设计模式等主题。
这一期，我们要聊一个非常酷的话题——**AI Agent（智能体）**。
## 什么是 AI Agent？
你可能听说过 ChatGPT、Claude 这些大语言模型（LLM）。它们能聊天、能写代码、能画画——但它们本身是"被动"的：你问一句，它答一句，不会主动做事。
而 AI Agent 不同。它是一个**能自主感知环境、做出决策、采取行动的智能程序**。简单说：
-**LLM** = 会思考的大脑
-**Agent** = 大脑 + 手脚 + 记忆 = 能真正帮你干活
举个例子：
> 你说："帮我查一下北京今天的天气，然后告诉我需不需要带伞。"
>
>- 普通 LLM：只能回答"我不知道实时天气数据"。
>- AI Agent：先调用天气 API 获取数据 → 分析结果 → 判断是否需要带伞 → 给你回复："今天北京多云，气温18度，没有降水概率，不用带伞。"
看到了吗？Agent 的关键能力在于**使用工具**和**自主规划任务**。
---
## 第一部分：Agent 的核心架构
一个典型的 AI Agent 由以下几个部分组成：
1.**大脑（Language Model）**：负责推理和决策
2.**记忆（Memory）**：记住之前的对话和历史
3.**工具（Tools）**：比如搜索、发邮件、调 API
4.**规划（Planning）**：将复杂任务拆解成步骤
这就是经典的 **ReAct** 框架——Reasoning（推理）+ Acting（行动）。
### 核心循环
```
用户输入 → Agent思考 → 决定调用哪个工具 → 执行工具 → 观察结果 → 再次思考 → ... → 最终回答
```
每个循环都是一次"思考—行动—观察"的过程。
---
## 第二部分：用 Python 搭建第一个简单的 Agent
现在我们动手来写。为了简化，我们先用纯 Python + requests 来实现，不依赖任何第三方框架。
### 示例一：天气查询 Agent
假设我们有一个免费的天气 API，我们可以这样实现：
```python
import json
import requests
classSimpleWeatherAgent:
"""一个简单的天气查询智能体"""
def__init__(self):
self.memory = [] # 记忆：存储对话历史
self.api_base = "https://api.open-meteo.com/v1/forecast"
defremember(self, role, content):
"""记录对话"""
self.memory.append({"role": role, "content": str(content)})
defget_weather(self, city):
"""调用天气 API 获取数据"""
city_coords = {
"北京": (39.9042, 116.4074),
"上海": (31.2304, 121.4737),
"广州": (23.1291, 113.2644),
"深圳": (22.5431, 114.0579),
"成都": (30.5728, 104.0668),
"杭州": (30.2741, 120.1551),
"武汉": (30.5928, 114.3055),
}
if city notin city_coords:
returnf"抱歉，我暂时不支持城市 '{city}' 的天气查询。"
lat, lon = city_coords[city]
params = {
"latitude": lat,
"longitude": lon,
"current": "temperature_2m,relative_humidity_2m,precipitation_probability",
"timezone": "Asia/Shanghai",
}
try:
resp = requests.get(self.api_base, params=params, timeout=10)
data = resp.json()
current = data.get("current", {})
temp = current.get("temperature_2m", "未知")
humidity = current.get("relative_humidity_2m", "未知")
rain_prob = current.get("precipitation_probability", "未知")
result = f"{city}当前气温{temp}°C，湿度{humidity}%，降雨概率{rain_prob}%。"
# 判断是否带伞
if rain_prob and rain_prob > 50:
result += "建议带伞！"
elif rain_prob and rain_prob <= 50:
result += "应该不用带伞。"
else:
result += "暂无降雨数据，建议你出门留意天空。"
return result
exceptExceptionas e:
returnf"查询天气时出了点问题：{e}"
defthink_and_act(self, user_input):
"""Agent 的"思考"：解析用户意图并执行行动"""
user_input_lower = user_input.lower()
# 简单的意图识别
for keyword in ["天气", "气温", "温度", "下雨", "带伞"]:
if keyword in user_input_lower:
# 尝试提取城市名
cities = ["北京", "上海", "广州", "深圳", "成都", "杭州", "武汉"]
city = "北京"# 默认
for c in cities:
if c in user_input:
city = c
break
self.remember("user", user_input)
self.remember("assistant", user_input)
result = self.get_weather(city)
self.remember("assistant", result)
return result
return"我能帮你查天气哦！比如问我'北京今天天气怎么样？'"
# 使用示例
if__name__ == "__main__":
agent = SimpleWeatherAgent()
print("=== AI 天气助手 ===")
print(agent.think_and_act("北京天气怎么样？"))
print(agent.think_and_act("上海需要带伞吗？"))
print(agent.think_and_act("你好，你能做什么？"))
# 查看记忆
print("\n=== 对话记忆 ===")
for msg in agent.memory:
prefix = "👤"if msg["role"] == "user"else"🤖"
print(f"{prefix}{msg['content']}")
```
运行这段代码，你就能得到一个具备记忆能力的简单天气 Agent。
### 示例二：多工具 Agent
现在我们把 Agent 的能力扩展一下——让它不仅能查天气，还能做数学运算和时间查询：
```python
import json
import math
import requests
from datetime import datetime
classMultiToolAgent:
"""具有多个工具的 Agent"""
def__init__(self):
self.memory = []
self.tools = {
"天气查询": self._weather_tool,
"数学计算": self._math_tool,
"时间查询": self._time_tool,
}
defremember(self, role, content):
self.memory.append({"role": role, "content": str(content)})
# ---------- 工具函数 ----------
def_weather_tool(self, query):
"""天气查询工具"""
city_map = {"北京": (39.9042, 116.4074)}
lat, lon = city_map.get(query.strip(), (39.9042, 116.4074))
url = "https://api.open-meteo.com/v1/forecast"
params = {"latitude": lat, "longitude": lon, "current": "temperature_2m"}
try:
r = requests.get(url, params=params, timeout=10).json()
temp = r["current"]["temperature_2m"]
returnf"工具结果: 北京当前气温 {temp}°C"
exceptExceptionas e:
returnf"工具结果: 天气查询失败 - {e}"
def_math_tool(self, query):
"""数学计算工具"""
query = query.strip()
try:
# 安全地只允许数学表达式
allowed_names = {k: v for k, v in math.__dict__.items()
ifnot k.startswith("_")}
result = eval(query, {"__builtins__": {}}, allowed_names)
returnf"工具结果: {query} = {result}"
exceptExceptionas e:
returnf"工具结果: 计算失败 - {e}"
def_time_tool(self, query):
"""时间查询工具"""
now = datetime.now().strftime("%Y年%m月%d日 %H:%M:%S")
returnf"工具结果: 现在是 {now}"
# ---------- Agent 核心 ----------
defparse_query(self, raw_input):
"""简单的意图识别，决定调用哪个工具"""
raw_lower = raw_input.lower()
ifany(w in raw_lower for w in ["天气", "气温", "温度"]):
return"天气查询", raw_input.strip()
elifany(w in raw_lower for w in ["等于", "计算", "平方根", "log", "sin"]):
return"数学计算", raw_input.strip()
elifany(w in raw_lower for w in ["几点了", "什么时间", "日期", "今天"]):
return"时间查询", raw_input.strip()
returnNone, raw_input
defthink_and_act(self, user_input):
"""Agent 主循环：思考 -> 选工具 -> 执行 -> 回答"""
self.remember("user", user_input)
tool_name, tool_arg = self.parse_query(user_input)
if tool_name and tool_name inself.tools:
self.remember("assistant", f"[思考] 调用工具: {tool_name}")
result = self.tools[tool_name](tool_arg)
self.remember("assistant", result)
return result
else:
answer = f"我没有找到合适的工具来处理你的请求。我能做的是：{', '.join(self.tools.keys())}"
self.remember("assistant", answer)
return answer
# 使用示例
if__name__ == "__main__":
agent = MultiToolAgent()
print("=== 多工具 AI Agent ===")
print(agent.think_and_act("北京天气怎么样"))
print(agent.think_and_act("sin(45)等于多少"))
print(agent.think_and_act("现在几点了"))
print(agent.think_and_act("帮我订个餐厅"))
```
---
## 第三部分：接入大语言模型——真正的 Agent
上面的例子用了硬编码的规则来做意图识别。真正的 Agent 应该是**让 LLM 自己来决定调用什么工具**。
### 3.0 前置准备：安装依赖和环境配置
```bash
pipinstallopenairequestspython-dotenvbeautifulsoup4
```
在项目根目录创建 `.env` 文件：
```ini
OPENAI_API_KEY="sk-your-api-key-here"
OPENAI_BASE_URL="https://api.your-provider.com/v1"
```
>**提示**：这里使用的 `openai` SDK 兼容所有遵循 OpenAI API 格式的服务商，包括通义千问、Moonshot、智谱等国内模型提供商。替换 `OPENAI_BASE_URL` 即可切换模型。
### 3.1 工具注册机制——让 LLM"知道"有什么工具可用
LLM 本身不知道你能做什么。你需要通过 **Function Calling（函数调用）** 接口告诉它：
```python
import os
import json
from datetime import datetime
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv()
client = OpenAI(
api_key=os.getenv("OPENAI_API_KEY"),
base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
)
classToolRegistry:
"""工具注册中心——LLM 通过这里发现所有可用工具"""
def__init__(self):
self.tools = {}
defregister(self, name, description, parameters, func):
"""注册一个工具"""
self.tools[name] = {
"name": name,
"description": description,
"parameters": parameters,
"func": func,
}
defget_openai_tools(self):
"""转换为 OpenAI Function Calling 格式"""
openai_tools = []
for tool inself.tools.values():
openai_tools.append({
"type": "function",
"function": {
"name": tool["name"],
"description": tool["description"],
"parameters": tool["parameters"],
},
})
return openai_tools
# 创建一个工具注册表
registry = ToolRegistry()
```
### 3.2 定义具体工具
```python
import requests
import math
@registry.register(
name="get_weather",
description="查询指定城市的当前天气信息",
parameters={
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，如 '北京'、'上海'"
}
},
"required": ["city"]
},
func=lambdaargs: _get_weather_impl(args.get("city", ""))
)
defweather_tool_stub(*args, **kwargs):
pass# 装饰器已处理注册
def_get_weather_impl(city):
"""实际天气查询实现"""
city_coords = {
"北京": (39.9042, 116.4074),
"上海": (31.2304, 121.4737),
"广州": (23.1291, 113.2644),
"深圳": (22.5431, 114.0579),
}
if city notin city_coords:
returnf"暂不支持城市: {city}"
lat, lon = city_coords[city]
try:
r = requests.get(
"https://api.open-meteo.com/v1/forecast",
params={
"latitude": lat,
"longitude": lon,
"current": "temperature_2m,relative_humidity_2m,weather_code"
},
timeout=10
).json()
cur = r["current"]
returnf"{city}: 气温{cur['temperature_2m']}°C, 湿度{cur['relative_humidity_2m']}%"
exceptExceptionas e:
returnf"天气查询失败: {e}"
@registry.register(
name="calculation",
description="执行数学表达式计算",
parameters={
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "要计算的数学表达式，如 'sqrt(144)' 或 'sin(3.14/2)'"
}
},
"required": ["expression"]
},
func=lambdaargs: _calc_impl(args.get("expression", ""))
)
defcalc_tool_stub(*args, **kwargs):
pass
def_calc_impl(expression):
allowed = {k: v for k, v in math.__dict__.items() ifnot k.startswith("_")}
try:
result = eval(expression, {"__builtins__": {}}, allowed)
returnf"{expression} = {result}"
exceptExceptionas e:
returnf"计算失败: {e}"
@registry.register(
name="get_current_time",
description="获取当前日期和时间",
parameters={
"type": "object",
"properties": {}
},
func=lambdaargs: datetime.now().strftime("%Y-%m-%d %H:%M:%S")
)
deftime_tool_stub(*args, **kwargs):
pass
```
### 3.3 Agent 主循环：让 LLM 自己决定调用哪个工具
```python
system_prompt = """你是一个有帮助的 AI 助手。
你可以使用各种工具来帮助用户。当需要获取实时数据时，主动调用合适的工具。
先分析用户意图，如果需要工具则调用工具，否则直接回答。"""
classLLMAgent:
"""基于 LLM Function Calling 的 Agent"""
def__init__(self, model="gpt-4o-mini"):
self.model = model
self.messages = [{"role": "system", "content": system_prompt}]
defchat(self, user_message):
"""发送一轮对话，自动处理工具调用"""
# 1. 记录用户消息
self.messages.append({"role": "user", "content": user_message})
# 2. 调用 LLM，传入工具定义
response = client.chat.completions.create(
model=self.model,
messages=self.messages,
tools=registry.get_openai_tools(),
max_tokens=500,
)
response_message = response.choices[0].message
self.messages.append(response_message.model_dump())
# 3. 检查是否需要调用工具
if response_message.tool_calls:
# 4. 执行 LLM 选择的工具
for tool_call in response_message.tool_calls:
fn_name = tool_call.function.name
fn_args = json.loads(tool_call.function.arguments)
print(f" 🛠️ LLM 调用工具: {fn_name}({fn_args})")
if fn_name in registry.tools:
result = registry.tools[fn_name]["func"](fn_args)
else:
result = f"未知工具: {fn_name}"
# 5. 将工具结果回传给 LLM
self.messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"name": fn_name,
"content": result,
})
# 6. 带着工具结果再次请求 LLM 生成最终回答
final_response = client.chat.completions.create(
model=self.model,
messages=self.messages,
max_tokens=1000,
)
answer = final_response.choices[0].message.content
self.messages.append({"role": "assistant", "content": answer})
return answer
# 不需要工具，直接返回 LLM 的回答
return response_message.content or"(无回复)"
# ========== 运行示例 ==========
if__name__ == "__main__":
agent = LLMAgent()
print("=== LLM 多工具 Agent ===\n")
queries = [
"北京现在天气怎么样？",
"请帮我计算 sin(3.14/2) 等于多少",
"现在几点了？",
]
for q in queries:
print(f"👤 你: {q}")
print(f"🤖 Agent: {agent.chat(q)}\n")
```
**运行流程示意：**
```
👤 你: 北京现在天气怎么样？
🛠️ LLM 调用工具: get_weather({'city': '北京'})
🤖 Agent: 北京当前气温约 18°C，湿度 45%。今天气温舒适，适合出行。
--- 完整对话记忆 ---
[{"role":"system","content":"你是一个有帮助的 AI 助手..."},
{"role":"user","content":"北京现在天气怎么样？"},
{"role":"assistant","tool_calls":[{"function":{"name":"get_weather","arguments":"{\"city\":\"北京\"}}"}}],
{"role":"tool","tool_call_id":"tc_xxx","name":"get_weather","content":"北京: 气温18°C, 湿度45%"},
{"role":"assistant","content":"北京当前气温约 18°C，湿度 45%。今天气温舒适..."}}]
```
这就是真正的 Agent：它**不是**靠硬编码规则来选择工具，而是通过 LLM 理解你的意图后**自主决定**该调用什么工具。
---
## 第四部分：进阶——带记忆的对话 Agent
前面的示例中，Agent 只能处理单轮对话。现实里，用户经常会给 Agent 一系列连续指令：
> "查一下北京的天气" → "那上海呢？" → "对比一下这两个城市"
这就需要 Agent 具备**长期记忆**。我们自己实现一个更轻量的版本：
### 4.1 记忆管理器
```python
import hashlib
from datetime import datetime
classMemoryManager:
"""管理 Agent 的记忆——包括短期对话和长期摘要"""
def__init__(self, max_history=10):
self.max_history = max_history
self.conversation_history = [] # 短期记忆
self.summary = ""# 长期记忆摘要
self.knowledge_base = {} # 事实性知识
defadd_turn(self, role, content):
"""添加一轮对话到记忆"""
self.conversation_history.append({
"id": hashlib.md5(f"{role}:{content}".encode()).hexdigest()[:8],
"role": role,
"content": content,
"timestamp": datetime.now().isoformat(),
})
# 保持历史长度，防止无限增长
iflen(self.conversation_history) > self.max_history * 2:
self.conversation_history = self.conversation_history[-self.max_history * 2:]
defsummarize_past(self):
"""将早期对话压缩为摘要——模拟人类"回忆"的方式"""
iflen(self.conversation_history) < 6:
returnNone
early_turns = self.conversation_history[:4]
summary_parts = []
for turn in early_turns:
if turn["role"] == "user":
summary_parts.append(f"- 用户问了: {turn['content']}")
else:
summary_parts.append(f"- 助手回答了: {turn['content'][:50]}...")
self.summary = "📋 之前的对话摘要:\n" + "\n".join(summary_parts)
# 移除已摘要的对话，释放上下文空间
self.conversation_history = self.conversation_history[4:]
returnself.summary
defadd_fact(self, fact_type, key, value):
"""添加一条事实性知识"""
if fact_type notinself.knowledge_base:
self.knowledge_base[fact_type] = {}
self.knowledge_base[fact_type][key] = value
defrecall_fact(self, fact_type, key):
"""回忆事实"""
returnself.knowledge_base.get(fact_type, {}).get(key)
defget_memory_context(self):
"""返回完整的记忆上下文（注入到 prompt 中）"""
parts = []
ifself.summary:
parts.append(self.summary)
ifself.conversation_history:
parts.append("\n📝 最近对话:")
for turn inself.conversation_history[-4:]:
icon = "👤"if turn["role"] == "user"else"🤖"
parts.append(f" {icon}{turn['content']}")
return"\n".join(parts) if parts else"(暂无记忆)"
# 测试记忆管理器
if__name__ == "__main__":
mem = MemoryManager(max_history=10)
mem.add_turn("user", "北京天气怎么样？")
mem.add_turn("assistant", "北京当前气温 18°C")
mem.add_turn("user", "那上海呢？")
mem.add_turn("assistant", "上海当前气温 22°C")
mem.add_fact("user_preference", "home_city", "北京")
mem.add_fact("user_preference", "language", "中文")
print(mem.get_memory_context())
print("\n--- 回忆事实 ---")
print(f"用户家乡: {mem.recall_fact('user_preference', 'home_city')}")
```
### 4.2 集成到 Agent
```python
memory_system_prompt = """你是一个有帮助的 AI 助手，具备对话记忆能力。
根据用户的消息和历史记忆，使用工具获取实时数据，然后给出有用的回答。"""
classConversationalAgent(LLMAgent):
"""带记忆的对话 Agent"""
def__init__(self, model="gpt-4o-mini"):
super().__init__(model)
# 覆盖 system prompt
self.messages[0] = {"role": "system", "content": memory_system_prompt}
self.memory = MemoryManager(max_history=10)
defchat(self, user_message):
"""带记忆增强的对话"""
# 如果历史太长，先做摘要压缩
iflen(self.memory.conversation_history) > 8:
summary = self.memory.summarize_past()
if summary:
user_message = f"[系统提示]\n{summary}\n\n[新消息]\n{user_message}"
# 记录到记忆管理器
self.memory.add_turn("user", user_message)
# 构建消息列表
messages = list(self.messages) # 拷贝
messages.append({"role": "user", "content": user_message})
# 调用 LLM
response = client.chat.completions.create(
model=self.model,
messages=messages,
tools=registry.get_openai_tools(),
max_tokens=500,
)
response_message = response.choices[0].message
answer = ""
if response_message.tool_calls:
self.memory.add_turn("assistant", f"🛠️ 调用工具: {response_message.tool_calls[0].function.name}")
# 执行工具调用
for tc in response_message.tool_calls:
fn_name = tc.function.name
fn_args = json.loads(tc.function.arguments)
if fn_name in registry.tools:
result = registry.tools[fn_name]["func"](fn_args)
else:
result = f"未知工具: {fn_name}"
self.memory.add_turn("tool", f"{fn_name} 结果: {result}")
# 用记忆中的最近对话构建新消息，让 LLM 生成最终回答
recent_msgs = [{"role": "system", "content": memory_system_prompt}]
for m inself.memory.conversation_history[-6:]:
recent_msgs.append({"role": m["role"], "content": m["content"]})
final = client.chat.completions.create(
model=self.model,
messages=recent_msgs,
max_tokens=1000,
)
answer = final.choices[0].message.content
else:
answer = response_message.content or"(无回复)"
self.memory.add_turn("assistant", answer)
return answer
if__name__ == "__main__":
agent = ConversationalAgent()
print("=== 带记忆的多轮对话 Agent ===\n")
session = [
"帮我查一下北京的天气",
"那上海呢？",
"刚才北京和上海的气温差了多少？请计算",
]
for msg in session:
print(f"👤 你: {msg}")
print(f"🤖 Agent: {agent.chat(msg)}\n")
print("-" * 50)
```
**核心改进：**
| 特性 | 无记忆 Agent | 带记忆 Agent |
|------|-------------|-------------|
| 单轮对话 | ✅ | ✅ |
| 多轮引用（"那上海呢"） | ❌ 不知道指代 | ✅ 从历史中找到前文 |
| 对话压缩 | N/A | ✅ 早期对话自动摘要化 |
| 事实记忆 | N/A | ✅ 可记住用户偏好、关键数据 |
---
## 第五部分：用 LangChain 快速搭建 Agent
现实中，我们不需要每次都从零写。**LangChain** 提供了开箱即用的 Agent 框架，大幅降低了开发成本。
### 5.1 安装依赖
```bash
pipinstalllangchain-openailangchain-coretavily-python
```
### 5.2 一行代码创建一个 Agent
```python
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.tools.tavily_search import TavilySearchResults
from dotenv import load_dotenv
import os
load_dotenv()
# 1. 选择模型
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 2. 定义工具列表
tools = [
TavilySearchResults(max_results=2), # 网络搜索工具
]
# 3. 定义 prompt——告诉 Agent 该怎么说话
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个乐于助人的 AI 助手。使用工具来获取最新信息。"),
("placeholder", "{chat_history}"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
# 4. 组装 Agent
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True, # 打印推理过程
handle_parsing_errors=True,
max_iterations=5, # 最多迭代 5 次
)
# 5. 运行！
result = agent_executor.invoke({"input": "2026年最新的AI发展趋势是什么？"})
print(result["output"])
```
### 5.3 自定义工具
你可以轻松地把任意 Python 函数变成 LangChain 工具：
```python
from langchain_core.tools import tool
@tool
defget_stock_price(stock_symbol: str) -> str:
"""获取指定股票的当前价格"""
prices = {"AAPL": "195.50 USD", "TSLA": "248.30 USD", "BTC-USDT": "68500 CNY"}
return prices.get(stock_symbol, f"股票 {stock_symbol} 价格暂不可用")
@tool
defsend_email(to: str, subject: str, body: str) -> str:
"""给指定邮箱发送邮件"""
# 对接邮件服务的逻辑...
returnf"✅ 邮件已发送至 {to}，主题: {subject}"
# 注册到 Agent
tools = [TavilySearchResults(max_results=2), get_stock_price, send_email]
agent = create_tool_calling_agent(llm, tools, prompt)
```
### 5.4 ReAct 格式的 Agent 思维链
设置 `verbose=True` 后，你会看到 Agent 的完整思考过程：
```
> Entering new AgentExecutor chain...
Invoking: get_weather with {'city': '北京'}
[天气结果: 北京气温 18°C]
[LLM 结合结果生成自然语言回答]
> Finished chain.
```
每一步都在"思考—行动—观察"—这就是 ReAct 框架的实际运作方式。
---
## 第六部分：实战项目——个人 AI 助手
让我们把所学知识整合起来，做一个综合性的个人 AI 助手：
```python
"""
PersonalAssistant Agent — 第 23 期最终项目
功能：网络搜索 · 记笔记 · 列笔记 · 设提醒
"""
import json
import os
import requests
from datetime import datetime, timedelta
from pathlib import Path
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv()
client = OpenAI(
api_key=os.getenv("OPENAI_API_KEY"),
base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
)
DATA_DIR = Path.home() / ".personal_assistant_data"
DATA_DIR.mkdir(exist_ok=True)
# ========== 工具层 ==========
defsearch_web(query: str, max_results: int = 3) -> str:
"""使用 DuckDuckGo 进行网络搜索"""
try:
from bs4 import BeautifulSoup
encoded = requests.utils.quote(query)
r = requests.get(
f"https://html.duckduckgo.com/html/?q={encoded}",
headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}
)
soup = BeautifulSoup(r.text, "html.parser")
results = []
for a in soup.select("a.result__a")[:max_results]:
results.append(a.get_text().strip())
return"\n".join(results) or"没有找到结果。"
exceptImportError:
return"需要安装 BeautifulSoup: pip install beautifulsoup4"
exceptExceptionas e:
returnf"搜索出错: {e}"
defadd_note(category: str, content: str) -> str:
"""添加一条笔记到本地文件"""
note_file = DATA_DIR / f"{category}.json"
notes = json.loads(note_file.read_text(encoding="utf-8")) if note_file.exists() else []
notes.append({
"time": datetime.now().isoformat(),
"content": content,
})
note_file.write_text(json.dumps(notes, ensure_ascii=False, indent=2), encoding="utf-8")
returnf"✅ 笔记已保存到 '{category}'，共 {len(notes)} 条。"
deflist_notes(category: str = "") -> str:
"""列出笔记"""
ifnot category:
files = list(DATA_DIR.glob("*.json"))
ifnot files:
return"暂无笔记。"
returnf"已有类别: {[f.stem for f in files]}"
note_file = DATA_DIR / f"{category}.json"
ifnot note_file.exists():
returnf"类别 '{category}' 不存在。"
notes = json.loads(note_file.read_text(encoding="utf-8"))
ifnot notes:
returnf"'{category}' 下暂无笔记。"
lines = [f" [{n['time']}] {n['content']}"for n in notes]
return"\n".join(lines)
defset_reminder(message: str, minutes: int = 1) -> str:
"""设置定时提醒（本地存储，记录到期时间）"""
due = datetime.now() + timedelta(minutes=minutes)
reminder_file = DATA_DIR / "_reminders.json"
reminders = json.loads(reminder_file.read_text(encoding="utf-8")) if reminder_file.exists() else []
reminders.append({
"message": message,
"due": due.isoformat(),
"done": False,
})
reminder_file.write_text(
json.dumps(reminders, ensure_ascii=False, indent=2), encoding="utf-8"
)
returnf"⏰ 已设提醒: {message}（预计 {minutes} 分钟后到期）"
# ========== Agent 主类 ==========
classPersonalAssistant:
"""个人 AI 助手——集成多个工具，让 LLM 自主决策"""
TOOL_DEFS = [
{
"type": "function",
"function": {
"name": "web_search",
"description": "搜索互联网获取最新信息",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "add_note",
"description": "添加一条笔记到指定类别",
"parameters": {
"type": "object",
"properties": {
"category": {"type": "string", "description": "笔记类别，如 '待办'、'灵感'"},
"content": {"type": "string", "description": "笔记内容"}
},
"required": ["category", "content"]
}
}
},
{
"type": "function",
"function": {
"name": "list_notes",
"description": "列出所有笔记或指定类别下的笔记",
"parameters": {
"type": "object",
"properties": {
"category": {"type": "string", "description": "可选，指定类别名"}
}
}
}
},
{
"type": "function",
"function": {
"name": "set_reminder",
"description": "设置定时提醒",
"parameters": {
"type": "object",
"properties": {
"message": {"type": "string", "description": "提醒内容"},
"minutes": {"type": "integer", "description": "多少分钟后提醒", "default": 1}
},
"required": ["message"]
}
}
},
]
TOOL_MAP = {
"web_search": lambda **kw: search_web(kw.get("query", "")),
"add_note": lambda **kw: add_note(kw.get("category", ""), kw.get("content", "")),
"list_notes": lambda **kw: list_notes(kw.get("category", "")),
"set_reminder": lambda **kw: set_reminder(kw.get("message", ""), kw.get("minutes", 1)),
}
SYSTEM_PROMPT = """你是一个智能个人助手，名字叫 Nova。
你善于使用工具来完成用户的各种需求——搜索信息、记笔记、设提醒等。
当用户的问题需要实时数据时，调用工具；否则直接回答。
请用中文回答用户。"""
def__init__(self, model="gpt-4o-mini"):
self.model = model
self.messages = [{"role": "system", "content": self.SYSTEM_PROMPT}]
defrun(self, user_input: str) -> str:
"""执行一轮对话"""
self.messages.append({"role": "user", "content": user_input})
resp = client.chat.completions.create(
model=self.model,
messages=self.messages,
tools=self.TOOL_DEFS,
max_tokens=500,
)
rm = resp.choices[0].message
self.messages.append(rm.model_dump())
# 如果需要调用工具
if rm.tool_calls:
for tc in rm.tool_calls:
fn = tc.function.name
args = json.loads(tc.function.arguments)
print(f" 🛠️ Nova → {fn}({args})")
result = self.TOOL_MAP[fn](**args)
self.messages.append({
"role": "tool",
"tool_call_id": tc.id,
"name": fn,
"content": result,
})
# 再次请求 LLM 生成自然语言回答
final = client.chat.completions.create(
model=self.model,
messages=self.messages,
max_tokens=1000,
)
answer = final.choices[0].message.content
self.messages.append({"role": "assistant", "content": answer})
return answer
# 不需要工具，直接返回
return rm.content or"(Nova 没有回复)"
# ========== 交互终端 ==========
if__name__ == "__main__":
print("=" * 50)
print(" 🤖 Nova — 你的个人 AI 助手")
print(" 支持: 搜索 · 记笔记 · 列笔记 · 设提醒")
print(" 输入 'quit' 退出，'clear' 清空对话")
print("=" * 50)
bot = PersonalAssistant()
whileTrue:
try:
text = input("\n👤 你: ").strip()
except (EOFError, KeyboardInterrupt):
break
ifnot text:
continue
if text.lower() in ("quit", "exit", "退出"):
print("👋 Nova: 再见！欢迎下次使用。")
break
if text.lower() in ("clear", "清空"):
bot.messages = [bot.messages[0]] # 保留 system prompt
print("🧹 对话已清空。")
continue
answer = bot.run(text)
print(f"\n🤖 Nova: {answer}")
```
**运行效果：**
```
==================================================
🤖 Nova — 你的个人 AI 助手
支持: 搜索 · 记笔记 · 列笔记 · 设提醒
输入 'quit' 退出，'clear' 清空对话
==================================================
👤 你: 帮我查一下 2026年AI的发展趋势
🛠️ Nova → web_search({'query': '2026年AI的发展趋势'})
🤖 Nova: 根据搜索结果，2026年AI的主要趋势包括：
1. 多模态模型的普及
2. AI Agent的兴起
3. 边缘AI计算的增长
...
👤 你: 把这些要点记录到"灵感"类别
🛠️ Nova → add_note({'category': '灵感', 'content': '2026年AI的主要趋势：多模态、Agent、边缘计算'})
🤖 Nova: ✅ 笔记已保存到 '灵感'，共 1 条。
👤 你: 列出我的所有笔记
🤖 Nova: 已有类别: ['灵感']
```
---
## 第七部分：Agent 的未来方向
学完本期内容后，你可能会对以下方向感兴趣：
### 7.1 多 Agent 协作
单个 Agent 能力有限，多个 Agent 可以分工合作：
```
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Researcher │────▶│ Writer │────▶│ Reviewer │
│ (搜索研究) │ │ (撰写文档) │ │ (审核修改) │
└─────────────┘ └─────────────┘ └─────────────┘
```
这种"多智能体系统"已经被用于代码审查、内容创作流水线等场景。
### 7.2 规划与推理
高级 Agent 不仅会"反应"，还会"规划"：
```
用户: 帮我做一个市场调研报告
Agent 的规划：
1. [搜索] 收集行业数据
2. [搜索] 收集竞品信息
3. [计算] 分析市场数据
4. [写作] 生成报告初稿
5. [校验] 检查数据一致性
6. [输出] 返回完整报告
```
这就是所谓的 **ReAct** 和 **Plan-and-Execute** 范式的演进。
### 7.3 主流 Agent 框架对比
| 框架 | 特点 | 适合场景 |
|------|------|---------|
| **LangChain** | 生态最丰富，工具最全 | 通用 Agent 开发 |
| **LlamaIndex** | 专注于数据索引和RAG | 知识库问答 |
| **AutoGen (微软)** | 多 Agent 协作 | 复杂任务分解 |
| **CrewAI** | 角色扮演的多 Agent 系统 | 团队协作模拟 |
| **自研** | 完全可控 | 对定制化要求高的场景 |
---
## 第八部分：总结与练习
### 本节核心知识点回顾
```
┌──────────────────────────────────────────────────┐
│ AI Agent 核心组件 │
├──────────────────────────────────────────────────┤
│ 1. LLM — 推理大脑 │
│ 2. Tools — 工具（API/函数） │
│ 3. Function Calling — 让 LLM 选择工具 │
│ 4. Memory — 对话记忆 │
│ 5. ReAct Loop — 思考→行动→观察 循环 │
│ 6. Framework — LangChain / 自研 │
└──────────────────────────────────────────────────┘
```
### 课后练习
1.**基础题**：给自己的天气 Agent 增加一个"汇率查询"工具，调用自由外汇 API。
2.**提高题**：实现一个支持中文拼音到城市的自动映射，让用户输入 "beijing" 也能查到北京天气。
3.**综合题**：基于 LangChain 框架，搭建一个能够搜索网页 + 查天气 + 做计算的个人助手，加入多轮对话记忆能力。
> 恭喜你完成了第23期！如果你对这个教程有任何建议或问题，欢迎反馈。
> 我们下期再见！🚀
