最近在跟进智能体技术发展时发现一个现象很多开发者对“智能体”的理解还停留在聊天机器人或简单API调用的层面认为它离真正的“自主”还很远。然而银河通用机器人发布的“银河星仔 Galbot ET1”及其背后的“星脑”系统却将“具备自主学习能力的智能体”从概念推向了工程现实。这不仅仅是产品发布更标志着一个技术范式的转变——智能体开始具备从环境中持续学习、反思并优化自身行为的能力。对于开发者而言这既是机遇也是挑战。机遇在于我们可以基于类似架构构建更强大、更自主的业务智能体挑战在于理解其背后的技术原理并付诸实践需要跨越认知和技术实现的鸿沟。本文将深入拆解Galbot ET1所代表的“自主学习智能体”核心技术并提供一个从零开始的、可运行的智能体项目实例涵盖环境搭建、核心模块开发到自主进化机制实现的全流程。无论你是想了解前沿动态还是希望亲手搭建一个具备学习能力的AI助手这篇文章都将提供清晰的路径和可复现的代码。1. 智能体演进从工具调用到自主学习在深入技术细节之前我们有必要厘清智能体Agent概念的演进这有助于理解Galbot ET1所代表的“自主学习”究竟意味着什么。1.1 智能体的三个发展阶段传统的AI应用或聊天机器人可以看作是被动响应型工具。它们根据预设的规则或当前的用户输入生成回应没有记忆没有目标也不会主动规划。随着大语言模型LLM能力的提升任务导向型智能体成为主流。这类智能体通常基于ReActReasoning Acting等框架能够理解复杂指令、调用工具如搜索、计算、写代码并分步完成任务。其核心是“规划-执行”循环但行为逻辑严重依赖预定义的提示词Prompt和工具集缺乏根据结果自我调整策略的能力。而Galbot ET1所代表的自主学习型智能体则进入了第三个阶段。其核心特征是具备**反思Reflection和技能获取Skill Acquisition**能力。它不仅能完成任务还能在任务执行后评估结果的好坏分析成功或失败的原因并将这些经验转化为内部知识或新的行动策略用于优化未来的表现。这使其能够应对更开放、动态的环境而无需开发者为其可能遇到的每一个新场景都预先编写规则。1.2 银河星仔 Galbot ET1 与 “星脑” 系统根据公开信息Galbot ET1并非一个孤立的机器人产品而是一个搭载了“星脑”系统的智能体实例。“星脑”可以理解为驱动其自主学习能力的核心引擎或架构。虽然其内部实现细节未完全公开但结合当前智能体领域的最新技术如LangGraph、AutoGPT、CrewAI等我们可以推断其关键组件感知与决策核心基于强大的多模态大模型理解环境包括物理传感器数据、用户指令、网络信息等。记忆与经验库长期存储任务历史、成功/失败案例、自我反思的结论以及学习到的新技能Skill。反思与优化模块在任务结束后自动启动分析执行轨迹识别可改进点并更新策略或知识库。技能学习与执行引擎能够将复杂的反思结论拆解为可执行、可复用的“技能”Skill并纳入其工具库。安全与伦理护栏确保自主学习过程在可控、安全、符合伦理的边界内进行。这种架构使得Galbot ET1能够从每一次交互中学习不断扩展其能力边界而不仅仅是执行固定脚本。2. 环境准备构建自主智能体的技术栈要动手实现一个具备自主学习雏形的智能体我们不需要从头造轮子。可以基于成熟的开发框架和开源模型来搭建。下面是我们本次实战演示的环境配置。2.1 核心工具与框架选择智能体框架LangGraph。它是一个基于LangChain的库专门用于构建有状态的、多步骤的智能体工作流。其核心概念是“图”Graph非常适合描述智能体的决策循环、工具调用和状态转移是实现反思-学习循环的理想基础。大语言模型LLMOllama本地部署。为了演示方便且可控我们使用Ollama在本地运行开源模型。推荐使用llama3.2:3b或qwen2.5:7b等较小但能力不错的模型进行实验。你也可以使用OpenAI、DeepSeek等云端API。开发语言Python 3.10。这是AI应用开发的主流语言拥有最丰富的生态。向量数据库Chroma。一个轻量级、易用的向量数据库用于存储和检索智能体的记忆经验、反思、技能描述。其他工具langchain、langchain-community、chromadb、pydantic。2.2 项目初始化与环境搭建首先创建一个新的项目目录并设置虚拟环境。# 创建项目目录 mkdir autonomous_agent_demo cd autonomous_agent_demo # 创建虚拟环境 (以conda为例也可用venv) conda create -n agent_env python3.10 -y conda activate agent_env # 安装核心依赖 pip install langgraph langchain langchain-community chromadb pydantic pip install ollama # 用于与本地Ollama服务交互 # 启动Ollama服务并拉取模型 (确保已安装Ollama客户端) # 在另一个终端执行 # ollama pull llama3.2:3b2.3 项目结构设计一个结构清晰的代码组织是复杂智能体项目的基础。我们采用以下结构autonomous_agent_demo/ ├── agent_core/ # 智能体核心模块 │ ├── __init__.py │ ├── graph_builder.py # 定义LangGraph工作流 │ ├── state.py # 定义智能体状态 │ └── nodes.py # 定义图中的各个节点函数 ├── memory/ # 记忆与经验存储 │ ├── __init__.py │ ├── experience_store.py # 经验向量库 │ └── skill_registry.py # 技能注册表 ├── skills/ # 技能工具集 │ ├── __init__.py │ ├── base_skill.py # 技能基类 │ ├── calculator.py # 示例计算技能 │ └── web_search.py # 示例网络搜索技能需API Key ├── config.py # 配置文件 ├── main.py # 主运行入口 └── requirements.txt # 依赖列表3. 核心概念拆解状态、图与反思循环在编码之前必须理解LangGraph构建智能体的几个核心概念这对应了Galbot ET1“星脑”中的关键设计。3.1 智能体状态State智能体是有状态的它需要记住之前的对话、工具调用结果、反思结论等。我们使用Pydantic模型来定义这个状态。# agent_core/state.py from typing import TypedDict, List, Dict, Any, Optional from pydantic import BaseModel class Skill(BaseModel): 技能定义 name: str description: str code_snippet: Optional[str] None # 或函数引用 usage_example: str class AgentState(TypedDict): 智能体的核心状态定义 # 输入与上下文 input: str # 用户当前输入 chat_history: List[Dict[str, str]] # 对话历史 # 执行过程 plan: List[str] # 当前任务的执行计划 actions: List[Dict] # 已执行的动作记录 observation: str # 上一步动作的观察结果工具返回 # 反思与学习 reflection: str # 对当前执行过程的反思文本 new_skills: List[Skill] # 本次学习到的新技能 # 最终输出 response: str # 最终给用户的回复这个AgentState像一个共享的白板图中的所有节点都读写这个状态从而完成协作。3.2 图Graph与节点NodeLangGraph将智能体的工作流建模为一张有向图。节点是执行具体任务的函数如“理解用户意图”、“调用工具”、“反思”边定义了节点之间的流转条件。我们的自主智能体将包含以下关键节点路由节点Router分析当前状态决定下一步是“规划”、“执行工具”还是“反思总结”。规划节点Planner将复杂任务分解为步骤。工具执行节点Actor调用相应的技能工具。反思节点Reflector评估工具执行结果思考是否成功、为何失败、有何收获。技能学习节点Skill Learner从成功的反思中抽象出可复用的技能并存入技能库。响应节点Responder生成最终回复给用户。3.3 反思Reflection机制这是实现“自主学习”的灵魂。反思不是简单的错误日志而是一个由LLM驱动的分析过程。其提示词Prompt大致结构如下你是一个善于总结和学习的智能体。请分析刚刚完成的任务步骤 任务目标{state[‘input’]} 已执行动作{state[‘actions’]} 动作结果{state[‘observation’]} 请思考 1. 当前结果是否满足了任务目标如果满足成功的关键步骤是什么 2. 如果没有满足是哪个环节出了问题是计划不周、工具不对、还是参数错误 3. 从这次经历中可以总结出什么新的经验或知识能否将其归纳为一个清晰的、可重复使用的“技能”或“注意事项” 4. 基于以上分析下一步应该做什么调整计划、使用其他工具、还是直接回答用户 请输出你的反思和分析。LLM根据这个提示生成的文本会被写入state[‘reflection’]进而影响路由节点的下一个决策。4. 完整实战构建具备反思能力的本地AI智能体现在我们将上述概念组合起来构建一个简化但完整的自主智能体。它能处理数学计算、信息查询任务并在失败后尝试新策略甚至“学会”使用新工具。4.1 构建记忆与技能库首先实现记忆存储和技能管理。# memory/experience_store.py import chromadb from chromadb.config import Settings from typing import List, Dict import uuid class ExperienceStore: 存储任务执行经验和反思结论 def __init__(self, persist_directory./chroma_exp_db): self.client chromadb.Client(Settings(persist_directorypersist_directory, chroma_db_implduckdbparquet)) # 创建或获取集合 self.collection self.client.get_or_create_collection(nameagent_experiences) def add_experience(self, task: str, plan: List[str], result: str, reflection: str, learned_skill: str None): 添加一条经验记录 doc_id str(uuid.uuid4()) document fTask: {task}\nPlan: {plan}\nResult: {result}\nReflection: {reflection}\nLearned: {learned_skill} self.collection.add( documents[document], metadatas[{task: task, has_skill: learned_skill is not None}], ids[doc_id] ) print(f[Memory] Experience saved with ID: {doc_id}) def search_similar_experiences(self, query: str, n_results: int 3) - List[Dict]: 搜索相似的历史经验 results self.collection.query(query_texts[query], n_resultsn_results) experiences [] for i in range(len(results[documents][0])): exp { doc: results[documents][0][i], dist: results[distances][0][i] } experiences.append(exp) return experiences# memory/skill_registry.py import json import os from typing import Dict, List, Optional from agent_core.state import Skill class SkillRegistry: 管理智能体已学会的技能 def __init__(self, file_path: str ./skills_registry.json): self.file_path file_path self.skills: Dict[str, Skill] self._load_skills() def _load_skills(self) - Dict[str, Skill]: if os.path.exists(self.file_path): with open(self.file_path, r, encodingutf-8) as f: data json.load(f) return {k: Skill(**v) for k, v in data.items()} return {} def _save_skills(self): with open(self.file_path, w, encodingutf-8) as f: json.dump({k: v.dict() for k, v in self.skills.items()}, f, indent2, ensure_asciiFalse) def register_skill(self, skill: Skill): 注册一个新技能 self.skills[skill.name] skill self._save_skills() print(f[SkillRegistry] New skill registered: {skill.name}) def get_skill(self, name: str) - Optional[Skill]: 根据名称获取技能 return self.skills.get(name) def list_skills(self) - List[Skill]: 列出所有技能 return list(self.skills.values())4.2 实现基础技能工具技能是智能体可调用的函数。我们实现两个基础技能。# skills/base_skill.py from abc import ABC, abstractmethod from typing import Any, Dict class BaseSkill(ABC): 所有技能的基类 property abstractmethod def name(self) - str: pass property abstractmethod def description(self) - str: pass abstractmethod def execute(self, **kwargs) - str: pass def as_tool_dict(self) - Dict[str, Any]: 转换为LangChain工具格式 return { name: self.name, description: self.description, func: self.execute }# skills/calculator.py import re from skills.base_skill import BaseSkill class CalculatorSkill(BaseSkill): 一个简单的数学计算技能 property def name(self): return calculator property def description(self): return Useful for performing arithmetic calculations. Input should be a mathematical expression like 2 2 or (15 - 3) * 4. def execute(self, expression: str) - str: try: # 安全评估只允许基本的数学运算符和数字 # 警告在生产环境中应使用更安全的评估方法如 ast.literal_eval 或专用库 if not re.match(r^[\d\s\\-\*\/\(\)\.]$, expression): return Error: Expression contains invalid characters for safe evaluation. # 这是一个简化示例。实际应用中请使用更安全的计算方式例如 # from ast import literal_eval # result literal_eval(expression) result eval(expression) # 仅用于演示注意安全风险 return fThe result of {expression} is {result}. except Exception as e: return fCalculation error: {e}4.3 构建LangGraph智能体工作流这是最核心的部分我们将定义各个节点并将其连接成图。# agent_core/nodes.py from typing import Literal from langchain_community.chat_models import ChatOllama from langchain_core.messages import HumanMessage, SystemMessage from agent_core.state import AgentState from memory.skill_registry import SkillRegistry from skills.calculator import CalculatorSkill # 初始化LLM和技能 llm ChatOllama(modelllama3.2:3b, temperature0.1) skill_registry SkillRegistry() calculator CalculatorSkill() available_tools {calculator.name: calculator} def router_node(state: AgentState) - Literal[plan, act, reflect, respond]: 路由节点决定下一步做什么 # 如果是第一次输入或刚完成反思且需要新计划则去规划 if not state.get(plan) or state.get(reflection, ).find(need new plan) ! -1: return plan # 如果有计划但未执行完且没有待反思的结果则去执行 if state[plan] and not state.get(observation): return act # 如果有了执行结果observation则需要反思 if state.get(observation): return reflect # 其他情况如反思后认为任务完成则生成响应 return respond def planner_node(state: AgentState) - AgentState: 规划节点分解任务为步骤 prompt f 你是一个任务规划专家。请将以下用户请求分解为具体的、可执行的步骤。 你可以调用的技能工具有{list(available_tools.keys())}。 用户请求{state[input]} 对话历史{state.get(chat_history, [])} 请输出一个步骤列表每个步骤应明确描述要做什么或调用哪个工具。 例如[1. 使用calculator计算(2517)的结果, 2. 将计算结果告知用户] messages [SystemMessage(content你是一个专业的任务规划助手。), HumanMessage(contentprompt)] response llm.invoke(messages) plan_text response.content # 简单解析实际可更复杂 steps [step.strip() for step in plan_text.split(\n) if step.strip() and step.strip()[0].isdigit()] state[plan] steps print(f[Planner] Plan created: {steps}) return state def actor_node(state: AgentState) - AgentState: 执行节点调用工具执行计划中的下一步 if not state[plan]: state[observation] No plan to execute. return state # 取出计划的第一步 current_step state[plan].pop(0) print(f[Actor] Executing step: {current_step}) # 简单的工具匹配逻辑实际应用应更智能如用LLM判断 action_record {step: current_step, tool_used: None, tool_input: None, result: None} if calculator in current_step.lower(): # 提取计算表达式这里简化处理 import re match re.search(r计算(.?)的结果, current_step) or re.search(rcalculate (.), current_step, re.IGNORECASE) expr match.group(1).strip() if match else current_step.split()[-1] tool available_tools[calculator] result tool.execute(expressionexpr) action_record.update({tool_used: calculator, tool_input: expr, result: result}) state[observation] result else: # 默认处理假设这一步是信息性步骤无需工具 state[observation] fCompleted step: {current_step}. (No tool needed) action_record.update({result: state[observation]}) state[actions].append(action_record) return state def reflector_node(state: AgentState) - AgentState: 反思节点分析执行结果思考下一步 reflection_prompt f 请对智能体刚刚的执行进行反思。 原始任务{state[input]} 已执行动作{state[actions]} 最新观察结果{state.get(observation, None)} 当前剩余计划{state[plan]} 请分析 1. 当前结果是否朝着完成任务的方向发展 2. 如果动作失败了原因可能是什么工具错误、参数错误、逻辑错误 3. 基于当前情况下一步应该做什么请给出具体建议例如 - 如果成功且任务完成建议 proceed to respond。 - 如果成功但任务未完成建议 continue with plan。 - 如果失败且需要新策略建议 need new plan。 - 如果发现了一个可重复的模式可以描述一个新技能。 请将你的分析和建议写在“反思”中。 messages [SystemMessage(content你是一个善于分析和总结的智能体反思模块。), HumanMessage(contentreflection_prompt)] response llm.invoke(messages) state[reflection] response.content print(f[Reflector] Reflection: {state[reflection][:200]}...) # 打印前200字符 # 简单解析反思结果决定是否学习新技能这里是一个极简示例 if 新技能 in state[reflection] or 可重复 in state[reflection]: # 在实际中这里应该用LLM提取技能描述并格式化 new_skill Skill( nameflearned_skill_{len(state.get(new_skills, []))}, descriptionA skill learned from reflection., usage_exampleUse when similar situation occurs. ) state.setdefault(new_skills, []).append(new_skill) skill_registry.register_skill(new_skill) return state def responder_node(state: AgentState) - AgentState: 响应节点生成最终回复给用户 final_prompt f 请根据以下信息生成对用户的最终回复。 用户原始问题{state[input]} 任务执行历史{state[actions]} 最终观察结果{state.get(observation, No final observation)} 反思总结{state.get(reflection, No reflection)} 请整合所有信息给出一个清晰、完整、有帮助的答复。 messages [SystemMessage(content你是一个友好的智能体助手。), HumanMessage(contentfinal_prompt)] response llm.invoke(messages) state[response] response.content # 更新对话历史 state[chat_history].extend([ {role: user, content: state[input]}, {role: assistant, content: state[response]} ]) return state# agent_core/graph_builder.py from langgraph.graph import StateGraph, END from langgraph.graph import add_messages from agent_core.state import AgentState from agent_core.nodes import router_node, planner_node, actor_node, reflector_node, responder_node def create_agent_graph(): 创建并返回智能体工作流图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(planner, planner_node) workflow.add_node(actor, actor_node) workflow.add_node(reflector, reflector_node) workflow.add_node(responder, responder_node) # 设置入口点路由决策 workflow.set_conditional_entry_point( router_node, { plan: planner, act: actor, reflect: reflector, respond: responder, } ) # 定义节点之间的无条件边 workflow.add_edge(planner, actor) workflow.add_edge(actor, reflector) workflow.add_edge(reflector, router) # 回到路由节点进行下一轮决策 workflow.add_edge(responder, END) # 编译图 graph workflow.compile() return graph4.4 主程序与运行演示最后我们编写主程序来运行这个智能体。# main.py from agent_core.graph_builder import create_agent_graph from agent_core.state import AgentState from memory.experience_store import ExperienceStore def main(): print(初始化自主智能体...) graph create_agent_graph() memory ExperienceStore() # 初始化智能体状态 initial_state: AgentState { input: , chat_history: [], plan: [], actions: [], observation: , reflection: , new_skills: [], response: } print(\n智能体已就绪。输入 quit 退出。) while True: user_input input(\n用户: ).strip() if user_input.lower() in [quit, exit, q]: print(再见) break if not user_input: continue # 准备本次交互的状态 current_state initial_state.copy() current_state[input] user_input print(f\n[智能体开始处理]) try: # 运行图直到到达END状态 final_state graph.invoke(current_state) # 打印最终回复 print(f\n智能体: {final_state.get(response, No response generated.)}) # 将本次经验存入记忆库简化示例 memory.add_experience( taskuser_input, planfinal_state.get(actions, []), resultfinal_state.get(response, ), reflectionfinal_state.get(reflection, ), learned_skillstr(final_state.get(new_skills, [])) ) except Exception as e: print(f智能体运行出错: {e}) if __name__ __main__: main()运行与测试确保Ollama服务运行且模型已下载。在项目根目录下执行python main.py尝试输入“请帮我计算一下 (15 27) * 2 等于多少”观察控制台输出你会看到[Planner]、[Actor]、[Reflector]等节点的日志最终智能体会给出计算结果。尝试一个更复杂的任务“先计算10的平方再用结果减去15。” 观察智能体如何规划步骤、执行、反思并最终响应。检查生成的chroma_exp_db目录和skills_registry.json文件看看经验和技能是否被存储。5. 进阶优化与工程实践上面的示例是一个高度简化的原型。要构建一个接近Galbot ET1理念的、健壮的自主智能体还需要在以下方面进行深度工程化。5.1 强化反思与学习机制分层反思不仅反思单步动作还要反思整体任务策略、工具选择的有效性。经验检索与复用在规划阶段从ExperienceStore中检索相似历史任务的成功经验直接复用或调整其计划避免重复试错。技能抽象与代码生成当反思发现一个可重复的模式时不应只存储文本描述。可以引导LLM生成对应的Python函数代码code_snippet并通过安全沙箱动态注册为可调用的新工具实现真正的“技能增长”。奖励模型为智能体的行动结果引入量化评分来自用户反馈或环境反馈让反思和优化过程有更明确的目标。5.2 提升规划与决策能力集成高级规划器使用LangChain的PlanAndExecute执行器或集成更专业的规划模型如HuggingFace上的T5-plan。动态工具选择使用LLM根据当前状态和工具描述动态决定调用哪个工具而不是简单的字符串匹配。处理不确定性当工具调用失败或返回模糊结果时规划器应能生成备选方案Plan B。5.3 工程化与生产部署状态持久化将AgentState持久化到数据库如Redis、PostgreSQL支持长对话和智能体休眠唤醒。异步与流式响应使用asyncio和LangGraph的异步支持处理耗时工具调用并实现流式输出响应。可观测性与监控记录智能体完整的决策轨迹Thought、行动Action、观察Observation便于调试和优化。集成Prometheus、Grafana进行监控。安全与护栏工具执行沙箱对于动态生成的技能代码必须在严格隔离的沙箱如Docker容器、restrictedpython中执行。输入输出过滤对用户输入和智能体输出进行内容安全过滤防止注入攻击和不当内容。权限控制为不同技能设定权限等级敏感操作如文件删除、网络访问需要额外授权。多智能体协作利用LangGraph的“多智能体”特性可以创建多个具有不同专长的智能体如分析员、执行员、审核员通过协作完成更复杂的任务这对应了“多智能体联合作战”的概念。6. 常见问题与排查思路在开发自主智能体过程中你可能会遇到以下典型问题问题现象可能原因排查与解决思路智能体陷入死循环规划-执行-反思循环不停1. 反思节点未能正确判断任务完成。2. 路由逻辑有缺陷无法进入respond节点。1. 检查reflector_node的提示词确保其能明确输出“任务完成”的判断。2. 在router_node中增加更严格的终止条件例如if ‘task_complete’ in state.get(‘reflection’, ‘’): return “respond”。3. 在图中设置最大循环次数限制。Ollama模型响应慢或无响应1. 模型未正确加载或内存不足。2. Ollama服务未启动或端口被占用。3. 网络问题如果使用远程API。1. 运行ollama list确认模型存在ollama ps查看运行状态。2. 检查Ollama服务日志 (ollama serve的输出)。3. 尝试更小的模型如tinyllama进行测试。4. 在代码中为LLM调用设置超时参数。工具调用失败或结果解析错误1. 工具输入参数格式错误。2. 从自然语言计划中提取工具参数的逻辑正则表达式太脆弱。3. 工具函数本身抛出异常。1. 在actor_node中增加详细的日志打印出准备传递给工具的参数。2. 用更鲁棒的方法提取参数例如使用一个小型的LLM调用专门做“参数解析”。3. 在工具函数内部做好异常捕获返回清晰的错误信息。记忆ChromaDB无法保存或查询1. 持久化目录权限问题。2. 集合Collection名称不一致。3. 向量化模型问题。1. 检查persist_directory的路径是否存在且可写。2. 确认存储和查询使用的是同一个collection对象。3. ChromaDB默认使用sentence-transformers确保已安装 (pip install sentence-transformers)。新技能注册后无法在后续任务中使用1.available_tools字典未更新。2. 技能代码未正确加载或存在语法错误。3. 规划器Planner不知道新技能的存在。1. 实现一个全局的ToolManager类动态管理工具字典。2. 新技能注册时不仅要保存到文件还要更新内存中的available_tools。3. 在planner_node的提示词中动态注入当前可用的工具列表。构建一个真正具备“自主学习”能力的智能体是一个系统工程涉及规划、执行、反思、记忆、学习等多个模块的紧密协作。本文通过Galbot ET1的灵感结合LangGraph等现代框架提供了一个从零开始的实战指南和可扩展的代码骨架。从理解状态管理、图工作流到实现核心的反思循环再到考虑安全、监控等生产级问题每一步都充满了挑战和乐趣。技术的终点不是复现某个产品而是理解其原理并将其应用于解决实际问题。你可以基于这个框架尝试为你的智能体添加网络搜索、数据库查询、文件操作等更多技能探索更复杂的多智能体协作场景或研究如何利用更优质的开源模型来提升其推理和反思能力。