1. AI Agent基础概念与核心组件AI Agent人工智能代理是一种能够感知环境、自主决策并执行行动的智能系统。与传统的程序不同AI Agent具备以下关键特征自主性无需人工干预即可独立运行反应性能够感知环境变化并做出响应目标导向为实现特定目标而采取行动学习能力能够从经验中改进行为现代AI Agent通常由以下几个核心组件构成语言模型LLM作为Agent的大脑负责理解、推理和决策工具系统扩展Agent的能力边界使其能够执行具体操作记忆系统包括短期记忆和长期记忆保存交互历史和重要信息决策循环控制Agent的行为流程如ReActReasoning and Acting模式2. 开发环境准备与工具选型2.1 基础开发环境配置构建AI Agent需要准备以下开发环境# 创建Python虚拟环境 python -m venv agent-env source agent-env/bin/activate # Linux/Mac # agent-env\Scripts\activate # Windows # 安装核心依赖 pip install openai chromadb sentence-transformers duckduckgo-search2.2 关键工具库选型分析语言模型接口OpenAI GPT系列gpt-3.5-turbo/gpt-4开源替代Llama 2、Falcon等向量数据库ChromaDB轻量级、易集成替代方案Pinecone、Weaviate等嵌入模型Sentence-Transformers的all-MiniLM-L6-v2替代方案OpenAI的text-embedding-ada-002工具库计算Python内置math库网络搜索DuckDuckGo Search文件操作Python标准os/pathlib库提示实际开发中应根据项目需求平衡功能性和复杂性。对于初学者建议从最小可行配置开始逐步扩展功能。3. Agent核心架构实现3.1 语言模型接口封装from typing import List, Dict, Any from dataclasses import dataclass import openai dataclass class Message: role: str # system, user, assistant, tool content: str tool_call_id: str None dataclass class LLMResponse: content: str tool_calls: List[Dict] None property def has_tool_calls(self) - bool: return bool(self.tool_calls) async def chat_completion(messages: List[Message], tools: List[Dict] None) - LLMResponse: 调用OpenAI聊天接口 response await openai.ChatCompletion.acreate( modelgpt-3.5-turbo, messages[m.__dict__ for m in messages], toolstools, tool_choiceauto if tools else None ) message response.choices[0].message return LLMResponse( contentmessage.content, tool_callsmessage.tool_calls )3.2 工具系统实现工具系统是Agent能力的扩展机制核心实现包括工具基类定义from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Any, Dict dataclass class ToolResult: success: bool output: Any error: str None metadata: Dict None class BaseTool(ABC): property abstractmethod def name(self) - str: ... property abstractmethod def description(self) - str: ... property abstractmethod def parameters(self) - Dict: ... abstractmethod def execute(self, **kwargs) - ToolResult: ...工具注册器实现from typing import Dict, List, Optional class ToolRegistry: def __init__(self): self._tools: Dict[str, BaseTool] {} def register(self, tool: BaseTool) - None: if not tool.name: raise ValueError(工具必须设置name属性) self._tools[tool.name] tool def execute_tool(self, tool_name: str, arguments: Dict) - ToolResult: tool self._tools.get(tool_name) if not tool: return ToolResult(False, None, f工具未找到: {tool_name}) try: return tool.execute(**arguments) except Exception as e: return ToolResult(False, None, f工具执行异常: {str(e)})3.3 记忆系统实现记忆系统分为短期记忆和长期记忆两个部分短期记忆对话历史from collections import deque from typing import List class ShortTermMemory: def __init__(self, max_size20): self._storage deque(maxlenmax_size) def add(self, content: str, metadata: dict None): self._storage.append({ content: content, metadata: metadata or {} }) def get_recent(self, n: int 5) - List[dict]: return list(self._storage)[-n:]长期记忆向量存储import chromadb from chromadb.utils import embedding_functions class LongTermMemory: def __init__(self, collection_nameagent_memory): self.embedding_func embedding_functions.SentenceTransformerEmbeddingFunction( model_nameall-MiniLM-L6-v2 ) self.client chromadb.Client() self.collection self.client.get_or_create_collection( namecollection_name, embedding_functionself.embedding_func ) def add(self, content: str, metadata: dict None): self.collection.add( documents[content], metadatas[metadata or {}], ids[str(hash(content))] ) def search(self, query: str, top_k: int 3) - List[dict]: results self.collection.query( query_texts[query], n_resultstop_k ) return [ {content: doc, metadata: meta} for doc, meta in zip(results[documents][0], results[metadatas][0]) ]4. Agent主循环与决策逻辑4.1 ReAct决策循环实现ReActReasoning and Acting是Agent常用的决策模式class Agent: def __init__(self, llm, tools, memory): self.llm llm self.tools tools self.memory memory self.max_iterations 10 async def run(self, user_input: str) - str: messages [ Message(system, self._get_system_prompt()), Message(user, user_input) ] for _ in range(self.max_iterations): # 获取LLM响应 response await self.llm.chat_completion(messages) if response.has_tool_calls: # 执行工具调用 for call in response.tool_calls: tool_name call.function.name arguments json.loads(call.function.arguments) result self.tools.execute_tool(tool_name, arguments) messages.append(Message( roletool, contentjson.dumps(result.output), tool_call_idcall.id )) else: # 返回最终响应 return response.content return 达到最大迭代次数任务未完成4.2 系统提示词设计良好的系统提示词对Agent行为至关重要def _get_system_prompt(self) - str: return 你是一个AI助手能够使用工具完成任务。遵循以下规则 1. 仔细分析用户需求必要时请求澄清 2. 合理使用工具每次只调用一个工具 3. 根据工具结果决定下一步行动 4. 保持响应简洁专业 可用工具 {工具列表}5. 实战构建多功能助手5.1 计算器工具实现from math import * from typing import Dict class CalculatorTool(BaseTool): property def name(self) - str: return calculator property def description(self) - str: return 执行数学计算 property def parameters(self) - Dict: return { type: object, properties: { expression: {type: string, description: 数学表达式} }, required: [expression] } def execute(self, expression: str) - ToolResult: try: allowed_names {k: v for k, v in globals().items() if not k.startswith(_)} allowed_names.update({abs: abs, round: round}) code compile(expression, string, eval) for name in code.co_names: if name not in allowed_names: return ToolResult(False, None, f禁止使用的函数/变量: {name}) result eval(code, {__builtins__: {}}, allowed_names) return ToolResult(True, result) except Exception as e: return ToolResult(False, None, str(e))5.2 网络搜索工具实现from duckduckgo_search import ddg class WebSearchTool(BaseTool): property def name(self) - str: return web_search property def description(self) - str: return 使用DuckDuckGo进行网络搜索 property def parameters(self) - Dict: return { type: object, properties: { query: {type: string, description: 搜索关键词}, max_results: {type: integer, description: 最大结果数, default: 3} }, required: [query] } def execute(self, query: str, max_results: int 3) - ToolResult: try: results ddg(query, max_resultsmax_results) simplified [{ title: r[title], url: r[link], snippet: r[body] } for r in results] return ToolResult(True, simplified) except Exception as e: return ToolResult(False, None, str(e))5.3 文件操作工具实现import os from pathlib import Path class FileReadTool(BaseTool): property def name(self) - str: return read_file property def description(self) - str: return 读取文件内容 property def parameters(self) - Dict: return { type: object, properties: { path: {type: string, description: 文件路径} }, required: [path] } def execute(self, path: str) - ToolResult: try: if not os.path.exists(path): return ToolResult(False, None, 文件不存在) with open(path, r, encodingutf-8) as f: content f.read() return ToolResult(True, content) except Exception as e: return ToolResult(False, None, str(e))6. 完整Agent集成与测试6.1 Agent初始化与配置def create_default_agent(): # 初始化组件 llm LLMInterface() tools ToolRegistry() memory { short_term: ShortTermMemory(), long_term: LongTermMemory() } # 注册工具 tools.register(CalculatorTool()) tools.register(WebSearchTool()) tools.register(FileReadTool()) tools.register(FileWriteTool()) # 创建Agent return Agent( llmllm, toolstools, memorymemory )6.2 测试用例设计import asyncio async def test_agent(): agent create_default_agent() test_cases [ 计算3的平方加4的平方, 搜索最新的Python 4.0发布消息, 创建一个test.txt文件内容写Hello World, 读取test.txt文件内容, 根据我的年龄32岁推荐适合的健身计划 ] for task in test_cases: print(f用户: {task}) response await agent.run(task) print(f助手: {response}\n) asyncio.run(test_agent())7. 性能优化与调试技巧7.1 常见问题排查工具调用失败检查工具参数是否符合JSON Schema定义验证工具执行权限如文件操作查看工具返回的错误信息记忆检索不准确调整向量相似度阈值优化嵌入模型选择增加检索结果数量后过滤无限循环问题设置合理的最大迭代次数添加超时机制记录决策历史用于调试7.2 性能优化策略缓存机制from functools import lru_cache lru_cache(maxsize100) def cached_embedding(text: str) - List[float]: return embedding_model.encode(text)批量处理# 批量处理工具调用 async def batch_execute_tools(calls: List[ToolCall]): return await asyncio.gather( *[self.execute_tool(call) for call in calls] )异步IO优化async def parallel_operations(): task1 asyncio.create_task(llm_response()) task2 asyncio.create_task(memory_search()) await asyncio.gather(task1, task2)8. 扩展方向与进阶功能8.1 多Agent协作系统class MultiAgentSystem: def __init__(self, agents: List[Agent]): self.agents {agent.name: agent for agent in agents} self.coordinator CoordinatorAgent() async def delegate_task(self, task: str, specialist: str None): if specialist: return await self.agents[specialist].run(task) # 由协调Agent分配任务 decision await self.coordinator.decide_agent(task) return await self.agents[decision].run(task)8.2 动态工具加载def load_tools_from_directory(dir_path: str) - List[BaseTool]: tools [] for file in Path(dir_path).glob(*.py): spec importlib.util.spec_from_file_location(file.stem, file) module importlib.util.module_from_spec(spec) spec.loader.exec_module(module) for name, obj in module.__dict__.items(): if isinstance(obj, type) and issubclass(obj, BaseTool) and obj ! BaseTool: tools.append(obj()) return tools8.3 强化学习微调class RLFineTuner: def __init__(self, agent: Agent): self.agent agent self.reward_model load_reward_model() async def train_episode(self, task: str): # 运行Agent获取轨迹 trajectory await self.agent.run_with_logging(task) # 计算奖励 reward self.reward_model.evaluate(trajectory) # 更新策略 self.update_agent_policy(trajectory, reward)构建AI Agent是一个迭代过程建议从简单功能开始逐步添加复杂性和新功能。关键是要建立良好的调试和评估机制确保每个组件的可靠性和整个系统的稳定性。
