在人工智能快速发展的今天，AI Agent 作为能自主决策、执行任务的智能实体，正成为技术探索的热点。对于程序员而言，从零构建一个 AI Agent 不仅能深入理解智能系统的工作原理，更能为后续复杂应用打下基础。本文将以 Python 为开发语言，详解 AI Agent 的核心架构设计、关键功能实现与交互逻辑开发，通过可运行的代码示例，带你完成第一个 AI Agent 的搭建过程。

核心架构：AI Agent 的基础框架设计

一个完整的 AI Agent 需要具备感知环境、决策规划和执行行动的核心能力。合理的架构设计是 Agent 稳定运行的基础，通常包括环境交互层、决策核心层和执行层三个关键部分，各模块通过标准化接口实现数据流转。

AI Agent 核心架构代码：

from typing import Dict, Any, List, Callable

import time

from dataclasses import dataclass

import logging

# 配置日志

logging.basicConfig(level=logging.INFO, format='[Agent] %(message)s')

logger = logging.getLogger(__name__)

@dataclass

class AgentState:

"""Agent状态数据类，存储运行时信息"""

current_task: str = ""

task_progress: float = 0.0 # 0.0-1.0

memory: Dict[str, Any] = None # 短期记忆

last_action: str = ""

environment_status: Dict[str, Any] = None # 环境状态快照

def __post_init__(self):

if self.memory is None:

self.memory = {}

class EnvironmentInterface:

"""环境交互接口，定义Agent与外部环境的交互规范"""

def __init__(self):

self.environment_data = {}

def get_state(self) -> Dict[str, Any]:

"""获取当前环境状态"""

return self.environment_data.copy()

def set_state(self, key: str, value: Any):

"""更新环境状态"""

self.environment_data[key] = value

def execute_action(self, action: str, params: Dict[str, Any]) -> Dict[str, Any]:

"""执行动作并返回结果"""

logger.info(f"执行动作: {action} 参数: {params}")

# 模拟环境响应

time.sleep(0.1)

return {

"success": True,

"result": f"动作 {action} 执行完成",

"timestamp": time.time()

}

class AIAgent:

"""AI Agent核心类，实现感知-决策-执行闭环"""

def __init__(self, name: str):

self.name = name

self.state = AgentState()

self.environment = EnvironmentInterface()

self.decision_strategies = {} # 决策策略注册表

self.skills = {} # 技能注册表

self.initialize_default_components()

def initialize_default_components(self):

"""初始化默认组件"""

# 注册默认决策策略

self.register_decision_strategy("simple", self.simple_decision_strategy)

# 注册基础技能

self.register_skill("greet", self.skill_greet)

self.register_skill("task_process", self.skill_task_process)

self.register_skill("memory_store", self.skill_memory_store)

def register_decision_strategy(self, name: str, strategy: Callable):

"""注册决策策略"""

self.decision_strategies[name] = strategy

logger.info(f"注册决策策略: {name}")

def register_skill(self, name: str, skill: Callable):

"""注册技能函数"""

self.skills[name] = skill

logger.info(f"注册技能: {name}")

def perceive(self) -> Dict[str, Any]:

"""感知环境状态"""

env_state = self.environment.get_state()

self.state.environment_status = env_state

logger.info(f"感知环境状态: {env_state}")

return env_state

def decide(self, strategy_name: str = "simple") -> Dict[str, Any]:

"""决策过程，选择要执行的动作"""

if strategy_name not in self.decision_strategies:

raise ValueError(f"未知决策策略: {strategy_name}")

strategy = self.decision_strategies[strategy_name]

decision = strategy()

self.state.last_action = decision.get("action", "none")

logger.info(f"决策结果: {decision}")

return decision

def act(self, decision: Dict[str, Any]) -> Dict[str, Any]:

"""执行决策动作"""

action_name = decision.get("action")

action_params = decision.get("params", {})

if not action_name or action_name not in self.skills:

return {"success": False, "error": "无效动作或未注册技能"}

# 执行技能

result = self.skills[action_name](**action_params)

# 更新任务进度

if "task" in decision:

self.state.current_task = decision["task"]

self.state.task_progress = decision.get("progress", 0.0)

return result

def run_cycle(self, strategy_name: str = "simple") -> Dict[str, Any]:

"""运行一次感知-决策-执行循环"""

logger.info("=== 开始Agent循环 ===")

self.perceive()

decision = self.decide(strategy_name)

result = self.act(decision)

logger.info("=== 完成Agent循环 ===")

return {

"decision": decision,

"result": result,

"state": self.state

}

上述架构采用模块化设计思想，具有三个显著特点：一是状态管理清晰，通过 AgentState 类统一存储运行时数据，包括当前任务、进度和环境快照；二是组件可扩展，支持动态注册决策策略和技能函数，便于功能扩展；三是接口标准化，感知、决策、执行三层通过明确的函数接口通信，降低模块间耦合。这种设计既符合 AI Agent 的经典 BDI（信念 - 愿望 - 意图）模型，又具备良好的可维护性，适合作为入门级 Agent 的基础框架。

功能实现：核心决策与技能模块开发

AI Agent 的智能性体现在其决策能力和执行技能上。决策模块需要根据环境状态和自身目标选择最优行动，而技能模块则实现具体的任务执行逻辑，两者结合形成完整的智能行为。

决策与技能模块代码实现：

class AIAgent(AIAgent): # 继承自上一节的AIAgent类

# 决策策略实现

def simple_decision_strategy(self) -> Dict[str, Any]:

"""简单决策策略实现"""

env_state = self.state.environment_status or {}

current_task = self.state.current_task

# 根据环境状态和当前任务决策

if not current_task:

# 无任务时，检查是否有新任务请求

if "task_request" in env_state:

new_task = env_state["task_request"]

return {

"action": "task_process",

"task": new_task,

"params": {"task": new_task},

"progress": 0.1

}

else:

# 无任务且无请求时，执行问候

return {

"action": "greet",

"params": {"name": "用户"},

"progress": 1.0

}

else:

# 有任务时，继续处理任务

current_progress = self.state.task_progress

if current_progress < 1.0:

new_progress = min(current_progress + 0.2, 1.0)

return {

"action": "task_process",

"task": current_task,

"params": {"task": current_task, "progress": new_progress},

"progress": new_progress

}

else:

# 任务完成，存储结果到记忆

return {

"action": "memory_store",

"params": {

"key": f"task_{current_task}",

"value": {"status": "completed", "time": time.time()}

},

"progress": 1.0

}

# 技能实现

def skill_greet(self, name: str) -> Dict[str, Any]:

"""问候技能"""

greeting = f"你好，{name}！我是{self.name}，你的AI助手。有什么我可以帮助你的吗？"

return {

"success": True,

"response": greeting,

"skill": "greet"

}

def skill_task_process(self, task: str, progress: float = 0.0) -> Dict[str, Any]:

"""任务处理技能"""

status = "处理中" if progress < 1.0 else "已完成"

message = f"任务「{task}」{status}，进度: {progress*100:.1f}%"

# 模拟任务处理逻辑

processing_steps = {

"数据整理": "正在收集和清洗数据...",

"报告生成": "正在生成分析报告...",

"文件转换": "正在转换文件格式..."

}

details = processing_steps.get(task, "执行默认任务流程...")

return {

"success": True,

"response": message,

"details": details,

"progress": progress,

"skill": "task_process"

}

def skill_memory_store(self, key: str, value: Any) -> Dict[str, Any]:

"""记忆存储技能"""

self.state.memory[key] = value

return {

"success": True,

"response": f"已将信息存储到记忆: {key}",

"memory_size": len(self.state.memory),

"skill": "memory_store"

}

def skill_query_memory(self, key: str) -> Dict[str, Any]:

"""记忆查询技能"""

if key in self.state.memory:

return {

"success": True,

"response": f"记忆中找到: {key}",

"value": self.state.memory[key],

"skill": "query_memory"

}

else:

return {

"success": False,

"response": f"记忆中未找到: {key}",

"skill": "query_memory"

}

# 扩展：添加自然语言理解能力

class NLUService:

"""自然语言理解服务，解析用户输入"""

def __init__(self):

# 简单的意图识别规则

self.intent_patterns = {

"greet": ["你好", "哈喽", "嗨"],

"request_task": ["帮我", "请处理", "我需要"],

"query_status": ["进度", "怎么样了", "完成了吗"],

"query_memory": ["记得", "之前的", "查看记录"]

}

# 任务提取关键词

self.task_keywords = ["数据整理", "报告生成", "文件转换"]

def parse_input(self, text: str) -> Dict[str, Any]:

"""解析用户输入文本，识别意图和实体"""

text = text.lower()

intent = "unknown"

entities = {}

# 识别意图

for intent_name, patterns in self.intent_patterns.items():

if any(pattern in text for pattern in patterns):

intent = intent_name

break

# 提取实体

if intent == "request_task":

for task in self.task_keywords:

if task in text:

entities["task"] = task

break

if "task" not in entities:

entities["task"] = "默认任务"

elif intent == "query_status" or intent == "query_memory":

for task in self.task_keywords:

if task in text:

entities["task"] = task

break

return {

"intent": intent,

"entities": entities,

"confidence": 0.8 if intent != "unknown" else 0.0

}

决策模块采用基于规则的策略实现，通过分析环境状态和当前任务进度决定下一步行动，这种实现方式简单直观且易于调试，适合入门级 Agent。技能模块则实现了四类基础能力：问候交互、任务处理、记忆存储和记忆查询，每个技能都封装了具体的业务逻辑，并通过统一的字典格式返回结果。

代码中特别集成了简单的自然语言理解（NLU）服务，通过关键词匹配识别用户意图和任务实体，为 Agent 添加了自然语言交互能力。这种设计的优势在于：决策逻辑与执行技能分离，便于单独优化；技能可独立扩展，新功能只需实现新的技能函数并注册；NLU 模块作为独立服务，可根据需求替换为更复杂的模型（如使用 Hugging Face 的 Transformers 库）。

系统集成：Agent 运行与交互优化

完成核心模块开发后，需要将各组件整合为完整系统，并实现用户交互界面，使 Agent 能够接收输入、展示结果。同时通过运行循环和状态管理，确保 Agent 能够持续处理任务并保持状态一致性。

系统集成与交互代码：

class AgentInterface:

"""Agent用户交互接口"""

def __init__(self, agent: AIAgent, nlu_service: NLUService):

self.agent = agent

self.nlu = nlu_service

self.running = False

def start_agent(self):

"""启动Agent运行循环"""

self.running = True

print(f"=== {self.agent.name} 已启动，输入 'exit' 退出 ===")

self.agent.run_cycle() # 初始循环

# 交互循环

while self.running:

user_input = input("\n你: ")

if user_input.lower() == "exit":

self.stop_agent()

break

self.process_user_input(user_input)

# 执行一次Agent循环

cycle_result = self.agent.run_cycle()

self.display_result(cycle_result)

def stop_agent(self):

"""停止Agent运行"""

self.running = False

print(f"\n=== {self.agent.name} 已停止 ===")

def process_user_input(self, text: str):

"""处理用户输入，更新环境状态"""

# 解析用户输入

nlu_result = self.nlu.parse_input(text)

print(f"[NLU] 意图: {nlu_result['intent']}, 实体: {nlu_result['entities']}")

# 根据意图更新环境状态

env_updates = {}

if nlu_result["intent"] == "request_task" and "task" in nlu_result["entities"]:

env_updates["task_request"] = nlu_result["entities"]["task"]

elif nlu_result["intent"] == "query_status" and "task" in nlu_result["entities"]:

env_updates["query_task"] = nlu_result["entities"]["task"]

elif nlu_result["intent"] == "query_memory" and "task" in nlu_result["entities"]:

# 触发记忆查询技能

self.agent.state.environment_status["memory_query"] = nlu_result["entities"]["task"]

# 更新环境状态

for key, value in env_updates.items():

self.agent.environment.set_state(key, value)

def display_result(self, cycle_result: Dict[str, Any]):

"""展示Agent执行结果"""

result = cycle_result["result"]

decision = cycle_result["decision"]

state = cycle_result["state"]

print(f"\n{self.agent.name}: {result['response']}")

# 显示任务进度

if state.current_task and state.task_progress < 1.0:

print(f"任务进度: {state.task_progress*100:.1f}%")

# 显示详细信息

if "details" in result:

print(f"详情: {result['details']}")

# 完整运行示例

if __name__ == "__main__":

# 创建Agent实例

my_agent = AIAgent(name="小助手")

# 注册额外技能

my_agent.register_skill("query_memory", my_agent.skill_query_memory)

# 创建NLU服务

nlu = NLUService()

# 创建交互接口

interface = AgentInterface(my_agent, nlu)

# 启动Agent

interface.start_agent()

# 扩展：添加长期记忆存储

import json

import os

class PersistentMemory:

"""持久化记忆存储，将Agent记忆保存到文件"""

def __init__(self, memory_file: str = "agent_memory.json"):

self.memory_file = memory_file

self.load_memory()

def load_memory(self) -> Dict[str, Any]:

"""从文件加载记忆"""

if os.path.exists(self.memory_file):

with open(self.memory_file, "r", encoding="utf-8") as f:

return json.load(f)

return {}

def save_memory(self, memory: Dict[str, Any]):

"""将记忆保存到文件"""

with open(self.memory_file, "w", encoding="utf-8") as f:

json.dump(memory, f, ensure_ascii=False, indent=2)

# 使用持久化记忆</doubaocanvas>