智能体入门

智能体 2026-04-20 12
预计阅读时间: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. 记忆系统设计

工作记忆 ←→ 情景记忆 ←→ 语义记忆
   ↓           ↓           ↓
 当前任务    过往经历    知识图谱

七、实战建议

  1. 从简单开始:先用规则引擎,需要时再引入 LLM
  2. 日志是关键:记录每一步的输入输出,便于调试
  3. 优雅降级:LLM 不可用时,要有备选方案
  4. 安全边界:工具调用需要权限控制和白名单
  5. 成本控制:缓存常见问题,减少 API 调用

写在最后

智能体的本质,是让程序具备「目标导向」的自主行为能力。不要被各种复杂概念吓到——回退到最基础,它就是一个 while 循环里的感知、思考、行动。

真正优秀的智能体设计,不在于用了多复杂的模型,而在于对场景的深刻理解和简洁有效的抽象。

希望这篇笔记能帮你迈出智能体开发的第一步。二十年前我也曾困惑于此,而今回头看,大道至简。


笔记作者:一位老工程师
最后更新:2024年


本文由 尚先生 原创,转载请注明出处。

📖相关推荐

评论

0
暂无评论,来发表第一条评论吧

发表评论

登录 后发表评论