尧图网站建设 
在探索AI技术落地的过程中你是否曾设想过一个能够像人类一样通过真实世界的交互和反馈来持续学习、进化的智能体传统的AI模型往往依赖于静态、封闭的数据集进行训练一旦部署其能力便基本固化。而一个能够从真实世界经验中学习的AI Agent则代表了迈向通用人工智能AGI的更近一步。本文将深入探讨如何构建一个具备“从经验中学习”能力的AI Agent涵盖其核心架构、关键技术实现并提供一个从零开始的实战项目手把手带你搭建一个能够通过与环境互动来优化自身决策的智能体。本文适合对AI Agent开发感兴趣的中高级开发者无论你是想深入理解智能体的学习机制还是希望将动态学习能力集成到自己的项目中都能从中获得一套完整的、可落地的技术方案。我们将从概念解析入手逐步深入到记忆系统、强化学习集成、环境模拟等核心模块的代码实现。1. AI Agent与经验学习核心概念解析在开始构建之前我们首先需要明确几个关键概念这有助于理解我们所要构建系统的本质。1.1 什么是AI AgentAI Agent智能体并非一个全新的概念。简单来说它是一个能够感知环境、自主决策并执行行动以实现特定目标的软件实体。与传统的程序不同AI Agent的核心在于其自主性和适应性。感知PerceptionAgent通过传感器或API接口从环境如网页、数据库、软件界面获取信息。决策Decision-making基于感知到的信息、内置的目标以及已有的知识模型决定采取什么行动。行动Action执行决策例如点击按钮、调用函数、输出文本等从而对环境产生影响。目标Goal驱动Agent所有行为的最终目的。我们常见的ChatGPT等大语言模型LLM可以看作是Agent的“大脑”或“决策核心”但它们本身不具备完整的Agent能力。一个完整的AI Agent系统需要将LLM与工具Tools、记忆Memory和规划Planning等组件结合起来。1.2 “从真实世界经验中学习”意味着什么这是本文的重点。所谓“从真实世界经验中学习”指的是Agent的能力不局限于其初始训练数据。在部署后它能够通过与环境互动执行行动并观察结果。接收反馈获取行动成功或失败的信号奖励或惩罚。更新内部状态根据反馈调整其策略、知识或模型参数使得未来的决策更有可能获得成功。这本质上是一个在线学习Online Learning或强化学习Reinforcement Learning, RL的过程。与监督学习需要大量标注数据不同这种学习方式更接近人类或动物的学习模式。1.3 关键组件记忆系统与学习引擎为了实现经验学习我们的Agent需要两大支柱记忆系统Memory System负责存储和检索经验。这包括短期记忆如对话上下文和长期记忆。长期记忆用于存储历史交互状态、行动、奖励、结果是学习的基础。记忆优化如向量检索、经验回放缓冲区至关重要。学习引擎Learning Engine负责从记忆中提取经验并更新Agent的决策模型。这可以是基于规则的策略调整也可以是与微调LLM或训练一个价值/策略网络相结合的复杂过程。2. 环境准备与核心技术栈在开始编码前我们需要搭建开发环境并选择合适的技术栈。本项目将采用Python作为主要语言因为它拥有最丰富的AI和机器学习库生态。2.1 基础环境与版本说明操作系统macOS / Linux (推荐) 或 Windows (WSL2)。Python版本 3.10。包管理工具pip或conda。核心依赖库LangChain / LangGraph用于构建Agent框架集成工具调用、记忆和流程编排。本文将使用更新更灵活的LangGraph。OpenAI API或本地LLM作为Agent的“大脑”。为方便演示我们使用OpenAI GPT-4 API。你也可以使用Ollama部署本地模型如Llama 3。强化学习库gymnasium(原OpenAI Gym) 用于创建模拟环境stable-baselines3或ray[rllib]用于高级RL算法但本文为简化会实现一个基础的学习逻辑。向量数据库用于实现长期记忆的快速语义检索。chromadb轻量且易用。其他工具库requests(网络请求)numpy(数值计算)pydantic(数据验证)。2.2 项目初始化与依赖安装首先创建项目目录并初始化虚拟环境。# 创建项目目录 mkdir learning_ai_agent cd learning_ai_agent # 创建虚拟环境 (以conda为例) conda create -n learning_agent python3.10 -y conda activate learning_agent # 安装核心依赖 pip install langgraph langchain-openai chromadb gymnasium requests numpy pydantic # 如果你使用OpenAI API需要设置API Key # export OPENAI_API_KEYyour-api-key-here创建项目基础结构learning_ai_agent/ ├── agent/ │ ├── __init__.py │ ├── core.py # Agent核心类定义 │ ├── memory.py # 记忆系统实现 │ └── learning.py # 学习引擎实现 ├── environment/ │ ├── __init__.py │ └── sim_env.py # 模拟环境定义 ├── config.py # 配置文件 ├── main.py # 主运行脚本 ├── requirements.txt └── README.md生成requirements.txt文件pip freeze requirements.txt3. 构建核心组件记忆系统与模拟环境一个能够学习的Agent需要一个记录经验的地方和一个可供交互的“世界”。我们先从这两个基础组件开始。3.1 实现长期记忆系统我们将实现一个基于chromadb的向量记忆存储它能够存储每次交互的“经验元组”状态、行动、奖励、新状态并支持根据当前状态语义检索相似历史经验。# agent/memory.py import chromadb from chromadb.config import Settings from typing import List, Dict, Any, Optional, Tuple import uuid import numpy as np class ExperienceMemory: 经验记忆系统存储 (state, action, reward, next_state, done) 元组。 支持基于state的语义相似性检索。 def __init__(self, collection_name: str agent_experiences, persist_directory: str ./chroma_db): # 初始化Chroma客户端持久化存储 self.client chromadb.PersistentClient(pathpersist_directory, settingsSettings(anonymized_telemetryFalse)) # 获取或创建集合 self.collection self.client.get_or_create_collection(namecollection_name) def add_experience(self, state: str, action: str, reward: float, next_state: str, done: bool, metadata: Optional[Dict] None): 添加一条经验到记忆库。 # 将经验组合成一段文本用于嵌入 experience_text fState: {state}\nAction: {action}\nReward: {reward}\nNext State: {next_state}\nDone: {done} doc_id str(uuid.uuid4()) # 准备元数据 exp_metadata { state: state, action: action, reward: reward, next_state: next_state, done: done, **(metadata or {}) } # 添加到集合 self.collection.add( documents[experience_text], metadatas[exp_metadata], ids[doc_id] ) return doc_id def retrieve_similar_experiences(self, query_state: str, n_results: int 5) - List[Dict[str, Any]]: 根据当前状态检索相似的历史经验。 results self.collection.query( query_texts[fState: {query_state}], n_resultsn_results ) retrieved_exps [] if results[metadatas]: for meta_list in results[metadatas]: for meta in meta_list: retrieved_exps.append(meta) return retrieved_exps def get_all_experiences(self) - List[Dict[str, Any]]: 获取所有经验仅用于调试或简单学习。 all_data self.collection.get() experiences [] for i, id_ in enumerate(all_data[ids]): exp {k: all_data[k][i] for k in [metadatas, documents] if k in all_data} experiences.append(exp) return experiences3.2 创建一个简单的模拟环境为了演示学习过程我们构建一个极简的文本环境一个“智能恒温器”环境。Agent的目标是通过调节“加热”或“冷却”动作将房间温度维持在舒适的22°C。# environment/sim_env.py import random from typing import Tuple, Dict, Any class ThermostatEnvironment: 一个简单的恒温器模拟环境。 def __init__(self, initial_temp: float 20.0, target_temp: float 22.0, noise_std: float 0.5): self.current_temp initial_temp self.target_temp target_temp self.noise_std noise_std self.time_step 0 self.max_steps 20 def reset(self) - str: 重置环境到初始状态。 self.current_temp 20.0 random.uniform(-2, 2) self.time_step 0 return self._get_state_observation() def _get_state_observation(self) - str: 返回当前状态的文本描述。 return fTime: {self.time_step}, Current Temperature: {self.current_temp:.2f}°C, Target: {self.target_temp}°C def step(self, action: str) - Tuple[str, float, bool, Dict[str, Any]]: 执行一个动作。 :param action: “heat”, “cool”, 或 “nothing” :return: (next_state, reward, done, info) self.time_step 1 # 环境动力学动作影响温度 if action heat: self.current_temp 1.5 random.normalvariate(0, self.noise_std) elif action cool: self.current_temp - 1.5 random.normalvariate(0, self.noise_std) elif action nothing: pass else: raise ValueError(fUnknown action: {action}) # 添加自然波动 self.current_temp random.normalvariate(0, 0.3) # 计算奖励越接近目标奖励越高 temp_diff abs(self.current_temp - self.target_temp) reward -temp_diff # 负差异作为奖励最大化奖励即最小化差异 # 判断是否结束 done self.time_step self.max_steps next_state self._get_state_observation() info {temperature: self.current_temp, action_taken: action} return next_state, reward, done, info4. 构建具备学习能力的AI Agent核心现在我们将LangGraph与记忆系统、学习引擎结合构建Agent的核心逻辑。Agent的决策将基于LLM但会参考历史经验。4.1 定义Agent状态与工具首先我们使用Pydantic定义Agent的状态并创建它可以调用的工具即动作。# agent/core.py from typing import TypedDict, Annotated, List, Dict, Any import operator from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, SystemMessage from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field import json # 定义Agent的持久化状态 class AgentState(TypedDict): Agent在运行过程中的状态。 messages: Annotated[List[Any], operator.add] # 对话消息历史 current_state: str # 环境当前状态描述 total_reward: float # 本轮累计奖励 steps: int # 本轮已执行步数 memory_context: List[Dict] # 从记忆库检索到的相关经验 # 定义可供Agent调用的工具动作 class AgentTools: staticmethod def act_heat() - str: 执行加热动作。 return heat staticmethod def act_cool() - str: 执行冷却动作。 return cool staticmethod def act_nothing() - str: 什么也不做。 return nothing # 一个用于从记忆库查询经验的工具 staticmethod def query_memory(query: str, memory_system) - List[Dict]: 查询记忆系统中与当前情况相似的经验。 return memory_system.retrieve_similar_experiences(query, n_results3)4.2 实现学习引擎学习引擎是一个相对独立模块它定期例如每轮结束后从记忆系统中采样一批经验并更新一个简单的策略表或提供反思总结。# agent/learning.py from .memory import ExperienceMemory from typing import List, Dict, Any import random class SimpleLearningEngine: 一个简单的基于经验回放的学习引擎。 def __init__(self, memory: ExperienceMemory): self.memory memory def learn_from_experience(self, batch_size: int 10) - Dict[str, Any]: 从记忆中随机采样一批经验进行学习。 这里实现一个简单的策略统计在相似状态下哪个动作带来的平均奖励最高。 all_exps self.memory.get_all_experiences() if not all_exps or len(all_exps) 5: return {message: Not enough experiences to learn from.} # 简化处理我们只分析元数据 experiences [] for exp in all_exps: if metadatas in exp and exp[metadatas]: experiences.extend(exp[metadatas]) if len(experiences) batch_size: batch experiences else: batch random.sample(experiences, batch_size) # 构建一个简单的策略统计表state - {action: (total_reward, count)} policy_stats {} for exp in batch: state exp.get(state, ) action exp.get(action, ) reward float(exp.get(reward, 0)) if state not in policy_stats: policy_stats[state] {} if action not in policy_stats[state]: policy_stats[state][action] {total_reward: 0, count: 0} policy_stats[state][action][total_reward] reward policy_stats[state][action][count] 1 # 计算平均奖励并找出每个状态下最好的动作 learned_policy {} for state, actions in policy_stats.items(): best_action None best_avg_reward -float(inf) for action, stats in actions.items(): avg_reward stats[total_reward] / stats[count] if avg_reward best_avg_reward: best_avg_reward avg_reward best_action action if best_action: learned_policy[state] best_action # 将学习到的策略总结成文本可供LLM参考 summary Learned from past experiences:\n for state, action in list(learned_policy.items())[:5]: # 只取前5条 summary f- When in state {state[:50]}..., the action {action} tended to yield good results.\n return { summary: summary, policy_sample: learned_policy, experiences_used: len(batch) }4.3 组装智能体与运行图我们将使用LangGraph来定义Agent的决策流程。这个流程包括观察状态、检索记忆、调用LLM决策、执行动作、存储经验。# agent/core.py (续) from langchain.tools import Tool from langgraph.prebuilt import ToolExecutor from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, SystemMessage from langchain_openai import ChatOpenAI def create_learning_agent(memory_system, learning_engine, env): 创建并返回一个具备学习能力的Agent图。 # 1. 初始化LLM llm ChatOpenAI(modelgpt-4o-mini, temperature0.1) # 使用更小更快的模型 # 2. 绑定工具 tools [ Tool(nameAct_Heat, funclambda x: AgentTools.act_heat(), descriptionIncrease the temperature.), Tool(nameAct_Cool, funclambda x: AgentTools.act_cool(), descriptionDecrease the temperature.), Tool(nameAct_Nothing, funclambda x: AgentTools.act_nothing(), descriptionDo nothing.), Tool(nameQuery_Memory, funclambda q: AgentTools.query_memory(q, memory_system), descriptionQuery past similar experiences from memory. Input should be a description of the current state.) ] tool_executor ToolExecutor(tools) # 3. 定义节点函数 def observe_state(state: AgentState) - AgentState: 节点观察环境状态并查询相关记忆。 current_state state[current_state] # 查询记忆 similar_exps memory_system.retrieve_similar_experiences(current_state, n_results2) state[memory_context] similar_exps return state def decide_action(state: AgentState) - AgentState: 节点LLM基于状态和记忆做出决策。 system_prompt SystemMessage(contentf 你是一个智能恒温器控制Agent。你的目标是通过选择heat、cool或nothing动作将房间温度维持在22°C附近。 当前环境状态{state[current_state]} 以下是过去类似情况下的经验仅供参考 {json.dumps(state.get(memory_context, []), indent2)} 请仔细分析当前状态和历史经验然后只输出一个JSON对象格式如下 {{reasoning: 你的简要推理过程, action: heat|cool|nothing, confidence: 0.0-1.0}} ) human_msg HumanMessage(content请决定下一步动作。) response llm.invoke([system_prompt, human_msg]) try: decision json.loads(response.content) except json.JSONDecodeError: decision {reasoning: Failed to parse, action: nothing, confidence: 0.5} # 将决策存入消息历史便于追溯 state[messages].append(HumanMessage(contentfDecision: {decision})) state[_last_decision] decision return state def execute_action(state: AgentState) - AgentState: 节点执行选定的动作并更新环境。 decision state.get(_last_decision, {action: nothing}) action decision.get(action, nothing) # 在真实环境中执行动作 next_state, reward, done, info env.step(action) # 存储经验到长期记忆 memory_system.add_experience( statestate[current_state], actionaction, rewardreward, next_statenext_state, donedone, metadata{step: state[steps], reasoning: decision.get(reasoning, )} ) # 更新Agent状态 state[current_state] next_state state[total_reward] reward state[steps] 1 state[_last_reward] reward state[_done] done state[messages].append(HumanMessage(contentfExecuted {action}. Reward: {reward:.2f}. New state: {next_state})) return state def should_continue(state: AgentState) - str: 条件边判断是否继续执行。 if state.get(_done, False) or state[steps] 20: return end return continue # 4. 构建图 workflow StateGraph(AgentState) workflow.add_node(observe, observe_state) workflow.add_node(decide, decide_action) workflow.add_node(act, execute_action) workflow.set_entry_point(observe) workflow.add_edge(observe, decide) workflow.add_edge(decide, act) workflow.add_conditional_edges( act, should_continue, { continue: observe, # 继续下一轮观察-决策-行动循环 end: END } ) return workflow.compile()5. 完整实战运行与评估学习型Agent现在我们将所有组件串联起来运行一个完整的实验观察Agent如何通过多轮交互进行学习。5.1 主运行脚本# main.py import asyncio from agent.memory import ExperienceMemory from agent.learning import SimpleLearningEngine from agent.core import create_learning_agent from environment.sim_env import ThermostatEnvironment from langgraph.checkpoint.memory import MemorySaver async def main(): print( 启动具备经验学习能力的AI Agent实验 \n) # 1. 初始化组件 print(初始化记忆系统、学习引擎和环境...) memory ExperienceMemory() learner SimpleLearningEngine(memory) env ThermostatEnvironment() # 2. 创建Agent print(构建智能体工作流...) app create_learning_agent(memory, learner, env) # 3. 运行多轮实验Episodes num_episodes 5 # 运行5轮 checkpoint_memory MemorySaver() # 用于保存Agent的对话状态可选 for episode in range(num_episodes): print(f\n--- 第 {episode 1} 轮实验开始 ---) # 重置环境 initial_state env.reset() total_reward 0.0 # 准备Agent的初始状态 initial_config { configurable: { thread_id: fepisode_{episode}, } } initial_agent_state { messages: [], current_state: initial_state, total_reward: 0.0, steps: 0, memory_context: [] } # 运行Agent直到本轮结束 print(f初始状态: {initial_state}) for step in range(20): # 最大步数 # 使用LangGraph的astream_events可以观察执行过程这里简化使用同步调用 output await app.ainvoke(initial_agent_state, configinitial_config) initial_agent_state output # 更新状态 step_reward initial_agent_state.get(_last_reward, 0) total_reward step_reward current_state initial_agent_state[current_state] print(f 步骤{step1}: 状态{current_state}, 单步奖励{step_reward:.2f}) if initial_agent_state.get(_done, False): print(f 环境提前终止。) break print(f第 {episode 1} 轮结束累计奖励: {total_reward:.2f}) # 4. 每轮结束后让学习引擎从积累的经验中学习 if episode 0: # 从第二轮开始学习 print( 正在进行经验学习...) learning_result learner.learn_from_experience(batch_size8) print(f 学习摘要: {learning_result.get(summary, N/A)[:200]}...) # 短暂暂停模拟真实时间间隔 await asyncio.sleep(0.5) # 5. 实验结束后展示最终的学习成果 print(\n 实验总结 ) final_learn learner.learn_from_experience(batch_size20) print(最终学习策略样本) for state, action in list(final_learn.get(policy_sample, {}).items())[:3]: print(f 状态 {state[:30]}... - 推荐动作 {action}) print(f\n总经验条数: {len(memory.get_all_experiences())}) print(实验完成) if __name__ __main__: asyncio.run(main())5.2 运行结果与分析运行上述脚本(python main.py)你将看到类似以下的输出 启动具备经验学习能力的AI Agent实验 初始化记忆系统、学习引擎和环境... 构建智能体工作流... --- 第 1 轮实验开始 --- 初始状态: Time: 0, Current Temperature: 19.35°C, Target: 22°C 步骤1: 状态Time: 1, Current Temperature: 20.72°C, Target: 22°C, 单步奖励-1.28 步骤2: 状态Time: 2, Current Temperature: 22.15°C, Target: 22°C, 单步奖励-0.15 ... 第 1 轮结束累计奖励: -15.32 --- 第 2 轮实验开始 --- 初始状态: Time: 0, Current Temperature: 21.10°C, Target: 22°C 正在进行经验学习... 学习摘要: Learned from past experiences: - When in state Time: 0, Current Temperature: 19.35°C, Target: 22°C..., the action heat tended to yield good results. ...结果分析第一轮Agent像一张白纸主要依靠LLM的常识进行决策例如“温度低于目标就加热”累计奖励可能较低。后续轮次学习引擎开始工作从记忆库中提取历史经验并总结出简单的策略如“在XX状态下加热动作效果更好”。这些总结会被注入到后续轮次LLM的系统提示中从而影响其决策。趋势随着轮次增加累计奖励应该呈现上升趋势因为Agent决策时参考了更多成功的经验避免了重复过去的错误。这就是“从经验中学习”的直观体现。6. 常见问题与排查思路在构建和运行此类AI Agent时你可能会遇到以下典型问题问题现象可能原因排查思路与解决方案Agent决策循环卡住或不执行动作1. LLM API调用失败或超时。2. LangGraph图定义有误边未正确连接。3.should_continue逻辑始终返回”continue”。1. 检查网络和API密钥添加重试和超时机制。2. 使用workflow.get_graph().draw_mermaid()输出图结构检查。3. 打印_done和steps变量确保终止条件被触发。记忆检索返回空或无关结果1. ChromaDB集合未正确持久化或路径错误。2. 经验文本嵌入相似度低。3. 存储的经验数量太少。1. 检查persist_directory路径权限确认数据已保存。2. 优化经验文本的格式化方式使其包含更关键的语义信息。3. 增加初始随机探索积累更多经验数据。累计奖励没有提升甚至下降1. 学习引擎过于简单未能提取有效模式。2. 环境奖励函数设计不合理。3. LLM未能有效利用记忆上下文。1. 升级学习引擎引入更复杂的RL算法如DQN。2. 重新设计奖励函数使其梯度更平滑更能引导学习。3. 优化提示词明确要求LLM参考提供的“Learned from past experiences”。运行速度非常慢1. 每步都调用LLM成本高且慢。2. ChromaDB检索大量数据。3. 同步调用阻塞。1. 对稳定策略进行缓存或使用更小、更快的模型。2. 为记忆集合建立索引限制每次检索数量。3. 使用异步调用 (ainvoke)。Agent行为不稳定决策随机1. LLM的temperature参数过高。2. 环境随机噪声过大。3. 记忆中的经验相互矛盾。1. 将temperature调低如0.1使输出更确定。2. 适当减小环境中的随机噪声标准差。3. 在学习引擎中引入置信度过滤只采纳被多次验证的经验。7. 进阶优化与工程最佳实践要将这个原型发展为可投入生产或研究的系统需要考虑以下优化方向和实践建议7.1 记忆系统优化分层记忆区分短期工作记忆当前任务上下文和长期经验记忆。短期记忆可用普通列表长期记忆用向量数据库。记忆摘要不是存储所有原始经验而是定期由LLM生成摘要性知识如“在清晨温度较低时立即加热效果最佳”再存入记忆。这能减少存储和检索负担。优先级经验回放像深度强化学习那样为那些带来高奖励或高学习价值的经验赋予更高采样权重。7.2 学习引擎强化集成标准RL算法将stable-baselines3中的PPO、DQN等算法与Agent结合。让LLM负责高层策略生成RL模型负责低层价值评估。LLM微调Fine-tuning收集高质量的成功轨迹状态-动作对用于微调一个专门的“策略LLM”使其初始性能更好。反思与元认知让Agent在任务失败或完成后主动调用LLM进行反思分析错误原因并将反思结论存入记忆实现更高级的元学习。7.3 工程化与部署配置化管理将模型类型、API密钥、环境参数、学习超参数等全部外置到config.yaml文件中。日志与监控记录每一步的决策、奖励、LLM调用消耗便于分析和调试。可使用structlog或prometheus。模块化设计确保记忆、学习、环境、Agent核心等模块之间接口清晰便于单独替换或升级例如将ChromaDB换成Pinecone将OpenAI换成Claude。测试与评估建立独立的测试环境集定义清晰的评估指标如平均奖励、收敛速度、任务成功率定期运行基准测试。构建一个能从真实世界经验中学习的AI Agent是一个持续迭代的过程。本文提供的框架是一个强大的起点它融合了LLM的推理能力、记忆系统的持久化以及学习引擎的适应性。你可以在此基础上将其应用到更复杂的场景如网页自动化、客户服务对话优化、游戏AI甚至机器人控制中。记住核心在于设计一个能让Agent安全、高效地与环境交互并接收反馈的闭环系统。 从零构建具备经验学习能力的AI Agent：核心架构与实战指南
在探索AI技术落地的过程中你是否曾设想过一个能够像人类一样通过真实世界的交互和反馈来持续学习、进化的智能体传统的AI模型往往依赖于静态、封闭的数据集进行训练一旦部署其能力便基本固化。而一个能够从真实世界经验中学习的AI Agent则代表了迈向通用人工智能AGI的更近一步。本文将深入探讨如何构建一个具备“从经验中学习”能力的AI Agent涵盖其核心架构、关键技术实现并提供一个从零开始的实战项目手把手带你搭建一个能够通过与环境互动来优化自身决策的智能体。本文适合对AI Agent开发感兴趣的中高级开发者无论你是想深入理解智能体的学习机制还是希望将动态学习能力集成到自己的项目中都能从中获得一套完整的、可落地的技术方案。我们将从概念解析入手逐步深入到记忆系统、强化学习集成、环境模拟等核心模块的代码实现。1. AI Agent与经验学习核心概念解析在开始构建之前我们首先需要明确几个关键概念这有助于理解我们所要构建系统的本质。1.1 什么是AI AgentAI Agent智能体并非一个全新的概念。简单来说它是一个能够感知环境、自主决策并执行行动以实现特定目标的软件实体。与传统的程序不同AI Agent的核心在于其自主性和适应性。感知PerceptionAgent通过传感器或API接口从环境如网页、数据库、软件界面获取信息。决策Decision-making基于感知到的信息、内置的目标以及已有的知识模型决定采取什么行动。行动Action执行决策例如点击按钮、调用函数、输出文本等从而对环境产生影响。目标Goal驱动Agent所有行为的最终目的。我们常见的ChatGPT等大语言模型LLM可以看作是Agent的“大脑”或“决策核心”但它们本身不具备完整的Agent能力。一个完整的AI Agent系统需要将LLM与工具Tools、记忆Memory和规划Planning等组件结合起来。1.2 “从真实世界经验中学习”意味着什么这是本文的重点。所谓“从真实世界经验中学习”指的是Agent的能力不局限于其初始训练数据。在部署后它能够通过与环境互动执行行动并观察结果。接收反馈获取行动成功或失败的信号奖励或惩罚。更新内部状态根据反馈调整其策略、知识或模型参数使得未来的决策更有可能获得成功。这本质上是一个在线学习Online Learning或强化学习Reinforcement Learning, RL的过程。与监督学习需要大量标注数据不同这种学习方式更接近人类或动物的学习模式。1.3 关键组件记忆系统与学习引擎为了实现经验学习我们的Agent需要两大支柱记忆系统Memory System负责存储和检索经验。这包括短期记忆如对话上下文和长期记忆。长期记忆用于存储历史交互状态、行动、奖励、结果是学习的基础。记忆优化如向量检索、经验回放缓冲区至关重要。学习引擎Learning Engine负责从记忆中提取经验并更新Agent的决策模型。这可以是基于规则的策略调整也可以是与微调LLM或训练一个价值/策略网络相结合的复杂过程。2. 环境准备与核心技术栈在开始编码前我们需要搭建开发环境并选择合适的技术栈。本项目将采用Python作为主要语言因为它拥有最丰富的AI和机器学习库生态。2.1 基础环境与版本说明操作系统macOS / Linux (推荐) 或 Windows (WSL2)。Python版本 3.10。包管理工具pip或conda。核心依赖库LangChain / LangGraph用于构建Agent框架集成工具调用、记忆和流程编排。本文将使用更新更灵活的LangGraph。OpenAI API或本地LLM作为Agent的“大脑”。为方便演示我们使用OpenAI GPT-4 API。你也可以使用Ollama部署本地模型如Llama 3。强化学习库gymnasium(原OpenAI Gym) 用于创建模拟环境stable-baselines3或ray[rllib]用于高级RL算法但本文为简化会实现一个基础的学习逻辑。向量数据库用于实现长期记忆的快速语义检索。chromadb轻量且易用。其他工具库requests(网络请求)numpy(数值计算)pydantic(数据验证)。2.2 项目初始化与依赖安装首先创建项目目录并初始化虚拟环境。# 创建项目目录 mkdir learning_ai_agent cd learning_ai_agent # 创建虚拟环境 (以conda为例) conda create -n learning_agent python3.10 -y conda activate learning_agent # 安装核心依赖 pip install langgraph langchain-openai chromadb gymnasium requests numpy pydantic # 如果你使用OpenAI API需要设置API Key # export OPENAI_API_KEYyour-api-key-here创建项目基础结构learning_ai_agent/ ├── agent/ │ ├── __init__.py │ ├── core.py # Agent核心类定义 │ ├── memory.py # 记忆系统实现 │ └── learning.py # 学习引擎实现 ├── environment/ │ ├── __init__.py │ └── sim_env.py # 模拟环境定义 ├── config.py # 配置文件 ├── main.py # 主运行脚本 ├── requirements.txt └── README.md生成requirements.txt文件pip freeze requirements.txt3. 构建核心组件记忆系统与模拟环境一个能够学习的Agent需要一个记录经验的地方和一个可供交互的“世界”。我们先从这两个基础组件开始。3.1 实现长期记忆系统我们将实现一个基于chromadb的向量记忆存储它能够存储每次交互的“经验元组”状态、行动、奖励、新状态并支持根据当前状态语义检索相似历史经验。# agent/memory.py import chromadb from chromadb.config import Settings from typing import List, Dict, Any, Optional, Tuple import uuid import numpy as np class ExperienceMemory: 经验记忆系统存储 (state, action, reward, next_state, done) 元组。 支持基于state的语义相似性检索。 def __init__(self, collection_name: str agent_experiences, persist_directory: str ./chroma_db): # 初始化Chroma客户端持久化存储 self.client chromadb.PersistentClient(pathpersist_directory, settingsSettings(anonymized_telemetryFalse)) # 获取或创建集合 self.collection self.client.get_or_create_collection(namecollection_name) def add_experience(self, state: str, action: str, reward: float, next_state: str, done: bool, metadata: Optional[Dict] None): 添加一条经验到记忆库。 # 将经验组合成一段文本用于嵌入 experience_text fState: {state}\nAction: {action}\nReward: {reward}\nNext State: {next_state}\nDone: {done} doc_id str(uuid.uuid4()) # 准备元数据 exp_metadata { state: state, action: action, reward: reward, next_state: next_state, done: done, **(metadata or {}) } # 添加到集合 self.collection.add( documents[experience_text], metadatas[exp_metadata], ids[doc_id] ) return doc_id def retrieve_similar_experiences(self, query_state: str, n_results: int 5) - List[Dict[str, Any]]: 根据当前状态检索相似的历史经验。 results self.collection.query( query_texts[fState: {query_state}], n_resultsn_results ) retrieved_exps [] if results[metadatas]: for meta_list in results[metadatas]: for meta in meta_list: retrieved_exps.append(meta) return retrieved_exps def get_all_experiences(self) - List[Dict[str, Any]]: 获取所有经验仅用于调试或简单学习。 all_data self.collection.get() experiences [] for i, id_ in enumerate(all_data[ids]): exp {k: all_data[k][i] for k in [metadatas, documents] if k in all_data} experiences.append(exp) return experiences3.2 创建一个简单的模拟环境为了演示学习过程我们构建一个极简的文本环境一个“智能恒温器”环境。Agent的目标是通过调节“加热”或“冷却”动作将房间温度维持在舒适的22°C。# environment/sim_env.py import random from typing import Tuple, Dict, Any class ThermostatEnvironment: 一个简单的恒温器模拟环境。 def __init__(self, initial_temp: float 20.0, target_temp: float 22.0, noise_std: float 0.5): self.current_temp initial_temp self.target_temp target_temp self.noise_std noise_std self.time_step 0 self.max_steps 20 def reset(self) - str: 重置环境到初始状态。 self.current_temp 20.0 random.uniform(-2, 2) self.time_step 0 return self._get_state_observation() def _get_state_observation(self) - str: 返回当前状态的文本描述。 return fTime: {self.time_step}, Current Temperature: {self.current_temp:.2f}°C, Target: {self.target_temp}°C def step(self, action: str) - Tuple[str, float, bool, Dict[str, Any]]: 执行一个动作。 :param action: “heat”, “cool”, 或 “nothing” :return: (next_state, reward, done, info) self.time_step 1 # 环境动力学动作影响温度 if action heat: self.current_temp 1.5 random.normalvariate(0, self.noise_std) elif action cool: self.current_temp - 1.5 random.normalvariate(0, self.noise_std) elif action nothing: pass else: raise ValueError(fUnknown action: {action}) # 添加自然波动 self.current_temp random.normalvariate(0, 0.3) # 计算奖励越接近目标奖励越高 temp_diff abs(self.current_temp - self.target_temp) reward -temp_diff # 负差异作为奖励最大化奖励即最小化差异 # 判断是否结束 done self.time_step self.max_steps next_state self._get_state_observation() info {temperature: self.current_temp, action_taken: action} return next_state, reward, done, info4. 构建具备学习能力的AI Agent核心现在我们将LangGraph与记忆系统、学习引擎结合构建Agent的核心逻辑。Agent的决策将基于LLM但会参考历史经验。4.1 定义Agent状态与工具首先我们使用Pydantic定义Agent的状态并创建它可以调用的工具即动作。# agent/core.py from typing import TypedDict, Annotated, List, Dict, Any import operator from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, SystemMessage from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field import json # 定义Agent的持久化状态 class AgentState(TypedDict): Agent在运行过程中的状态。 messages: Annotated[List[Any], operator.add] # 对话消息历史 current_state: str # 环境当前状态描述 total_reward: float # 本轮累计奖励 steps: int # 本轮已执行步数 memory_context: List[Dict] # 从记忆库检索到的相关经验 # 定义可供Agent调用的工具动作 class AgentTools: staticmethod def act_heat() - str: 执行加热动作。 return heat staticmethod def act_cool() - str: 执行冷却动作。 return cool staticmethod def act_nothing() - str: 什么也不做。 return nothing # 一个用于从记忆库查询经验的工具 staticmethod def query_memory(query: str, memory_system) - List[Dict]: 查询记忆系统中与当前情况相似的经验。 return memory_system.retrieve_similar_experiences(query, n_results3)4.2 实现学习引擎学习引擎是一个相对独立模块它定期例如每轮结束后从记忆系统中采样一批经验并更新一个简单的策略表或提供反思总结。# agent/learning.py from .memory import ExperienceMemory from typing import List, Dict, Any import random class SimpleLearningEngine: 一个简单的基于经验回放的学习引擎。 def __init__(self, memory: ExperienceMemory): self.memory memory def learn_from_experience(self, batch_size: int 10) - Dict[str, Any]: 从记忆中随机采样一批经验进行学习。 这里实现一个简单的策略统计在相似状态下哪个动作带来的平均奖励最高。 all_exps self.memory.get_all_experiences() if not all_exps or len(all_exps) 5: return {message: Not enough experiences to learn from.} # 简化处理我们只分析元数据 experiences [] for exp in all_exps: if metadatas in exp and exp[metadatas]: experiences.extend(exp[metadatas]) if len(experiences) batch_size: batch experiences else: batch random.sample(experiences, batch_size) # 构建一个简单的策略统计表state - {action: (total_reward, count)} policy_stats {} for exp in batch: state exp.get(state, ) action exp.get(action, ) reward float(exp.get(reward, 0)) if state not in policy_stats: policy_stats[state] {} if action not in policy_stats[state]: policy_stats[state][action] {total_reward: 0, count: 0} policy_stats[state][action][total_reward] reward policy_stats[state][action][count] 1 # 计算平均奖励并找出每个状态下最好的动作 learned_policy {} for state, actions in policy_stats.items(): best_action None best_avg_reward -float(inf) for action, stats in actions.items(): avg_reward stats[total_reward] / stats[count] if avg_reward best_avg_reward: best_avg_reward avg_reward best_action action if best_action: learned_policy[state] best_action # 将学习到的策略总结成文本可供LLM参考 summary Learned from past experiences:\n for state, action in list(learned_policy.items())[:5]: # 只取前5条 summary f- When in state {state[:50]}..., the action {action} tended to yield good results.\n return { summary: summary, policy_sample: learned_policy, experiences_used: len(batch) }4.3 组装智能体与运行图我们将使用LangGraph来定义Agent的决策流程。这个流程包括观察状态、检索记忆、调用LLM决策、执行动作、存储经验。# agent/core.py (续) from langchain.tools import Tool from langgraph.prebuilt import ToolExecutor from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, SystemMessage from langchain_openai import ChatOpenAI def create_learning_agent(memory_system, learning_engine, env): 创建并返回一个具备学习能力的Agent图。 # 1. 初始化LLM llm ChatOpenAI(modelgpt-4o-mini, temperature0.1) # 使用更小更快的模型 # 2. 绑定工具 tools [ Tool(nameAct_Heat, funclambda x: AgentTools.act_heat(), descriptionIncrease the temperature.), Tool(nameAct_Cool, funclambda x: AgentTools.act_cool(), descriptionDecrease the temperature.), Tool(nameAct_Nothing, funclambda x: AgentTools.act_nothing(), descriptionDo nothing.), Tool(nameQuery_Memory, funclambda q: AgentTools.query_memory(q, memory_system), descriptionQuery past similar experiences from memory. Input should be a description of the current state.) ] tool_executor ToolExecutor(tools) # 3. 定义节点函数 def observe_state(state: AgentState) - AgentState: 节点观察环境状态并查询相关记忆。 current_state state[current_state] # 查询记忆 similar_exps memory_system.retrieve_similar_experiences(current_state, n_results2) state[memory_context] similar_exps return state def decide_action(state: AgentState) - AgentState: 节点LLM基于状态和记忆做出决策。 system_prompt SystemMessage(contentf 你是一个智能恒温器控制Agent。你的目标是通过选择heat、cool或nothing动作将房间温度维持在22°C附近。 当前环境状态{state[current_state]} 以下是过去类似情况下的经验仅供参考 {json.dumps(state.get(memory_context, []), indent2)} 请仔细分析当前状态和历史经验然后只输出一个JSON对象格式如下 {{reasoning: 你的简要推理过程, action: heat|cool|nothing, confidence: 0.0-1.0}} ) human_msg HumanMessage(content请决定下一步动作。) response llm.invoke([system_prompt, human_msg]) try: decision json.loads(response.content) except json.JSONDecodeError: decision {reasoning: Failed to parse, action: nothing, confidence: 0.5} # 将决策存入消息历史便于追溯 state[messages].append(HumanMessage(contentfDecision: {decision})) state[_last_decision] decision return state def execute_action(state: AgentState) - AgentState: 节点执行选定的动作并更新环境。 decision state.get(_last_decision, {action: nothing}) action decision.get(action, nothing) # 在真实环境中执行动作 next_state, reward, done, info env.step(action) # 存储经验到长期记忆 memory_system.add_experience( statestate[current_state], actionaction, rewardreward, next_statenext_state, donedone, metadata{step: state[steps], reasoning: decision.get(reasoning, )} ) # 更新Agent状态 state[current_state] next_state state[total_reward] reward state[steps] 1 state[_last_reward] reward state[_done] done state[messages].append(HumanMessage(contentfExecuted {action}. Reward: {reward:.2f}. New state: {next_state})) return state def should_continue(state: AgentState) - str: 条件边判断是否继续执行。 if state.get(_done, False) or state[steps] 20: return end return continue # 4. 构建图 workflow StateGraph(AgentState) workflow.add_node(observe, observe_state) workflow.add_node(decide, decide_action) workflow.add_node(act, execute_action) workflow.set_entry_point(observe) workflow.add_edge(observe, decide) workflow.add_edge(decide, act) workflow.add_conditional_edges( act, should_continue, { continue: observe, # 继续下一轮观察-决策-行动循环 end: END } ) return workflow.compile()5. 完整实战运行与评估学习型Agent现在我们将所有组件串联起来运行一个完整的实验观察Agent如何通过多轮交互进行学习。5.1 主运行脚本# main.py import asyncio from agent.memory import ExperienceMemory from agent.learning import SimpleLearningEngine from agent.core import create_learning_agent from environment.sim_env import ThermostatEnvironment from langgraph.checkpoint.memory import MemorySaver async def main(): print( 启动具备经验学习能力的AI Agent实验 \n) # 1. 初始化组件 print(初始化记忆系统、学习引擎和环境...) memory ExperienceMemory() learner SimpleLearningEngine(memory) env ThermostatEnvironment() # 2. 创建Agent print(构建智能体工作流...) app create_learning_agent(memory, learner, env) # 3. 运行多轮实验Episodes num_episodes 5 # 运行5轮 checkpoint_memory MemorySaver() # 用于保存Agent的对话状态可选 for episode in range(num_episodes): print(f\n--- 第 {episode 1} 轮实验开始 ---) # 重置环境 initial_state env.reset() total_reward 0.0 # 准备Agent的初始状态 initial_config { configurable: { thread_id: fepisode_{episode}, } } initial_agent_state { messages: [], current_state: initial_state, total_reward: 0.0, steps: 0, memory_context: [] } # 运行Agent直到本轮结束 print(f初始状态: {initial_state}) for step in range(20): # 最大步数 # 使用LangGraph的astream_events可以观察执行过程这里简化使用同步调用 output await app.ainvoke(initial_agent_state, configinitial_config) initial_agent_state output # 更新状态 step_reward initial_agent_state.get(_last_reward, 0) total_reward step_reward current_state initial_agent_state[current_state] print(f 步骤{step1}: 状态{current_state}, 单步奖励{step_reward:.2f}) if initial_agent_state.get(_done, False): print(f 环境提前终止。) break print(f第 {episode 1} 轮结束累计奖励: {total_reward:.2f}) # 4. 每轮结束后让学习引擎从积累的经验中学习 if episode 0: # 从第二轮开始学习 print( 正在进行经验学习...) learning_result learner.learn_from_experience(batch_size8) print(f 学习摘要: {learning_result.get(summary, N/A)[:200]}...) # 短暂暂停模拟真实时间间隔 await asyncio.sleep(0.5) # 5. 实验结束后展示最终的学习成果 print(\n 实验总结 ) final_learn learner.learn_from_experience(batch_size20) print(最终学习策略样本) for state, action in list(final_learn.get(policy_sample, {}).items())[:3]: print(f 状态 {state[:30]}... - 推荐动作 {action}) print(f\n总经验条数: {len(memory.get_all_experiences())}) print(实验完成) if __name__ __main__: asyncio.run(main())5.2 运行结果与分析运行上述脚本(python main.py)你将看到类似以下的输出 启动具备经验学习能力的AI Agent实验 初始化记忆系统、学习引擎和环境... 构建智能体工作流... --- 第 1 轮实验开始 --- 初始状态: Time: 0, Current Temperature: 19.35°C, Target: 22°C 步骤1: 状态Time: 1, Current Temperature: 20.72°C, Target: 22°C, 单步奖励-1.28 步骤2: 状态Time: 2, Current Temperature: 22.15°C, Target: 22°C, 单步奖励-0.15 ... 第 1 轮结束累计奖励: -15.32 --- 第 2 轮实验开始 --- 初始状态: Time: 0, Current Temperature: 21.10°C, Target: 22°C 正在进行经验学习... 学习摘要: Learned from past experiences: - When in state Time: 0, Current Temperature: 19.35°C, Target: 22°C..., the action heat tended to yield good results. ...结果分析第一轮Agent像一张白纸主要依靠LLM的常识进行决策例如“温度低于目标就加热”累计奖励可能较低。后续轮次学习引擎开始工作从记忆库中提取历史经验并总结出简单的策略如“在XX状态下加热动作效果更好”。这些总结会被注入到后续轮次LLM的系统提示中从而影响其决策。趋势随着轮次增加累计奖励应该呈现上升趋势因为Agent决策时参考了更多成功的经验避免了重复过去的错误。这就是“从经验中学习”的直观体现。6. 常见问题与排查思路在构建和运行此类AI Agent时你可能会遇到以下典型问题问题现象可能原因排查思路与解决方案Agent决策循环卡住或不执行动作1. LLM API调用失败或超时。2. LangGraph图定义有误边未正确连接。3.should_continue逻辑始终返回”continue”。1. 检查网络和API密钥添加重试和超时机制。2. 使用workflow.get_graph().draw_mermaid()输出图结构检查。3. 打印_done和steps变量确保终止条件被触发。记忆检索返回空或无关结果1. ChromaDB集合未正确持久化或路径错误。2. 经验文本嵌入相似度低。3. 存储的经验数量太少。1. 检查persist_directory路径权限确认数据已保存。2. 优化经验文本的格式化方式使其包含更关键的语义信息。3. 增加初始随机探索积累更多经验数据。累计奖励没有提升甚至下降1. 学习引擎过于简单未能提取有效模式。2. 环境奖励函数设计不合理。3. LLM未能有效利用记忆上下文。1. 升级学习引擎引入更复杂的RL算法如DQN。2. 重新设计奖励函数使其梯度更平滑更能引导学习。3. 优化提示词明确要求LLM参考提供的“Learned from past experiences”。运行速度非常慢1. 每步都调用LLM成本高且慢。2. ChromaDB检索大量数据。3. 同步调用阻塞。1. 对稳定策略进行缓存或使用更小、更快的模型。2. 为记忆集合建立索引限制每次检索数量。3. 使用异步调用 (ainvoke)。Agent行为不稳定决策随机1. LLM的temperature参数过高。2. 环境随机噪声过大。3. 记忆中的经验相互矛盾。1. 将temperature调低如0.1使输出更确定。2. 适当减小环境中的随机噪声标准差。3. 在学习引擎中引入置信度过滤只采纳被多次验证的经验。7. 进阶优化与工程最佳实践要将这个原型发展为可投入生产或研究的系统需要考虑以下优化方向和实践建议7.1 记忆系统优化分层记忆区分短期工作记忆当前任务上下文和长期经验记忆。短期记忆可用普通列表长期记忆用向量数据库。记忆摘要不是存储所有原始经验而是定期由LLM生成摘要性知识如“在清晨温度较低时立即加热效果最佳”再存入记忆。这能减少存储和检索负担。优先级经验回放像深度强化学习那样为那些带来高奖励或高学习价值的经验赋予更高采样权重。7.2 学习引擎强化集成标准RL算法将stable-baselines3中的PPO、DQN等算法与Agent结合。让LLM负责高层策略生成RL模型负责低层价值评估。LLM微调Fine-tuning收集高质量的成功轨迹状态-动作对用于微调一个专门的“策略LLM”使其初始性能更好。反思与元认知让Agent在任务失败或完成后主动调用LLM进行反思分析错误原因并将反思结论存入记忆实现更高级的元学习。7.3 工程化与部署配置化管理将模型类型、API密钥、环境参数、学习超参数等全部外置到config.yaml文件中。日志与监控记录每一步的决策、奖励、LLM调用消耗便于分析和调试。可使用structlog或prometheus。模块化设计确保记忆、学习、环境、Agent核心等模块之间接口清晰便于单独替换或升级例如将ChromaDB换成Pinecone将OpenAI换成Claude。测试与评估建立独立的测试环境集定义清晰的评估指标如平均奖励、收敛速度、任务成功率定期运行基准测试。构建一个能从真实世界经验中学习的AI Agent是一个持续迭代的过程。本文提供的框架是一个强大的起点它融合了LLM的推理能力、记忆系统的持久化以及学习引擎的适应性。你可以在此基础上将其应用到更复杂的场景如网页自动化、客户服务对话优化、游戏AI甚至机器人控制中。记住核心在于设计一个能让Agent安全、高效地与环境交互并接收反馈的闭环系统。