AI Agent

深入探讨AI Agent的设计原理与开发实践，帮助你构建能够自主规划、调用工具、协同工作的智能代理系统

引言 AI Agent与Web3的融合代表了两个最前沿技术的交汇点。当AI Agent能够自主地与区块链交互，我们将迎来全新的应用范式。本文将探讨如何构建能够理解、操作和优化链上系统的智能体。 链上AI推理 去中心化AI推理网络 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 from typing import List, Dict import hashlib import json class OnChainInferenceNetwork: """链上AI推理网络""" def __init__(self, blockchain_rpc: str): self.web3 = Web3(Web3.HTTPProvider(blockchain_rpc)) self.private_key = os.getenv("PRIVATE_KEY") self.account = self.web3.eth.account.from_key(self.private_key) # 加载推理合约ABI self.inference_contract = self.web3.eth.contract( address="0x...", # AI推理合约地址 abi=[...] # 合约ABI ) def submit_inference_task( self, model_id: str, input_data: Dict, reward: int ) -> str: """提交推理任务""" # 准备任务数据 task_data = { "modelId": model_id, "input": input_data, "reward": reward, "timeout": 3600, # 1小时超时 "timestamp": int(time.time()) } # 计算任务哈希 task_hash = self._compute_task_hash(task_data) # 提交到链上 tx_hash = self.inference_contract.functions.submitTask( task_hash, json.dumps(input_data), reward, task_data["timeout"] ).transact({'from': self.account.address}) self.web3.eth.wait_for_transaction_receipt(tx_hash) return task_hash def submit_inference_result( self, task_hash: str, output_data: Dict ) -> str: """提交推理结果""" # 准备结果数据 result_hash = self._compute_result_hash({ "taskHash": task_hash, "output": output_data, "submitter": self.account.address }) # 提交结果 tx_hash = self.inference_contract.functions.submitResult( task_hash, json.dumps(output_data), result_hash ).transact({'from': self.account.address}) self.web3.eth.wait_for_transaction_receipt(tx_hash) return tx_hash def claim_reward(self, task_hash: str) -> str: """领取奖励""" # 检查任务是否完成 task = self.inference_contract.functions.tasks(task_hash).call() if not task["completed"]: raise Exception("Task not completed yet") # 领取奖励 tx_hash = self.inference_contract.functions.claimReward( task_hash ).transact({'from': self.account.address}) self.web3.eth.wait_for_transaction_receipt(tx_hash) return tx_hash def verify_result( self, task_hash: str, output_data: Dict ) -> bool: """验证结果""" # 从链上获取任务 task = self.inference_contract.functions.tasks(task_hash).call() # 计算期望的输出哈希 expected_hash = self._compute_output_hash( task["input"], task["modelId"] ) # 验证结果哈希 result_hash = self._compute_result_hash({ "taskHash": task_hash, "output": output_data }) return result_hash == expected_hash def _compute_task_hash(self, task_data: Dict) -> str: """计算任务哈希""" data_string = json.dumps(task_data, sort_keys=True) return hashlib.sha256(data_string.encode()).hexdigest() def _compute_result_hash(self, result_data: Dict) -> str: """计算结果哈希""" data_string = json.dumps(result_data, sort_keys=True) return hashlib.sha256(data_string.encode()).hexdigest() def _compute_output_hash(self, input_data: Dict, model_id: str) -> str: """计算输出哈希（模拟AI推理）""" # 实际应用中，这里应该运行AI模型 # 这里简化处理 output = self._run_model(input_data, model_id) return hashlib.sha256(json.dumps(output).encode()).hexdigest() def _run_model(self, input_data: Dict, model_id: str) -> Dict: """运行AI模型""" # 实际应用中，这里应该调用真实的AI模型 # 可以使用OpenAI API、本地模型等 if model_id == "text-classifier": return self._classify_text(input_data["text"]) elif model_id == "image-analyzer": return self._analyze_image(input_data["imageUrl"]) elif model_id == "sentiment-analyzer": return self._analyze_sentiment(input_data["text"]) else: raise Exception(f"Unknown model: {model_id}") def _classify_text(self, text: str) -> Dict: """文本分类""" # 简化实现，实际应该调用真实模型 categories = { "technology": 0.8, "finance": 0.6, "sports": 0.1 } predicted_category = max(categories, key=categories.get) return { "category": predicted_category, "confidence": categories[predicted_category] } def _analyze_image(self, image_url: str) -> Dict: """图像分析""" return { "objects": ["person", "car", "building"], "scene": "street", "confidence": 0.95 } def _analyze_sentiment(self, text: str) -> Dict: """情感分析""" # 简化实现 positive_words = ["good", "great", "excellent", "happy"] negative_words = ["bad", "terrible", "awful", "sad"] words = text.lower().split() positive_count = sum(1 for word in words if word in positive_words) negative_count = sum(1 for word in words if word in negative_words) if positive_count > negative_count: sentiment = "positive" elif negative_count > positive_count: sentiment = "negative" else: sentiment = "neutral" return { "sentiment": sentiment, "score": (positive_count - negative_count) / len(words) } 智能合约与AI协作 AI辅助的智能合约审计 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 from transformers import AutoTokenizer, AutoModelForCausalLM from typing import List, Dict class AIContractAuditor: """AI智能合约审计助手""" def __init__(self, model_name="microsoft/CodeGPT-small"): self.tokenizer = AutoTokenizer.from_pretrained(model_name) self.model = AutoModelForCausalLM.from_pretrained(model_name) def audit_contract( self, contract_source: str ) -> Dict: """审计智能合约""" # 1. 代码结构分析 structure_analysis = self._analyze_structure(contract_source) # 2. 漏洞检测 vulnerabilities = self._detect_vulnerabilities(contract_source) # 3. 最佳实践检查 best_practices = self._check_best_practices(contract_source) # 4. Gas优化建议 gas_optimization = self._suggest_gas_optimization(contract_source) # 5. 生成审计报告 report = self._generate_report({ "structure": structure_analysis, "vulnerabilities": vulnerabilities, "best_practices": best_practices, "gas_optimization": gas_optimization }) return report def _analyze_structure(self, source: str) -> Dict: """分析合约结构""" prompt = f""" Analyze the following smart contract code structure: {source} Provide: 1. Contract architecture 2. Key functions and their roles 3. State variables and their purposes 4. Access control mechanisms 5. External dependencies """ response = self._generate(prompt) return { "architecture": self._parse_response(response, "architecture"), "functions": self._parse_response(response, "functions"), "state_variables": self._parse_response(response, "state_variables"), "access_control": self._parse_response(response, "access_control"), "dependencies": self._parse_response(response, "dependencies") } def _detect_vulnerabilities(self, source: str) -> List[Dict]: """检测漏洞""" known_vulnerabilities = { "reentrancy": { "patterns": [ r"\.call\{.*value:\s*msg\.value", r"\.send\{.*value:\s*msg\.value" ], "severity": "critical", "description": "Reentrancy vulnerability detected" }, "overflow": { "patterns": [ r"uint256.*=.*\+.*(?!\.add\()", r"uint256.*=.*-.*(?!\.sub\()" ], "severity": "high", "description": "Potential integer overflow/underflow" }, "access_control": { "patterns": [ r"function\s+\w+\s*\(\s*\)\s*public(?!\s*onlyOwner|onlyRole)", r"tx\.origin" ], "severity": "high", "description": "Weak access control" }, "unchecked_call": { "patterns": [ r"\.call\s*\(", r"\.send\s*\(" ], "severity": "medium", "description": "Unchecked external call" } } detected = [] for vuln_type, vuln_info in known_vulnerabilities.items(): for pattern in vuln_info["patterns"]: matches = re.finditer(pattern, source) for match in matches: detected.append({ "type": vuln_type, "severity": vuln_info["severity"], "description": vuln_info["description"], "location": match.span(), "code_snippet": source[match.start()-20:match.end()+20] }) return detected def _check_best_practices(self, source: str) -> List[Dict]: """检查最佳实践""" checks = { "uses_safe_math": r"SafeMath|\.add\(|\.sub\(" in source, "has_reentrancy_guard": r"ReentrancyGuard|nonReentrant" in source, "uses_openzeppelin": r"@openzeppelin" in source, "has_events": r"event\s+\w+" in source, "uses_checks_effects_interactions": r"Checks-Effects-Interactions" in source, "has_pause": r"whenNotPaused|Pausable" in source, "has_timelock": r"TimelockController|releaseTimeLock" in source } results = [] for check_name, check_result in checks.items(): results.append({ "check": check_name, "passed": check_result, "description": self._get_check_description(check_name) }) return results def _suggest_gas_optimization(self, source: str) -> List[str]: """建议Gas优化""" optimizations = [] # 检查循环 if "for (" in source: optimizations.append("Consider using unchecked blocks for loop iterations") # 检查storage操作 if re.search(r"uint256\s+public\s+\w+", source): optimizations.append("Consider packing struct variables to save storage") # 检查重复计算 if re.search(r"keccak256\(", source): optimizations.append("Cache keccak256 results in local variables") # 检查memory vs storage if re.search(r".*\.\w+\s*=\s*\w+\[.*\]\s*\+\s*1", source): optimizations.append("Consider using calldata instead of memory for arrays") return optimizations def _generate_report(self, audit_data: Dict) -> Dict: """生成审计报告""" # 计算风险评分 risk_score = self._calculate_risk_score(audit_data) # 生成总结 summary = self._generate_summary(audit_data, risk_score) # 生成修复建议 recommendations = self._generate_recommendations(audit_data) return { "risk_score": risk_score, "summary": summary, "vulnerabilities": audit_data["vulnerabilities"], "best_practices": audit_data["best_practices"], "optimizations": audit_data["gas_optimization"], "recommendations": recommendations } def _calculate_risk_score(self, audit_data: Dict) -> int: """计算风险评分（0-100）""" score = 100 for vuln in audit_data["vulnerabilities"]: if vuln["severity"] == "critical": score -= 30 elif vuln["severity"] == "high": score -= 15 elif vuln["severity"] == "medium": score -= 5 elif vuln["severity"] == "low": score -= 2 return max(score, 0) def _generate_summary(self, audit_data: Dict, risk_score: int) -> str: """生成审计总结""" vuln_count = len(audit_data["vulnerabilities"]) critical_count = sum(1 for v in audit_data["vulnerabilities"] if v["severity"] == "critical") summary = f""" Smart Contract Audit Summary ========================== Risk Score: {risk_score}/100 Total Vulnerabilities: {vuln_count} Critical Issues: {critical_count} """ if risk_score >= 80: summary += "Overall Assessment: LOW RISK" elif risk_score >= 50: summary += "Overall Assessment: MEDIUM RISK" else: summary += "Overall Assessment: HIGH RISK" return summary def _generate_recommendations(self, audit_data: Dict) -> List[str]: """生成修复建议""" recommendations = [] for vuln in audit_data["vulnerabilities"]: if vuln["type"] == "reentrancy": recommendations.append( "Use ReentrancyGuard or implement Checks-Effects-Interactions pattern" ) elif vuln["type"] == "overflow": recommendations.append( "Use Solidity 0.8.0+ or SafeMath library for arithmetic operations" ) elif vuln["type"] == "access_control": recommendations.append( "Implement proper access control using onlyOwner or role-based access" ) return recommendations def _generate(self, prompt: str) -> str: """生成文本""" inputs = self.tokenizer(prompt, return_tensors="pt") with torch.no_grad(): outputs = self.model.generate( **inputs, max_new_tokens=500, temperature=0.3, do_sample=True ) response = self.tokenizer.decode(outputs[0], skip_special_tokens=True) return response 自主交易Agent DeFi交易Agent 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 from typing import List, Dict, Optional from datetime import datetime import ccxt import pandas as pd import numpy as np class DeFiTradingAgent: """DeFi交易Agent""" def __init__( self, initial_capital: float, exchanges: List[str], llm_model: str = "gpt-4" ): self.capital = initial_capital self.portfolio = {} # {token: amount} self.exchanges = {} self.trade_history = [] # 初始化交易所连接 for exchange_name in exchanges: if exchange_name == "uniswap": exchange = ccxt.uniswap({ "enableRateLimit": True }) elif exchange_name == "pancakeswap": exchange = ccxt.pancakeswap({ "enableRateLimit": True }) else: exchange = ccxt.binance({ "enableRateLimit": True }) self.exchanges[exchange_name] = exchange # 初始化LLM self.llm = self._init_llm(llm_model) def _init_llm(self, model_name: str): """初始化LLM""" # 实际应用中，这里应该连接到真实的LLM API # 或运行本地模型 from transformers import AutoTokenizer, AutoModelForCausalLM model = AutoModelForCausalLM.from_pretrained(model_name) tokenizer = AutoTokenizer.from_pretrained(model_name) return { "model": model, "tokenizer": tokenizer } def analyze_market( self, tokens: List[str], timeframe: str = "1h" ) -> Dict: """分析市场""" # 获取市场数据 market_data = self._fetch_market_data(tokens, timeframe) # 技术分析 ta_analysis = self._technical_analysis(market_data) # 使用LLM生成市场洞察 market_insight = self._generate_market_insight( market_data, ta_analysis ) return { "market_data": market_data, "technical_analysis": ta_analysis, "insight": market_insight } def _fetch_market_data( self, tokens: List[str], timeframe: str ) -> Dict: """获取市场数据""" data = {} for token in tokens: # 从各个交易所获取数据 for exchange_name, exchange in self.exchanges.items(): try: ohlcv = exchange.fetch_ohlcv( f"{token}/USDT", timeframe, limit=100 ) if token not in data: data[token] = [] # 转换为DataFrame df = pd.DataFrame( ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'] ) df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') data[token].append(df) break # 使用第一个成功的数据源 except Exception as e: print(f"Error fetching data for {token}: {e}") return data def _technical_analysis(self, market_data: Dict) -> Dict: """技术分析""" analysis = {} for token, dfs in market_data.items(): if not dfs: continue df = dfs[0] # 使用第一个数据源 # 计算技术指标 df['sma_20'] = df['close'].rolling(window=20).mean() df['sma_50'] = df['close'].rolling(window=50).mean() df['rsi'] = self._calculate_rsi(df['close'], 14) df['macd'] = self._calculate_macd(df['close']) # 趋势分析 latest_close = df['close'].iloc[-1] sma_20 = df['sma_20'].iloc[-1] sma_50 = df['sma_50'].iloc[-1] trend = "bullish" if latest_close > sma_20 > sma_50 else "bearish" analysis[token] = { "current_price": latest_close, "sma_20": sma_20, "sma_50": sma_50, "rsi": df['rsi'].iloc[-1], "trend": trend, "support_levels": self._find_support_levels(df), "resistance_levels": self._find_resistance_levels(df) } return analysis def _calculate_rsi(self, prices: pd.Series, period: int = 14) -> pd.Series: """计算RSI""" delta = prices.diff() gain = (delta.where(delta > 0, 0)).rolling(window=period).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean() rs = gain / loss rsi = 100 - (100 / (1 + rs)) return rsi def _calculate_macd(self, prices: pd.Series) -> Dict: """计算MACD""" exp1 = prices.ewm(span=12, adjust=False).mean() exp2 = prices.ewm(span=26, adjust=False).mean() macd = exp1 - exp2 signal = macd.ewm(span=9, adjust=False).mean() histogram = macd - signal return { "macd": macd.iloc[-1], "signal": signal.iloc[-1], "histogram": histogram.iloc[-1] } def _find_support_levels(self, df: pd.DataFrame) -> List[float]: """寻找支撑位""" # 简化实现：使用局部最小值 from scipy.signal import argrelextrema prices = df['close'].values local_min = argrelextrema(prices, np.less, order=20) support_levels = sorted(prices[local_min]) return support_levels[-5:] # 返回最近的5个支撑位 def _find_resistance_levels(self, df: pd.DataFrame) -> List[float]: """寻找阻力位""" from scipy.signal import argrelextrema prices = df['close'].values local_max = argrelextrema(prices, np.greater, order=20) resistance_levels = sorted(prices[local_max], reverse=True) return resistance_levels[-5:] # 返回最近的5个阻力位 def _generate_market_insight( self, market_data: Dict, ta_analysis: Dict ) -> str: """生成市场洞察""" # 准备prompt prompt = f""" Analyze the following cryptocurrency market data and provide trading insights: Technical Analysis: {json.dumps(ta_analysis, indent=2)} Based on this analysis, provide: 1. Market trend analysis 2. Key support and resistance levels 3. Trading recommendations 4. Risk factors to consider 5. Optimal entry and exit points Be specific and actionable. """ # 调用LLM response = self._generate(prompt) return response def execute_trade( self, exchange: str, symbol: str, side: str, amount: float, price: Optional[float] = None ) -> Dict: """执行交易""" exchange_obj = self.exchanges[exchange] try: if side == "buy": # 限价买单 if price: order = exchange_obj.create_limit_buy_order( symbol, amount, price ) else: # 市价买单 order = exchange_obj.create_market_buy_order( symbol, amount ) else: # 卖单 if price: order = exchange_obj.create_limit_sell_order( symbol, amount, price ) else: order = exchange_obj.create_market_sell_order( symbol, amount ) # 记录交易 trade_record = { "exchange": exchange, "symbol": symbol, "side": side, "amount": amount, "price": price, "timestamp": datetime.now().isoformat(), "status": "executed" } self.trade_history.append(trade_record) return trade_record except Exception as e: print(f"Trade execution failed: {e}") return { "status": "failed", "error": str(e) } def run_strategy( self, strategy_config: Dict ) -> List[Dict]: """运行交易策略""" # 1. 分析市场 market_analysis = self.analyze_market( strategy_config["tokens"], strategy_config.get("timeframe", "1h") ) # 2. 生成交易信号 signals = self._generate_trading_signals( market_analysis, strategy_config ) # 3. 执行交易 executed_trades = [] for signal in signals: if signal["action"] == "hold": continue trade = self.execute_trade( exchange=signal["exchange"], symbol=signal["symbol"], side=signal["side"], amount=signal["amount"], price=signal.get("price") ) if trade.get("status") == "executed": executed_trades.append(trade) return executed_trades def _generate_trading_signals( self, market_analysis: Dict, strategy_config: Dict ) -> List[Dict]: """生成交易信号""" signals = [] ta_analysis = market_analysis["technical_analysis"] for token, analysis in ta_analysis.items(): # 简单的移动平均策略 if (analysis["trend"] == "bullish" and analysis["rsi"] < 70 and analysis["current_price"] > analysis["sma_20"]): signals.append({ "action": "buy", "exchange": "uniswap", "symbol": f"{token}/USDT", "side": "buy", "amount": strategy_config.get("trade_size", 100), "reason": "Bullish trend with RSI below overbought" }) elif (analysis["trend"] == "bearish" and analysis["rsi"] > 30): signals.append({ "action": "sell", "exchange": "uniswap", "symbol": f"{token}/USDT", "side": "sell", "amount": strategy_config.get("trade_size", 100), "reason": "Bearish trend detected" }) return signals DAO治理Agent 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 from typing import List, Dict, Optional class DAOGovernanceAgent: """DAO治理Agent""" def __init__( self, dao_address: str, llm_model: str = "gpt-4" ): self.dao_address = dao_address self.llm = self._init_llm(llm_model) # 治理历史 self.governance_history = [] # 提案分析 self.proposals_db = {} def analyze_proposal( self, proposal_data: Dict ) -> Dict: """分析提案""" # 1. 提取关键信息 key_info = self._extract_proposal_info(proposal_data) # 2. 风险评估 risk_assessment = self._assess_risk(proposal_data) # 3. 财务影响分析 financial_impact = self._analyze_financial_impact(proposal_data) # 4. 生成投票建议 voting_recommendation = self._generate_voting_recommendation({ "key_info": key_info, "risk_assessment": risk_assessment, "financial_impact": financial_impact }) return { "proposal_id": proposal_data["id"], "key_info": key_info, "risk_assessment": risk_assessment, "financial_impact": financial_impact, "recommendation": voting_recommendation } def _extract_proposal_info(self, proposal: Dict) -> Dict: """提取提案关键信息""" prompt = f""" Extract key information from this DAO proposal: Title: {proposal.get('title', '')} Description: {proposal.get('description', '')} Please extract: 1. Proposal type (e.g., parameter change, spending, governance change) 2. Key changes proposed 3. Affected stakeholders 4. Implementation timeline 5. Required resources """ response = self._generate(prompt) # 解析LLM响应 key_info = { "type": self._parse_field(response, "Proposal type"), "changes": self._parse_field(response, "Key changes"), "stakeholders": self._parse_field(response, "Stakeholders"), "timeline": self._parse_field(response, "Timeline"), "resources": self._parse_field(response, "Resources") } return key_info def _assess_risk(self, proposal: Dict) -> Dict: """评估风险""" risk_factors = [] # 检查提案类型 proposal_type = self._extract_proposal_type(proposal) if proposal_type == "spending": # 检查金额 amount = self._extract_amount(proposal) if amount > 1000000: risk_factors.append({ "type": "financial", "severity": "high", "description": "Large expenditure proposed" }) elif proposal_type == "parameter_change": # 检查参数范围 params = self._extract_parameters(proposal) if self._is_risk_parameter_change(params): risk_factors.append({ "type": "governance", "severity": "medium", "description": "Parameter changes may affect protocol stability" }) return { "risk_score": self._calculate_risk_score(risk_factors), "risk_factors": risk_factors, "mitigation_strategies": self._suggest_mitigation(risk_factors) } def _generate_voting_recommendation(self, analysis: Dict) -> Dict: """生成投票建议""" prompt = f""" Based on the following DAO proposal analysis: Key Information: {json.dumps(analysis['key_info'], indent=2)} Risk Assessment: {json.dumps(analysis['risk_assessment'], indent=2)} Financial Impact: {json.dumps(analysis['financial_impact'], indent=2)} Provide a voting recommendation: 1. Vote: For/Against/Abstain 2. Confidence: High/Medium/Low 3. Reasoning: Detailed explanation 4. Conditions: Any conditions for changing the vote Consider: - Long-term sustainability - Community impact - Financial health - Innovation vs stability """ response = self._generate(prompt) # 解析建议 recommendation = { "vote": self._parse_field(response, "Vote"), "confidence": self._parse_field(response, "Confidence"), "reasoning": self._parse_field(response, "Reasoning"), "conditions": self._parse_field(response, "Conditions") } return recommendation def automate_governance(self) -> None: """自动化治理决策""" # 获取待处理提案 pending_proposals = self._fetch_pending_proposals() for proposal in pending_proposals: # 分析提案 analysis = self.analyze_proposal(proposal) # 根据建议自动投票 if analysis["recommendation"]["vote"].lower() == "for": self._cast_vote( proposal["id"], "for", analysis["recommendation"]["reasoning"] ) def _fetch_pending_proposals(self) -> List[Dict]: """获取待处理提案""" # 实际应用中，这里应该从链上或DAO的API获取 # 简化实现 proposals = [] # 示例提案 proposals.append({ "id": "proposal-123", "title": "Grant Program Funding", "description": "Allocate $500,000 for grants", "status": "pending", "voting_deadline": datetime.now() + timedelta(days=7) }) return proposals def _cast_vote(self, proposal_id: str, vote: str, reason: str) -> str: """投票""" # 实际应用中，这里应该调用链上治理合约 print(f"Voting {vote} on proposal {proposal_id}") print(f"Reason: {reason}") return f"voted-{vote}-{proposal_id}" 总结 AI Agent与Web3的融合将开启全新的应用范式： ...

引言 AI Agent的强大能力来自于其工作流编排能力——将复杂任务分解为多个步骤，并智能地协调执行。从简单的链式调用到复杂的多Agent协作，工作流编排是Agent系统的核心。本文将深入探讨主流的Agent编排框架和实战技巧。 LangChain Chains 基础Chain 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 from langchain.chains import LLMChain from langchain.prompts import PromptTemplate from langchain_openai import OpenAI # 创建LLM llm = OpenAI(temperature=0) # 创建Prompt模板 prompt_template = PromptTemplate( input_variables=["product"], template="为{product}写一段吸引人的产品描述。" ) # 创建Chain chain = LLMChain(llm=llm, prompt=prompt_template) # 运行 description = chain.run(product="智能手表") print(description) Sequential Chain 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 from langchain.chains import SequentialChain # Chain 1: 生成产品名称 name_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["product_type"], template="为{product_type}产品想一个创意名称，只返回名称。" ), output_key="product_name" ) # Chain 2: 生成Slogan slogan_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["product_name"], template="为{product_name}写一句简短有力的广告语。" ), output_key="slogan" ) # Chain 3: 生成完整描述 description_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["product_name", "slogan"], template="产品名称：{product_name}\n广告语：{slogan}\n请基于以上信息写一段100字的产品描述。" ), output_key="description" ) # 组合Chain overall_chain = SequentialChain( chains=[name_chain, slogan_chain, description_chain], input_variables=["product_type"], output_variables=["product_name", "slogan", "description"] ) # 执行 result = overall_chain("智能手表") print(result) # { # 'product_name': 'TimePulse', # 'slogan': 'TimePulse - 让时间更有价值', # 'description': '...' # } Conditional Chain 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 from langchain.chains import TransformChain # 条件判断函数 def categorize_price(inputs: dict) -> dict: price = inputs.get('price', 0) if price < 100: category = "low" elif price < 500: category = "medium" else: category = "high" return {"price_category": category} # 条件Chain price_categorize_chain = TransformChain( transform=categorize_price, input_variables=["price"], output_variables=["price_category"] ) # 不同价格段的不同处理 low_price_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["product"], template="{product}是经济实惠的选择，适合预算有限的用户。写一段强调性价比的描述。" ) ) high_price_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["product"], template="{product}是高端产品，强调其品质和独特价值。写一段描述。" ) ) from langchain.chains import RouterChain # 路由Chain router_chain = RouterChain( chains={ "low": low_price_chain, "medium": medium_price_chain, "high": high_price_chain }, default_chain=medium_price_chain ) LCEL (LangChain Expression Language) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser # 使用LCEL构建Chain prompt = PromptTemplate.from_template( "Tell me a joke about {topic}" ) # 使用管道操作符 (|) chain = ( prompt | llm | StrOutputParser() ) # 等价于 chain = prompt | llm | StrOutputParser() # 执行 result = chain.invoke({"topic": "programming"}) print(result) # 复杂的LCEL示例 from langchain_community.utilities import WikipediaSearch wiki_search = WikipediaSearch() research_chain = ( { "context": lambda x: wiki_search.run(x["topic"]), "topic": RunnablePassthrough() } | PromptTemplate.from_template( "Topic: {topic}\n\nResearch: {context}\n\nBased on the research, explain {topic} in simple terms." ) | llm | StrOutputParser() ) LangChain Agents ReAct Agent 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 from langchain.agents import AgentExecutor, create_react_agent from langchain.tools import Tool from langchain_openai import OpenAI # 定义工具 def search_tool(query: str) -> str: """搜索工具""" # 实际应用中调用搜索API return f"搜索'{query}'的结果：..." def calculator_tool(expression: str) -> str: """计算器工具""" try: result = eval(expression) return f"计算结果：{result}" except: return "计算错误" tools = [ Tool( name="Search", func=search_tool, description="用于搜索网络信息，输入应该是搜索查询" ), Tool( name="Calculator", func=calculator_tool, description="用于数学计算，输入应该是数学表达式" ) ] # 创建Agent llm = OpenAI(temperature=0) prompt = PromptTemplate.from_template( """Answer the following questions as best you can. You have access to the following tools: {tools} Use the following format: Question: the input question you must answer Thought: you should always think about what to do Action: the action to take, should be one of [{tool_names}] Action Input: the input to the action Observation: the result of the action ... (this Thought/Action/Action Input/Observation can repeat N times) Thought: I now know the final answer Final Answer: the final answer to the original input question Begin! Question: {input} Thought: {agent_scratchpad}""" ) agent = create_react_agent( llm=llm, tools=tools, prompt=prompt ) # 创建Agent执行器 agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, max_iterations=5 ) # 执行 result = agent_executor.invoke({ "input": "苹果公司现在的股价是多少？如果我有100股，总价值多少？" }) print(result["output"]) Custom Agent 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 from langchain.agents import AgentExecutor, BaseSingleActionAgent from langchain_openai import BaseOpenAI class CustomAgent(BaseSingleActionAgent): """自定义Agent""" llm: BaseOpenAI tools: list[Tool] @property def input_keys(self): return ["input"] def plan( self, intermediate_steps: list[tuple[str, str]], **kwargs: Any ) -> tuple[AgentAction, str]: """规划下一步行动""" user_input = kwargs["input"] # 构建思考过程 thoughts = "" for action, observation in intermediate_steps: thoughts += f"Action: {action.tool}\n" thoughts += f"Input: {action.tool_input}\n" thoughts += f"Observation: {observation}\n" # 让LLM决定下一步 prompt = f""" 输入: {user_input} 之前的步骤: {thoughts} 可用工具: {[tool.name for tool in self.tools]} 请决定下一步行动，格式为： Action: [工具名称] Input: [工具输入] 或如果已完成： Final Answer: [最终答案] """ response = self.llm.predict(prompt) # 解析响应 if "Final Answer:" in response: final_answer = response.split("Final Answer:")[-1].strip() return AgentAction( tool="FINAL", tool_input=final_answer, log=response ), final_answer else: # 提取Action和Input action_line = [l for l in response.split("\n") if "Action:" in l][0] input_line = [l for l in response.split("\n") if "Input:" in l][0] tool_name = action_line.split("Action:")[-1].strip() tool_input = input_line.split("Input:")[-1].strip() return AgentAction( tool=tool_name, tool_input=tool_input, log=response ), "" async def aplan( self, intermediate_steps: list[tuple[str, str]], **kwargs: Any ) -> tuple[AgentAction, str]: """异步规划""" return self.plan(intermediate_steps, **kwargs) # 使用自定义Agent custom_agent = CustomAgent( llm=OpenAI(temperature=0), tools=tools ) agent_executor = AgentExecutor( agent=custom_agent, tools=tools ) result = agent_executor.invoke({"input": "查询北京今天的天气"}) AutoGPT模式 基础AutoGPT实现 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 from typing import List, Dict import json class AutoGPTAgent: """AutoGPT风格的Agent""" def __init__( self, name: str, role: str, goals: List[str], llm, tools: Dict[str, callable] ): self.name = name self.role = role self.goals = goals self.llm = llm self.tools = tools self.memory = [] self.task_list = [] def think(self) -> Dict: """思考下一步行动""" prompt = f""" Name: {self.name} Role: {self.role} Goals: {', '.join(self.goals)} Memory: {self.format_memory()} Current Tasks: {self.format_tasks()} 请决定下一步行动。返回JSON格式： {{ "thought": "思考过程", "reasoning": "推理过程", "plan": "计划", "criticism": "自我批评", "action": "行动名称", "action_input": "行动输入" }} """ response = self.llm.generate(prompt) try: return json.loads(response) except: return { "thought": "解析错误", "action": "finish", "action_input": "" } def execute(self, action: str, action_input: str) -> str: """执行行动""" if action == "finish": return "任务完成" if action in self.tools: result = self.tools[action](action_input) # 记录到记忆 self.memory.append({ "action": action, "input": action_input, "result": result }) return result else: return f"未知行动: {action}" def format_memory(self) -> str: """格式化记忆""" if not self.memory: return "No memories yet." return "\n".join([ f"- {m['action']}: {m['input']} -> {m['result'][:100]}" for m in self.memory[-5:] ]) def format_tasks(self) -> str: """格式化任务列表""" if not self.task_list: return "No tasks." return "\n".join([ f"{i+1}. {task}" for i, task in enumerate(self.task_list) ]) def run(self, max_iterations: int = 10) -> str: """运行Agent""" for i in range(max_iterations): # 思考 thought_process = self.think() print(f"\n=== Iteration {i+1} ===") print(f"Thought: {thought_process['thought']}") print(f"Reasoning: {thought_process['reasoning']}") print(f"Plan: {thought_process['plan']}") print(f"Criticism: {thought_process['criticism']}") # 执行 action = thought_process['action'] action_input = thought_process['action_input'] result = self.execute(action, action_input) print(f"Action: {action}") print(f"Result: {result[:200]}") # 检查是否完成 if action == "finish": return result return "达到最大迭代次数" # 使用示例 def search_web(query: str) -> str: """搜索网络""" return f"搜索'{query}'的结果..." def write_file(content: str) -> str: """写入文件""" return "文件已写入" def read_file(filename: str) -> str: """读取文件""" return f"文件{filename}的内容..." tools = { "search": search_web, "write": write_file, "read": read_file } agent = AutoGPTAgent( name="Researcher", role="AI研究员", goals=["研究最新AI技术", "生成研究报告"], llm=OpenAI(temperature=0), tools=tools ) result = agent.run() print(result) BabyAGI模式 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 class BabyAGI: """BabyAGI实现""" def __init__( self, objective: str, llm, tools: Dict[str, callable], max_iterations: int = 10 ): self.objective = objective self.llm = llm self.tools = tools self.max_iterations = max_iterations self.task_list = [] self.completed_tasks = [] def create_initial_tasks(self) -> List[str]: """创建初始任务列表""" prompt = f""" 目标: {self.objective} 请为这个目标创建一个任务列表。返回JSON数组格式： ["任务1", "任务2", "任务3"] """ response = self.llm.generate(prompt) try: tasks = json.loads(response) return tasks except: return ["研究相关资料", "分析问题", "制定方案"] def prioritize_tasks(self) -> List[str]: """任务优先级排序""" if not self.task_list: return [] prompt = f""" 目标: {self.objective} 当前任务列表: {json.dumps(self.task_list, ensure_ascii=False)} 已完成任务: {json.dumps(self.completed_tasks[-5:], ensure_ascii=False)} 请根据当前情况重新排列任务优先级。 返回JSON数组格式（从高到低）： ["任务1", "任务2", ...] """ response = self.llm.generate(prompt) try: return json.loads(response) except: return self.task_list def execute_task(self, task: str) -> str: """执行任务""" prompt = f""" 目标: {self.objective} 任务: {task} 请执行这个任务并返回结果。 如果需要使用工具，请说明： - search: 搜索信息 - calculate: 计算数据 - write: 写入内容 """ response = self.llm.generate(prompt) # 记录完成的任务 self.completed_tasks.append({ "task": task, "result": response }) return response def run(self) -> Dict: """运行BabyAGI""" # 创建初始任务 self.task_list = self.create_initial_tasks() for i in range(self.max_iterations): if not self.task_list: print("所有任务已完成！") break # 优先级排序 self.task_list = self.prioritize_tasks() # 执行第一个任务 current_task = self.task_list[0] print(f"\n=== 迭代 {i+1} ===") print(f"当前任务: {current_task}") result = self.execute_task(current_task) print(f"执行结果: {result[:200]}") # 从列表中移除 self.task_list.pop(0) return { "objective": self.objective, "completed_tasks": self.completed_tasks } # 使用 baby_agi = BabyAGI( objective="研究并总结2024年AI大模型的最新进展", llm=OpenAI(temperature=0), tools=tools ) result = baby_agi.run() CrewAI多Agent协作 Crew定义 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 from crewai import Agent, Task, Crew, Process # 定义Agent researcher = Agent( role='研究员', goal='研究最新的AI技术趋势', backstory="""你是一位经验丰富的AI研究员， 专注于追踪和分析最新的AI技术发展""", verbose=True, tools=[search_tool, wikipedia_tool] ) writer = Agent( role='技术作家', goal='将复杂的技术内容转化为易懂的文章', backstory="""你是一位技术写作专家， 擅长将技术细节转化为吸引人的内容""", verbose=True ) reviewer = Agent( role='内容审核员', goal='确保内容准确、完整、有价值', backstory="""你是一位资深的内容审核专家， 对技术内容的质量有极高要求""", verbose=True ) # 定义任务 research_task = Task( description="""研究2024年大语言模型的最新发展， 包括GPT-4、Claude、Gemini等模型的更新""", expected_output='详细的研究报告，包含关键发现和技术突破', agent=researcher ) write_task = Task( description="""基于研究报告，撰写一篇关于2024年LLM发展的技术文章。 文章应该面向技术读者，但保持通俗易懂""", expected_output='结构完整、内容丰富的技术文章（1000-1500字）', agent=writer ) review_task = Task( description="""审核技术文章，确保： 1. 技术准确性 2. 内容完整性 3. 可读性 4. 价值性 提供修改建议和最终评价""", expected_output='详细的审核报告，包含修改建议和最终评分', agent=reviewer ) # 创建Crew tech_crew = Crew( agents=[researcher, writer, reviewer], tasks=[research_task, write_task, review_task], process=Process.sequential, # 顺序执行 verbose=True ) # 执行 result = tech_crew.kickoff() print(result) 并行Process 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 # 定义并行任务 crew_parallel = Crew( agents=[agent1, agent2, agent3], tasks=[task1, task2, task3], process=Process.parallel, # 并行执行 verbose=True ) # 或者使用层级Process crew_hierarchical = Crew( agents=[manager_agent, worker_agent1, worker_agent2], tasks=[manager_task, worker_task1, worker_task2], process=Process.hierarchical, # 层级执行 manager_llm=OpenAI(temperature=0), verbose=True ) Agent编排框架对比 LangGraph 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 from langgraph.graph import StateGraph, END from typing import TypedDict # 定义状态 class AgentState(TypedDict): input: str research: str draft: str review: str final: str # 创建图 workflow = StateGraph(AgentState) # 添加节点 def research_node(state: AgentState) -> AgentState: result = researcher_agent.run(state["input"]) return {**state, "research": result} def write_node(state: AgentState) -> AgentState: result = writer_agent.run(state["research"]) return {**state, "draft": result} def review_node(state: AgentState) -> AgentState: result = reviewer_agent.run(state["draft"]) return {**state, "review": result} # 添加节点到图 workflow.add_node("researcher", research_node) workflow.add_node("writer", write_node) workflow.add_node("reviewer", review_node) # 添加边 workflow.set_entry_point("researcher") workflow.add_edge("researcher", "writer") workflow.add_edge("writer", "reviewer") workflow.add_edge("reviewer", END) # 编译图 app = workflow.compile() # 执行 result = app.invoke({"input": "研究AI最新进展"}) Semantic Kernel 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 from semantic_kernel import Kernel from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion from semantic_kernel.planning import SequentialPlanner # 初始化Kernel kernel = Kernel() kernel.add_chat_service( "chat-gpt", OpenAIChatCompletion("gpt-4", api_key="...") ) # 定义技能（Skill） from semantic_kernel.skill_definition import sk_function class ResearchSkills: @sk_function(description="搜索信息") def search(self, query: str) -> str: return f"搜索'{query}'的结果..." @sk_function(description="总结内容") def summarize(self, content: str) -> str: return f"总结: {content[:100]}..." # 注册技能 kernel.import_skill(ResearchSkills(), skill_name="research") # 创建计划器 planner = SequentialPlanner(kernel) # 执行计划 ask = "研究2024年AI技术进展并生成报告" plan = await planner.create_plan_async(ask) result = await plan.invoke_async(kernel) print(result) 实战案例 案例一：智能研究报告生成 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 class ReportGenerator: """智能报告生成器""" def __init__(self): self.llm = OpenAI(temperature=0) self.tools = self._init_tools() def _init_tools(self) -> Dict[str, callable]: return { "search": self._search, "analyze": self._analyze, "write": self._write, "format": self._format } def generate_report(self, topic: str) -> str: """生成报告""" # 阶段1：研究 research_data = self._research_stage(topic) # 阶段2：分析 analysis = self._analysis_stage(research_data) # 阶段3：撰写 draft = self._writing_stage(analysis) # 阶段4：审阅 final_report = self._review_stage(draft) return final_report def _research_stage(self, topic: str) -> Dict: """研究阶段""" # 生成研究计划 plan = self.llm.generate(f""" 为"{topic}"创建一个研究计划， 包含需要研究的关键点。 """) # 执行研究 research_data = {} for key_point in plan.split('\n'): if key_point.strip(): result = self._search(key_point.strip()) research_data[key_point.strip()] = result return research_data def _analysis_stage(self, data: Dict) -> str: """分析阶段""" prompt = f""" 分析以下研究数据： {json.dumps(data, ensure_ascii=False, indent=2)} 提供关键发现和洞察。 """ return self.llm.generate(prompt) def _writing_stage(self, analysis: str) -> str: """撰写阶段""" prompt = f""" 基于以下分析，撰写一份专业的研究报告： {analysis} 报告应该包含： 1. 执行摘要 2. 背景介绍 3. 主要发现 4. 结论和建议 """ return self.llm.generate(prompt) def _review_stage(self, draft: str) -> str: """审阅阶段""" prompt = f""" 审阅以下报告草稿： {draft} 提供改进建议并进行必要的修改。 """ reviewed = self.llm.generate(prompt) return reviewed def _search(self, query: str) -> str: """搜索实现""" # 实际调用搜索API return f"关于'{query}'的搜索结果..." 案例二：客户服务自动化 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 class CustomerServiceAgent: """客户服务Agent""" def __init__(self): self.llm = OpenAI(temperature=0.7) self.knowledge_base = self._load_kb() self.conversation_history = {} def handle_customer_query( self, customer_id: str, query: str ) -> str: """处理客户查询""" # 获取历史对话 history = self.conversation_history.get(customer_id, []) # 检索相关知识 relevant_docs = self._retrieve_knowledge(query) # 理解意图 intent = self._classify_intent(query) # 生成响应 response = self._generate_response( query=query, intent=intent, history=history, knowledge=relevant_docs ) # 更新历史 history.append({"role": "user", "content": query}) history.append({"role": "assistant", "content": response}) self.conversation_history[customer_id] = history[-10:] return response def _classify_intent(self, query: str) -> str: """分类意图""" prompt = f""" 分类以下客户查询的意图： 查询: {query} 可能的意图： 1. 产品咨询 2. 订单查询 3. 投诉建议 4. 售后服务 5. 其他 只返回意图名称。 """ return self.llm.generate(prompt).strip() def _retrieve_knowledge(self, query: str) -> List[str]: """检索相关知识""" # 使用向量搜索 # 简化示例 return [ doc for doc in self.knowledge_base if any(word in doc.lower() for word in query.lower().split()) ][:3] def _generate_response( self, query: str, intent: str, history: List[Dict], knowledge: List[str] ) -> str: """生成响应""" history_text = "\n".join([ f"{msg['role']}: {msg['content']}" for msg in history[-5:] ]) knowledge_text = "\n".join(knowledge) prompt = f""" 意图: {intent} 知识库: {knowledge_text} 对话历史: {history_text} 客户查询: {query} 请提供专业、友好的回复。 """ return self.llm.generate(prompt) 总结 AI Agent工作流编排是构建复杂AI应用的关键技术。从简单的LangChain Chains到复杂的Multi-Agent Systems，不同的框架和模式适用于不同的场景。 ...

引言 AI Agent（人工智能智能体）作为大语言模型最重要的应用范式之一，正在重塑我们与AI交互的方式。不同于传统的聊天机器人，AI Agent具备自主感知、决策和执行能力，能够使用工具、维护记忆、进行多步推理。本文将深入探讨AI Agent的架构设计，从理论到实践，帮助开发者构建生产级的智能体应用。 AI Agent核心概念 什么是AI Agent AI Agent是一个能够： 感知环境：理解用户输入和系统状态 推理决策：基于目标和上下文制定行动方案 执行工具：调用外部API和服务完成任务 记忆管理：维护短期和长期记忆 反思学习：从执行结果中学习和改进 Agent vs Chatbot 1 2 3 4 5 6 7 8 9 10 11 12 # 传统Chatbot chatbot_response = llm.generate("帮我查询天气") # 单轮对话，无状态，无法执行操作 # AI Agent agent = Agent( tools=[weather_api, calendar_api], memory=LongTermMemory(), planner=ReActPlanner() ) result = agent.run("帮我查明天天气，如果有雨则安排线上会议") # 多步推理，工具调用，状态管理 核心架构设计 1. 整体架构 ┌─────────────────────────────────────────────┐ │ User Interface │ └──────────────────┬──────────────────────────┘ │ ┌──────────────────▼──────────────────────────┐ │ Agent Core │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ Planner │──│ Executor │──│ Reflector│ │ │ └──────────┘ └──────┬───┘ └──────────┘ │ │ │ │ │ ┌─────────────────────┼─────────────────┐ │ │ │ Memory System │ │ │ │ ┌─────────┐ ┌─────────┐ ┌────────┐ │ │ │ │ │ShortTerm│ │LongTerm │ │Vector │ │ │ │ │ └─────────┘ └─────────┘ └────────┘ │ │ │ └────────────────────────────────────────┘ │ └──────────────────┬──────────────────────────┘ │ ┌──────────────────▼──────────────────────────┐ │ Tool Layer │ │ ┌─────────┐ ┌─────────┐ ┌─────────────┐ │ │ │ API │ │ Database│ │ Functions │ │ │ │ Calls │ │ Query │ │ Execution │ │ │ └─────────┘ └─────────┘ └─────────────┘ │ └─────────────────────────────────────────────┘ 2. Planner模块 规划器负责将用户目标分解为可执行的步骤。 ...

深入探讨AI Agent的架构设计模式，涵盖ReAct、CoT、多智能体协作等核心模式，帮助开发者构建更强大的AI应用系统。

深入探讨多Agent协作系统的设计架构、通信机制和协调策略，让多个AI智能体像人类团队一样高效协作完成复杂任务。

深入解析AI Agent的设计模式和实现方法，包括ReAct模式、工具使用、记忆系统、规划与推理，以及如何构建可靠的Agent应用。