预计阅读时间:15 分钟
智能体(Agent)入门学习笔记
写在前面
从业二十余年,我见证了 AI 从符号主义到神经网络,再到今天的大模型智能体。智能体不是什么神秘概念,它本质上是一个能感知环境、做出决策、执行动作的程序实体。
让我用最朴素的方式,带你理解智能体的核心。
一、什么是智能体?
┌─────────────────────────────────┐
│ 环 境 │
│ ┌─────────────────────┐ │
│ │ 智 能 体 │ │
│ │ ┌───┐ ┌───┐ ┌───┐│ │
│ │ │感知├─→│决策├─→│执行││ │
│ │ └───┘ └───┘ └───┘│ │
│ │ ↑ ↓ │ │
│ │ ┌─┴───┴─┐ │ │
│ │ │ 记 忆 │ │ │
│ │ └───────┘ │ │
│ └─────────────────────┘ │
└─────────────────────────────────┘
三个核心问题智能体必须回答: 1. 我现在知道什么?(感知) 2. 我该怎么做?(决策) 3. 我做了什么?(执行)
二、最简智能体实现
先从最基础的开始——一个能感知温度并决策的智能体:
class ThermostatAgent:
"""恒温器智能体 - 最简单示例"""
def __init__(self, target_temp=22, tolerance=1):
self.target_temp = target_temp
self.tolerance = tolerance
self.state = "off"
def perceive(self, current_temp):
"""感知环境温度"""
return current_temp
def decide(self, perceived_temp):
"""做出决策"""
if perceived_temp < self.target_temp - self.tolerance:
return "heat_on"
elif perceived_temp > self.target_temp + self.tolerance:
return "cool_on"
else:
return "off"
def act(self, decision):
"""执行动作"""
self.state = decision
print(f"温度偏差,执行动作: {decision}")
return decision
# 使用示例
agent = ThermostatAgent(target_temp=24)
temperatures = [18, 20, 23, 25, 28, 24]
for temp in temperatures:
perceived = agent.perceive(temp)
decision = agent.decide(perceived)
agent.act(decision)
print(f"环境温度: {temp}°C → 智能体状态: {agent.state}\n")
三、带记忆的智能体
真实世界的智能体需要记住过去:
from collections import deque
from datetime import datetime
import json
class MemoryAgent:
"""带记忆的智能体"""
def __init__(self, memory_size=10):
self.short_term_memory = deque(maxlen=memory_size)
self.long_term_memory = []
self.internal_state = {"mood": "neutral", "confidence": 0.5}
def perceive(self, observation):
"""感知并存入短期记忆"""
perception = {
"timestamp": datetime.now().isoformat(),
"observation": observation,
}
self.short_term_memory.append(perception)
return perception
def reflect(self):
"""反思:将重要的短期记忆转为长期记忆"""
if len(self.short_term_memory) >= 3:
# 简单规则:检测模式
recent = list(self.short_term_memory)
observations = [r["observation"] for r in recent]
# 如果连续3次相同观察,存入长期记忆
if len(set(observations)) == 1:
memory = {
"pattern": observations[0],
"frequency": 3,
"recorded_at": datetime.now().isoformat()
}
self.long_term_memory.append(memory)
return f"发现模式: {observations[0]} 频繁出现"
return None
def decide(self, perceived):
"""基于记忆做出决策"""
obs = perceived["observation"]
# 检查长期记忆中是否有相似模式
relevant_memories = [
m for m in self.long_term_memory
if m.get("pattern") == obs
]
if relevant_memories:
self.internal_state["confidence"] = min(1.0, self.internal_state["confidence"] + 0.1)
return f"基于过去的经验,我认识这个: {obs}"
else:
self.internal_state["confidence"] = max(0.1, self.internal_state["confidence"] - 0.05)
return f"这是新情况,谨慎处理: {obs}"
def get_memory_snapshot(self):
"""获取记忆快照"""
return {
"short_term": list(self.short_term_memory),
"long_term": self.long_term_memory,
"internal_state": self.internal_state
}
# 演示
agent = MemoryAgent(memory_size=3)
observations = ["猫", "猫", "猫", "狗", "猫", "鸟"]
for obs in observations:
p = agent.perceive(obs)
decision = agent.decide(p)
reflection = agent.reflect()
print(f"观察到: {obs} → 决策: {decision}")
if reflection:
print(f" ⚡ {reflection}")
print("\n--- 记忆快照 ---")
print(json.dumps(agent.get_memory_snapshot(), indent=2, ensure_ascii=False))
四、基于 LLM 的智能体
现代智能体的核心是大语言模型。下面是一个可实际运行的 LLM Agent:
import openai
from typing import List, Dict, Any
import json
class LLMAgent:
"""基于大语言模型的智能体"""
def __init__(self, api_key: str, model: str = "gpt-3.5-turbo"):
self.client = openai.OpenAI(api_key=api_key)
self.model = model
self.conversation_history = []
self.tools = self._register_tools()
def _register_tools(self):
"""注册智能体可以使用的工具"""
return {
"calculator": {
"description": "执行数学计算",
"function": lambda expr: eval(expr),
"parameters": {"expression": "string, 数学表达式"}
},
"get_time": {
"description": "获取当前时间",
"function": lambda: datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"parameters": {}
},
"search_memory": {
"description": "搜索对话历史",
"function": self._search_history,
"parameters": {"keyword": "string, 搜索关键词"}
}
}
def _search_history(self, keyword: str) -> List[str]:
"""在对话历史中搜索"""
return [
msg["content"] for msg in self.conversation_history
if keyword.lower() in msg["content"].lower()
]
def _build_system_prompt(self) -> str:
"""构建系统提示词"""
tools_desc = "\n".join([
f"- {name}: {info['description']}"
for name, info in self.tools.items()
])
return f"""你是一个智能助手,可以使用以下工具完成任务:
{tools_desc}
当你需要使用工具时,请按以下 JSON 格式回复:
{{"tool": "工具名", "parameters": {{"参数名": "参数值"}}}}}
如果不需要使用工具,请直接回复用户。
"""
def think(self, user_input: str) -> Dict[str, Any]:
"""智能体思考过程"""
messages = [
{"role": "system", "content": self._build_system_prompt()},
*self.conversation_history[-10:], # 保留最近10条对话
{"role": "user", "content": user_input}
]
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0.7
)
content = response.choices[0].message.content
# 尝试解析工具调用
try:
if content.strip().startswith("{"):
tool_call = json.loads(content)
if "tool" in tool_call:
return {"type": "tool_call", "data": tool_call}
except:
pass
return {"type": "text", "data": content}
def execute_tool(self, tool_name: str, parameters: Dict) -> str:
"""执行工具调用"""
if tool_name not in self.tools:
return f"错误:未知工具 {tool_name}"
tool = self.tools[tool_name]
try:
result = tool["function"](**parameters)
return str(result)
except Exception as e:
return f"工具执行错误: {e}"
def run(self, user_input: str) -> str:
"""运行智能体"""
# 思考
thought = self.think(user_input)
# 如果是工具调用
if thought["type"] == "tool_call":
tool_call = thought["data"]
tool_result = self.execute_tool(
tool_call["tool"],
tool_call.get("parameters", {})
)
# 将工具结果作为新的输入,再次思考
final_response = self.think(
f"工具执行结果:{tool_result}\n请根据这个结果回答用户。"
)
response_text = final_response["data"]
else:
response_text = thought["data"]
# 更新对话历史
self.conversation_history.append({"role": "user", "content": user_input})
self.conversation_history.append({"role": "assistant", "content": response_text})
return response_text
# 使用示例(需要 OpenAI API Key)
"""
agent = LLMAgent(api_key="your-api-key")
# 对话
print(agent.run("现在几点了?"))
print(agent.run("计算 123 * 456"))
print(agent.run("我刚才问过什么关于时间的问题?"))
"""
五、多智能体协作
单个智能体能力有限,真正的威力在于协作:
from abc import ABC, abstractmethod
from typing import List, Dict, Any
import asyncio
class BaseAgent(ABC):
"""智能体基类"""
def __init__(self, name: str, role: str):
self.name = name
self.role = role
self.memory = []
self.peers = {}
@abstractmethod
async def process(self, message: Dict[str, Any]) -> Dict[str, Any]:
"""处理消息"""
pass
async def send(self, target: str, message: Dict[str, Any]):
"""发送消息给其他智能体"""
if target in self.peers:
response = await self.peers[target].process(message)
return response
return {"error": f"Unknown target: {target}"}
def remember(self, key: str, value: Any):
"""记忆存储"""
self.memory.append({"key": key, "value": value, "timestamp": datetime.now()})
class ResearcherAgent(BaseAgent):
"""研究员智能体 - 负责收集信息"""
async def process(self, message: Dict[str, Any]) -> Dict[str, Any]:
task = message.get("task", "")
if task == "research":
topic = message.get("topic", "")
# 模拟研究过程
await asyncio.sleep(1) # 模拟耗时操作
result = {
"status": "success",
"data": f"关于 '{topic}' 的研究资料:...(已收集5篇相关文献)",
"sources": 5
}
self.remember("last_research", result)
return result
return {"status": "error", "message": f"Unknown task: {task}"}
class AnalystAgent(BaseAgent):
"""分析师智能体 - 负责分析数据"""
async def process(self, message: Dict[str, Any]) -> Dict[str, Any]:
task = message.get("task", "")
if task == "analyze":
data = message.get("data", "")
# 模拟分析过程
await asyncio.sleep(0.5)
analysis = {
"status": "success",
"insights": [
f"从 '{data}' 中发现3个关键洞察",
"趋势显示增长态势",
"建议进一步深入研究"
],
"confidence": 0.85
}
self.remember("last_analysis", analysis)
return analysis
return {"status": "error", "message": f"Unknown task: {task}"}
class CoordinatorAgent(BaseAgent):
"""协调员智能体 - 协调其他智能体工作"""
async def process(self, message: Dict[str, Any]) -> Dict[str, Any]:
task = message.get("task", "")
if task == "execute_project":
topic = message.get("topic", "")
# 第一步:研究
print(f"[协调员] 委派研究任务给研究员...")
research_result = await self.send("researcher", {
"task": "research",
"topic": topic
})
# 第二步:分析
print(f"[协调员] 委派分析任务给分析师...")
analysis_result = await self.send("analyst", {
"task": "analyze",
"data": research_result.get("data", "")
})
# 整合结果
final_report = {
"topic": topic,
"research": research_result,
"analysis": analysis_result,
"summary": f"项目 '{topic}' 已完成,收集{research_result['sources']}个来源,"
f"分析置信度{analysis_result['confidence']}"
}
return {"status": "success", "report": final_report}
return {"status": "error", "message": f"Unknown task: {task}"}
class MultiAgentSystem:
"""多智能体系统"""
def __init__(self):
self.agents = {}
def register_agent(self, agent: BaseAgent):
"""注册智能体"""
self.agents[agent.name] = agent
# 建立智能体间的连接
for name, other in self.agents.items():
if name != agent.name:
agent.peers[name] = other
other.peers[agent.name] = agent
async def run(self, task: Dict[str, Any]):
"""运行系统"""
coordinator = self.agents.get("coordinator")
if coordinator:
result = await coordinator.process(task)
return result
return {"error": "No coordinator found"}
# 使用示例
async def demo_multi_agent():
# 创建智能体
researcher = ResearcherAgent("researcher", "研究员")
analyst = AnalystAgent("analyst", "分析师")
coordinator = CoordinatorAgent("coordinator", "协调员")
# 组建系统
system = MultiAgentSystem()
system.register_agent(researcher)
system.register_agent(analyst)
system.register_agent(coordinator)
# 执行任务
result = await system.run({
"task": "execute_project",
"topic": "AI在医疗领域的应用"
})
print("\n=== 最终报告 ===")
print(json.dumps(result, indent=2, ensure_ascii=False))
# 运行演示
# asyncio.run(demo_multi_agent())
六、智能体设计核心原则
1. 感知-决策-执行循环
class AgentLoop:
async def run(self):
while self.is_active:
perception = await self.perceive() # 感知
decision = await self.think(perception) # 决策
result = await self.act(decision) # 执行
await self.learn(perception, decision, result) # 学习
2. 状态管理
- 短期状态:当前任务的上下文
- 长期状态:知识和经验积累
- 元状态:智能体自身的目标和约束
3. 工具使用能力
智能体的能力边界由工具决定: - 计算工具(解决数学问题) - 搜索工具(获取实时信息) - 代码执行(自动化操作) - API调用(与外部系统交互)
4. 记忆系统设计
工作记忆 ←→ 情景记忆 ←→ 语义记忆
↓ ↓ ↓
当前任务 过往经历 知识图谱
七、实战建议
- 从简单开始:先用规则引擎,需要时再引入 LLM
- 日志是关键:记录每一步的输入输出,便于调试
- 优雅降级:LLM 不可用时,要有备选方案
- 安全边界:工具调用需要权限控制和白名单
- 成本控制:缓存常见问题,减少 API 调用
写在最后
智能体的本质,是让程序具备「目标导向」的自主行为能力。不要被各种复杂概念吓到——回退到最基础,它就是一个 while 循环里的感知、思考、行动。
真正优秀的智能体设计,不在于用了多复杂的模型,而在于对场景的深刻理解和简洁有效的抽象。
希望这篇笔记能帮你迈出智能体开发的第一步。二十年前我也曾困惑于此,而今回头看,大道至简。
笔记作者:一位老工程师
最后更新:2024年
本文由 尚先生 原创,转载请注明出处。
评论
0