在AI技術(shù)快速發(fā)展的今天智能體Agent已成為連接大語言模型與現(xiàn)實(shí)應(yīng)用的重要橋梁。很多開發(fā)者習(xí)慣直接使用現(xiàn)成的Agent框架但往往陷入黑箱困境——不清楚內(nèi)部機(jī)制遇到問題難以排查。本文將從零開始手把手帶你用Python構(gòu)建一個(gè)具備完整思考鏈路的Agent系統(tǒng)不依賴任何第三方框架讓你真正掌握Agent的核心原理。無論你是剛接觸AI應(yīng)用開發(fā)的新手還是希望深入理解Agent機(jī)制的中級(jí)開發(fā)者都能通過本文獲得扎實(shí)的實(shí)戰(zhàn)經(jīng)驗(yàn)。我們將從最基礎(chǔ)的概念講起逐步實(shí)現(xiàn)一個(gè)能夠理解任務(wù)、制定計(jì)劃、執(zhí)行工具調(diào)用并自我反思的完整Agent。1. Agent核心概念與架構(gòu)設(shè)計(jì)1.1 什么是AgentAgent本質(zhì)上是一個(gè)能夠感知環(huán)境、自主決策并執(zhí)行動(dòng)作的智能系統(tǒng)。在AI語境下Agent通常由大語言模型LLM驅(qū)動(dòng)通過思考-行動(dòng)-觀察的循環(huán)來完成任務(wù)。與簡(jiǎn)單調(diào)用API不同真正的Agent具備以下特征自主性能夠獨(dú)立分析任務(wù)并制定執(zhí)行策略工具使用可以調(diào)用外部工具如計(jì)算器、搜索引擎、API等狀態(tài)保持在多次交互中維持對(duì)話歷史和任務(wù)狀態(tài)反思能力能夠評(píng)估執(zhí)行結(jié)果并調(diào)整策略1.2 自建Agent vs 框架選擇目前市面上的Agent框架如LangChain、AutoGPT等確實(shí)提供了便利但也存在一些問題抽象層次過高隱藏了關(guān)鍵實(shí)現(xiàn)細(xì)節(jié)框架依賴性強(qiáng)版本升級(jí)可能帶來兼容性問題調(diào)試?yán)щy錯(cuò)誤信息經(jīng)過多層封裝通過從零開始構(gòu)建你可以完全掌控每個(gè)組件的實(shí)現(xiàn)邏輯根據(jù)業(yè)務(wù)需求靈活定制功能深入理解Agent的工作機(jī)制為后續(xù)優(yōu)化和故障排查打下基礎(chǔ)1.3 基礎(chǔ)架構(gòu)設(shè)計(jì)我們?cè)O(shè)計(jì)的Agent包含以下核心模塊Agent System ├── 思考引擎 (LLM接口) ├── 工具庫 (可擴(kuò)展的函數(shù)集合) ├── 記憶系統(tǒng) (對(duì)話歷史與狀態(tài)管理) ├── 執(zhí)行控制器 (任務(wù)規(guī)劃與調(diào)度) └── 反思機(jī)制 (結(jié)果評(píng)估與策略調(diào)整)2. 環(huán)境準(zhǔn)備與基礎(chǔ)搭建2.1 開發(fā)環(huán)境要求本文基于以下環(huán)境進(jìn)行演示你可以根據(jù)實(shí)際情況調(diào)整# 環(huán)境要求 Python版本: 3.8 必要庫: - openai1.0.0 # LLM接口調(diào)用 - requests2.25.0 # 外部API調(diào)用 - python-dotenv1.0.0 # 環(huán)境變量管理 # 可選工具庫: - wolframalpha5.0.0 # 數(shù)學(xué)計(jì)算 - wikipedia1.4.0 # 知識(shí)查詢2.2 項(xiàng)目結(jié)構(gòu)規(guī)劃創(chuàng)建以下項(xiàng)目目錄結(jié)構(gòu)my_agent/ ├── core/ │ ├── __init__.py │ ├── agent.py # Agent主類 │ ├── memory.py # 記憶管理 │ └── planner.py # 任務(wù)規(guī)劃 ├── tools/ │ ├── __init__.py │ ├── base_tool.py # 工具基類 │ └── calculator.py # 計(jì)算工具示例 ├── config/ │ └── settings.py # 配置文件 ├── utils/ │ └── helpers.py # 工具函數(shù) └── main.py # 啟動(dòng)入口2.3 基礎(chǔ)配置設(shè)置創(chuàng)建配置文件管理API密鑰等敏感信息# config/settings.py import os from dotenv import load_dotenv load_dotenv() class Config: 配置管理類 # OpenAI API配置 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) OPENAI_BASE_URL os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) MODEL_NAME os.getenv(MODEL_NAME, gpt-3.5-turbo) # Agent配置 MAX_ITERATIONS int(os.getenv(MAX_ITERATIONS, 10)) # 最大迭代次數(shù) TEMPERATURE float(os.getenv(TEMPERATURE, 0.1)) # 生成溫度 classmethod def validate(cls): 驗(yàn)證必要配置 if not cls.OPENAI_API_KEY: raise ValueError(OPENAI_API_KEY環(huán)境變量未設(shè)置)在項(xiàng)目根目錄創(chuàng)建.env文件存儲(chǔ)敏感信息# .env 文件 OPENAI_API_KEYyour_api_key_here MODEL_NAMEgpt-3.5-turbo MAX_ITERATIONS10 TEMPERATURE0.13. 核心組件實(shí)現(xiàn)3.1 LLM接口封裝首先實(shí)現(xiàn)一個(gè)基礎(chǔ)的LLM調(diào)用類# core/llm_client.py import json from openai import OpenAI from config.settings import Config class LLMClient: LLM客戶端封裝 def __init__(self): self.client OpenAI(api_keyConfig.OPENAI_API_KEY, base_urlConfig.OPENAI_BASE_URL) self.model Config.MODEL_NAME def generate_response(self, messages, temperature0.1): 生成LLM響應(yīng) try: response self.client.chat.completions.create( modelself.model, messagesmessages, temperaturetemperature, max_tokens2000 ) return response.choices[0].message.content except Exception as e: raise Exception(fLLM調(diào)用失敗: {str(e)}) def structured_output(self, prompt, output_schema): 生成結(jié)構(gòu)化輸出 system_message { role: system, content: f請(qǐng)嚴(yán)格按照以下JSON格式輸出: {json.dumps(output_schema, ensure_asciiFalse)} } response self.generate_response([system_message, {role: user, content: prompt}]) try: return json.loads(response) except json.JSONDecodeError: # 如果JSON解析失敗嘗試提取JSON部分 import re json_match re.search(r\{.*\}, response, re.DOTALL) if json_match: return json.loads(json_match.group()) raise Exception(無法解析LLM返回的結(jié)構(gòu)化數(shù)據(jù))3.2 工具系統(tǒng)實(shí)現(xiàn)工具是Agent與外界交互的核心我們先定義工具基類# tools/base_tool.py from abc import ABC, abstractmethod import inspect class BaseTool(ABC): 工具基類 def __init__(self, name, description, parameters): self.name name self.description description self.parameters parameters # 參數(shù)schema abstractmethod def execute(self, **kwargs): 執(zhí)行工具 pass def get_schema(self): 獲取工具schema return { name: self.name, description: self.description, parameters: self.parameters } class CalculatorTool(BaseTool): 計(jì)算器工具示例 def __init__(self): super().__init__( namecalculator, description執(zhí)行數(shù)學(xué)計(jì)算, parameters{ type: object, properties: { expression: { type: string, description: 數(shù)學(xué)表達(dá)式如: 2 3 * 4 } }, required: [expression] } ) def execute(self, expression): 執(zhí)行計(jì)算 try: # 安全評(píng)估數(shù)學(xué)表達(dá)式 allowed_chars set(0123456789-*/(). ) if not all(c in allowed_chars for c in expression): return 錯(cuò)誤: 表達(dá)式包含不安全字符 result eval(expression) # 注意生產(chǎn)環(huán)境需要更安全的方式 return f計(jì)算結(jié)果: {expression} {result} except Exception as e: return f計(jì)算錯(cuò)誤: {str(e)} # tools/__init__.py from .calculator import CalculatorTool __all__ [CalculatorTool]3.3 記憶系統(tǒng)實(shí)現(xiàn)記憶系統(tǒng)負(fù)責(zé)維護(hù)對(duì)話歷史和Agent狀態(tài)# core/memory.py from typing import List, Dict, Any class Memory: 記憶管理系統(tǒng) def __init__(self, max_history10): self.max_history max_history self.conversation_history: List[Dict] [] self.agent_state: Dict[str, Any] {} def add_message(self, role: str, content: str): 添加消息到歷史 self.conversation_history.append({role: role, content: content}) # 保持歷史記錄不超過最大值 if len(self.conversation_history) self.max_history * 2: # 考慮user和assistant對(duì)話 self.conversation_history self.conversation_history[-self.max_history*2:] def get_recent_history(self, num_messages6): 獲取最近的對(duì)話歷史 return self.conversation_history[-num_messages:] def update_state(self, key: str, value: Any): 更新Agent狀態(tài) self.agent_state[key] value def get_state(self, key: str, defaultNone): 獲取Agent狀態(tài) return self.agent_state.get(key, default) def get_context(self): 獲取當(dāng)前上下文歷史狀態(tài) context { recent_history: self.get_recent_history(), current_state: self.agent_state } return context4. 完整Agent實(shí)現(xiàn)4.1 Agent主類設(shè)計(jì)現(xiàn)在我們將各個(gè)組件組合成完整的Agent# core/agent.py import json import re from typing import List, Dict, Any from .llm_client import LLMClient from .memory import Memory from tools.base_tool import BaseTool class SimpleAgent: 基礎(chǔ)Agent實(shí)現(xiàn) def __init__(self, tools: List[BaseTool] None): self.llm LLMClient() self.memory Memory() self.tools tools or [] self.iteration_count 0 self.max_iterations 10 # 注冊(cè)工具 self.tool_map {tool.name: tool for tool in self.tools} def add_tool(self, tool: BaseTool): 添加工具 self.tools.append(tool) self.tool_map[tool.name] tool def plan_action(self, user_input: str) - Dict[str, Any]: 規(guī)劃下一步行動(dòng) prompt self._build_planning_prompt(user_input) planning_schema { type: object, properties: { thought: {type: string, description: 當(dāng)前思考}, action: {type: string, description: 下一步行動(dòng): think|use_tool|final_answer}, tool_name: {type: string, description: 工具名稱如果action是use_tool}, tool_input: {type: object, description: 工具輸入?yún)?shù)}, response: {type: string, description: 直接回復(fù)如果action是final_answer} }, required: [thought, action] } return self.llm.structured_output(prompt, planning_schema) def _build_planning_prompt(self, user_input: str) - str: 構(gòu)建規(guī)劃提示詞 available_tools \n.join([ f- {tool.name}: {tool.description} for tool in self.tools ]) prompt f 你是一個(gè)智能助手需要根據(jù)用戶請(qǐng)求決定下一步行動(dòng)。 可用工具: {available_tools} 當(dāng)前對(duì)話歷史: {json.dumps(self.memory.get_recent_history(), ensure_asciiFalse, indent2)} 用戶輸入: {user_input} 請(qǐng)分析當(dāng)前情況選擇以下行動(dòng)之一: 1. think - 需要更多思考但不使用工具 2. use_tool - 需要使用工具來獲取信息 3. final_answer - 可以直接給出最終答案 請(qǐng)以JSON格式回復(fù)包含你的思考過程和行動(dòng)決策。 return prompt def execute_tool(self, tool_name: str, tool_input: Dict) - str: 執(zhí)行工具 if tool_name not in self.tool_map: return f錯(cuò)誤: 工具 {tool_name} 不存在 tool self.tool_map[tool_name] try: return tool.execute(**tool_input) except Exception as e: return f工具執(zhí)行錯(cuò)誤: {str(e)} def run(self, user_input: str) - str: 運(yùn)行Agent處理用戶輸入 self.memory.add_message(user, user_input) self.iteration_count 0 while self.iteration_count self.max_iterations: self.iteration_count 1 # 規(guī)劃下一步行動(dòng) plan self.plan_action(user_input) self.memory.add_message(system, f思考: {plan.get(thought, )}) action plan.get(action, ).lower() if action think: # 繼續(xù)思考不執(zhí)行工具 continue elif action use_tool: tool_name plan.get(tool_name) tool_input plan.get(tool_input, {}) if not tool_name: self.memory.add_message(system, 錯(cuò)誤: 未指定工具名稱) continue # 執(zhí)行工具 tool_result self.execute_tool(tool_name, tool_input) self.memory.add_message(system, f工具 {tool_name} 結(jié)果: {tool_result}) # 將工具結(jié)果作為新的用戶輸入繼續(xù)處理 user_input f工具執(zhí)行結(jié)果: {tool_result}. 請(qǐng)基于此繼續(xù)分析原始問題: {user_input} elif action final_answer: final_response plan.get(response, ) self.memory.add_message(assistant, final_response) return final_response else: self.memory.add_message(system, f未知行動(dòng): {action}) continue # 達(dá)到最大迭代次數(shù) timeout_response 抱歉經(jīng)過多次嘗試仍無法完全解決您的問題。建議您提供更具體的信息或嘗試其他方式。 self.memory.add_message(assistant, timeout_response) return timeout_response4.2 工具管理器擴(kuò)展為了更好管理工具我們添加一個(gè)工具管理器# core/tool_manager.py from typing import List, Dict, Any from tools.base_tool import BaseTool class ToolManager: 工具管理器 def __init__(self): self.tools: Dict[str, BaseTool] {} def register_tool(self, tool: BaseTool): 注冊(cè)工具 self.tools[tool.name] tool def get_tool(self, name: str) - BaseTool: 獲取工具 return self.tools.get(name) def list_tools(self) - List[Dict]: 列出所有工具信息 return [tool.get_schema() for tool in self.tools.values()] def validate_tool_input(self, tool_name: str, input_params: Dict) - bool: 驗(yàn)證工具輸入?yún)?shù) if tool_name not in self.tools: return False tool self.tools[tool_name] required_params tool.parameters.get(required, []) # 檢查必需參數(shù) for param in required_params: if param not in input_params: return False return True5. 完整實(shí)戰(zhàn)案例數(shù)學(xué)問題求解Agent5.1 創(chuàng)建增強(qiáng)版計(jì)算器工具首先創(chuàng)建一個(gè)更安全的計(jì)算器工具# tools/advanced_calculator.py import math import re from tools.base_tool import BaseTool class AdvancedCalculatorTool(BaseTool): 增強(qiáng)版計(jì)算器工具 def __init__(self): super().__init__( nameadvanced_calculator, description執(zhí)行復(fù)雜的數(shù)學(xué)計(jì)算支持三角函數(shù)、對(duì)數(shù)等, parameters{ type: object, properties: { expression: { type: string, description: 數(shù)學(xué)表達(dá)式如: sin(30) log(100) } }, required: [expression] } ) # 安全允許的函數(shù)和常量 self.safe_dict { abs: abs, min: min, max: max, round: round, sin: math.sin, cos: math.cos, tan: math.tan, log: math.log, log10: math.log10, sqrt: math.sqrt, pi: math.pi, e: math.e } def execute(self, expression: str) - str: 安全執(zhí)行數(shù)學(xué)計(jì)算 try: # 安全檢查只允許數(shù)字、運(yùn)算符和預(yù)定義函數(shù) pattern r^[0-9\-*/().\sabcdefghijklmnopqrstuvwxyz,]$ if not re.match(pattern, expression.lower()): return 錯(cuò)誤: 表達(dá)式包含不安全字符 # 替換數(shù)學(xué)函數(shù)名為安全版本 safe_expression expression.lower() for func_name in self.safe_dict.keys(): safe_expression safe_expression.replace(func_name, fsafe_dict[{func_name}]) # 執(zhí)行計(jì)算 result eval(safe_expression, {__builtins__: {}}, {safe_dict: self.safe_dict}) return f計(jì)算結(jié)果: {expression} {result} except Exception as e: return f計(jì)算錯(cuò)誤: {str(e)}5.2 創(chuàng)建網(wǎng)絡(luò)搜索工具模擬由于安全考慮我們創(chuàng)建一個(gè)模擬的網(wǎng)絡(luò)搜索工具# tools/web_search.py from tools.base_tool import BaseTool class WebSearchTool(BaseTool): 模擬網(wǎng)絡(luò)搜索工具 def __init__(self): super().__init__( nameweb_search, description搜索網(wǎng)絡(luò)信息模擬版本, parameters{ type: object, properties: { query: { type: string, description: 搜索關(guān)鍵詞 } }, required: [query] } ) # 模擬的知識(shí)庫 self.knowledge_base { python: Python是一種高級(jí)編程語言以簡(jiǎn)潔易讀著稱。, 人工智能: 人工智能是計(jì)算機(jī)科學(xué)的一個(gè)分支研究如何使機(jī)器智能地行動(dòng)。, 機(jī)器學(xué)習(xí): 機(jī)器學(xué)習(xí)是人工智能的子領(lǐng)域讓計(jì)算機(jī)通過數(shù)據(jù)學(xué)習(xí)模式。 } def execute(self, query: str) - str: 模擬網(wǎng)絡(luò)搜索 query_lower query.lower() # 在模擬知識(shí)庫中查找 for keyword, info in self.knowledge_base.items(): if keyword in query_lower: return f搜索 {query} 的結(jié)果:\n{info} return f未找到關(guān)于 {query} 的詳細(xì)信息。這是一個(gè)模擬搜索工具實(shí)際使用時(shí)需要接入真實(shí)的搜索API。5.3 創(chuàng)建完整應(yīng)用示例現(xiàn)在創(chuàng)建一個(gè)完整的應(yīng)用示例# main.py from core.agent import SimpleAgent from tools.advanced_calculator import AdvancedCalculatorTool from tools.web_search import WebSearchTool from config.settings import Config def main(): 主函數(shù) try: Config.validate() # 創(chuàng)建Agent并注冊(cè)工具 agent SimpleAgent() agent.add_tool(AdvancedCalculatorTool()) agent.add_tool(WebSearchTool()) print( 自制Agent系統(tǒng)啟動(dòng) ) print(可用工具: 計(jì)算器, 網(wǎng)絡(luò)搜索) print(輸入 quit 退出程序\n) while True: user_input input(用戶: ).strip() if user_input.lower() in [quit, exit, 退出]: print(感謝使用) break if not user_input: continue print(Agent思考中...) response agent.run(user_input) print(fAgent: {response}\n) except Exception as e: print(f系統(tǒng)錯(cuò)誤: {e}) if __name__ __main__: main()5.4 運(yùn)行演示啟動(dòng)程序后進(jìn)行測(cè)試# 運(yùn)行程序 python main.py # 示例對(duì)話 用戶: 計(jì)算一下 2的10次方加上sin(30度) Agent思考中... Agent: 計(jì)算結(jié)果: 2**10 sin(30*pi/180) 1024 0.5 1024.5 用戶: 告訴我什么是機(jī)器學(xué)習(xí) Agent思考中... Agent: 搜索 機(jī)器學(xué)習(xí) 的結(jié)果: 機(jī)器學(xué)習(xí)是人工智能的子領(lǐng)域讓計(jì)算機(jī)通過數(shù)據(jù)學(xué)習(xí)模式。6. 高級(jí)功能擴(kuò)展6.1 添加反思機(jī)制讓Agent能夠評(píng)估自己的表現(xiàn)并改進(jìn)策略# core/reflection.py import json from .llm_client import LLMClient class ReflectionEngine: 反思引擎 def __init__(self): self.llm LLMClient() def analyze_performance(self, conversation_history, final_outcome): 分析執(zhí)行表現(xiàn) prompt f 請(qǐng)分析以下Agent對(duì)話的執(zhí)行效果 對(duì)話歷史: {json.dumps(conversation_history, ensure_asciiFalse, indent2)} 最終結(jié)果: {final_outcome} 請(qǐng)?jiān)u估 1. Agent是否成功解決了用戶問題 2. 工具使用是否合理 3. 思考過程是否有優(yōu)化空間 4. 給出具體的改進(jìn)建議。 return self.llm.generate_response([ {role: system, content: 你是一個(gè)經(jīng)驗(yàn)豐富的AI系統(tǒng)分析師}, {role: user, content: prompt} ])6.2 實(shí)現(xiàn)狀態(tài)持久化添加將Agent狀態(tài)保存到文件的功能# utils/persistence.py import json import pickle from datetime import datetime class PersistenceManager: 持久化管理器 staticmethod def save_agent_state(agent, filepath): 保存Agent狀態(tài) state { conversation_history: agent.memory.conversation_history, agent_state: agent.memory.agent_state, save_time: datetime.now().isoformat() } with open(filepath, w, encodingutf-8) as f: json.dump(state, f, ensure_asciiFalse, indent2) staticmethod def load_agent_state(agent, filepath): 加載Agent狀態(tài) try: with open(filepath, r, encodingutf-8) as f: state json.load(f) agent.memory.conversation_history state.get(conversation_history, []) agent.memory.agent_state state.get(agent_state, {}) return True except FileNotFoundError: return False7. 常見問題與解決方案7.1 性能優(yōu)化問題問題1: Agent響應(yīng)速度慢原因: LLM調(diào)用延遲、工具執(zhí)行時(shí)間長、迭代次數(shù)過多解決方案:設(shè)置合理的超時(shí)時(shí)間緩存常用工具結(jié)果優(yōu)化提示詞減少不必要的思考步驟# 優(yōu)化示例添加超時(shí)控制 import signal import time class TimeoutAgent(SimpleAgent): 帶超時(shí)控制的Agent def run_with_timeout(self, user_input, timeout30): 帶超時(shí)運(yùn)行 def timeout_handler(signum, frame): raise TimeoutError(Agent執(zhí)行超時(shí)) signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(timeout) try: result self.run(user_input) signal.alarm(0) # 取消超時(shí) return result except TimeoutError: return 執(zhí)行超時(shí)請(qǐng)簡(jiǎn)化問題或重試7.2 工具安全與錯(cuò)誤處理問題2: 工具執(zhí)行不安全風(fēng)險(xiǎn): 代碼注入、資源耗盡、敏感信息泄露防護(hù)措施:# 增強(qiáng)工具安全性 class SecureTool(BaseTool): 安全工具基類 def safe_execute(self, **kwargs): 安全執(zhí)行方法 # 參數(shù)驗(yàn)證 if not self.validate_input(kwargs): return 錯(cuò)誤: 參數(shù)驗(yàn)證失敗 # 資源限制 try: return self.execute_with_limits(**kwargs) except Exception as e: return f工具執(zhí)行失敗: {str(e)} def validate_input(self, input_params): 驗(yàn)證輸入?yún)?shù) # 檢查必需參數(shù) required self.parameters.get(required, []) for param in required: if param not in input_params: return False # 參數(shù)類型檢查 for param, value in input_params.items(): param_schema self.parameters[properties].get(param, {}) expected_type param_schema.get(type) if expected_type string and not isinstance(value, str): return False elif expected_type number and not isinstance(value, (int, float)): return False return True7.3 記憶管理優(yōu)化問題3: 對(duì)話歷史過長導(dǎo)致性能下降現(xiàn)象: 響應(yīng)變慢、token消耗增加、上下文理解混亂解決方案:# 智能記憶管理 class SmartMemory(Memory): 智能記憶管理系統(tǒng) def compress_history(self): 壓縮對(duì)話歷史 if len(self.conversation_history) self.max_history: return # 保留重要的系統(tǒng)消息和最近的對(duì)話 important_messages [] recent_messages self.conversation_history[-4:] # 保留最近2輪對(duì)話 for msg in self.conversation_history: if msg.get(role) system and 關(guān)鍵信息 in msg.get(content, ): important_messages.append(msg) # 合并重要消息和最近消息 self.conversation_history important_messages recent_messages def summarize_conversation(self, llm_client): 總結(jié)對(duì)話內(nèi)容 if len(self.conversation_history) 8: return prompt f請(qǐng)用一段話總結(jié)以下對(duì)話的核心內(nèi)容:\n{json.dumps(self.conversation_history[:-4], ensure_asciiFalse)} summary llm_client.generate_response([ {role: system, content: 你是一個(gè)專業(yè)的對(duì)話總結(jié)助手}, {role: user, content: prompt} ]) # 用總結(jié)替換舊的歷史記錄 self.conversation_history [ {role: system, content: f對(duì)話總結(jié): {summary}} ] self.conversation_history[-4:]8. 生產(chǎn)環(huán)境最佳實(shí)踐8.1 安全部署建議API密鑰管理:使用環(huán)境變量或?qū)I(yè)的密鑰管理服務(wù)定期輪換密鑰不同環(huán)境使用不同密鑰# 安全的密鑰管理示例 import os from cryptography.fernet import Fernet class SecureConfig: 安全配置管理 staticmethod def get_encrypted_key(): 獲取加密的API密鑰 key os.getenv(ENCRYPTION_KEY) encrypted_api_key os.getenv(ENCRYPTED_API_KEY) if key and encrypted_api_key: fernet Fernet(key.encode()) return fernet.decrypt(encrypted_api_key.encode()).decode() return os.getenv(API_KEY) # 回退到普通環(huán)境變量8.2 監(jiān)控與日志建立完整的監(jiān)控體系# utils/monitoring.py import logging import time from datetime import datetime class AgentMonitor: Agent監(jiān)控器 def __init__(self): self.logger logging.getLogger(agent_monitor) self.logger.setLevel(logging.INFO) # 添加文件處理器 handler logging.FileHandler(agent_operations.log, encodingutf-8) formatter logging.Formatter(%(asctime)s - %(levelname)s - %(message)s) handler.setFormatter(formatter) self.logger.addHandler(handler) def log_operation(self, operation, details): 記錄操作日志 self.logger.info(f{operation}: {details}) def log_performance(self, start_time, end_time, operation): 記錄性能數(shù)據(jù) duration end_time - start_time self.logger.info(f性能 - {operation}: {duration:.2f}秒) # 使用示例 monitor AgentMonitor() def monitored_run(agent, user_input): 帶監(jiān)控的運(yùn)行 start_time time.time() monitor.log_operation(user_input, user_input) try: result agent.run(user_input) end_time time.time() monitor.log_performance(start_time, end_time, agent_run) monitor.log_operation(agent_response, result) return result except Exception as e: monitor.log_operation(error, str(e)) raise8.3 性能優(yōu)化策略緩存優(yōu)化:from functools import lru_cache import hashlib class CachedLLMClient(LLMClient): 帶緩存的LLM客戶端 lru_cache(maxsize100) def generate_response(self, messages, temperature0.1): 帶緩存的響應(yīng)生成 # 生成緩存鍵 cache_key self._generate_cache_key(messages, temperature) return super().generate_response(messages, temperature) def _generate_cache_key(self, messages, temperature): 生成緩存鍵 content json.dumps(messages, sort_keysTrue) str(temperature) return hashlib.md5(content.encode()).hexdigest()通過本文的完整實(shí)現(xiàn)你已經(jīng)掌握了從零開始構(gòu)建Agent的核心技術(shù)。這種深度理解將幫助你在實(shí)際項(xiàng)目中更好地使用和定制Agent系統(tǒng)無論是基于現(xiàn)有框架還是完全自研都能游刃有余。建議下一步可以嘗試集成真實(shí)的API工具、實(shí)現(xiàn)多Agent協(xié)作、或者優(yōu)化Agent的決策算法。真正的技術(shù)能力來自于動(dòng)手實(shí)踐和不斷迭代現(xiàn)在就開始你的Agent開發(fā)之旅吧