新建网站后域名跳转到别的网站了珠海品牌机械网站建设

新建网站后域名跳转到别的网站了,珠海品牌机械网站建设,网站页面设计教程,一个企业seo网站的优化流程LangChain 工具API#xff1a;超越简单函数调用的智能体工程实践 引言#xff1a;工具调用作为大模型能力的延伸 在大语言模型(LLM)应用开发领域#xff0c;LangChain已成为事实标准框架之一。其核心价值之一在于通过工具API#xff08;Tools API#xff09;使大模型能够…LangChain 工具API超越简单函数调用的智能体工程实践引言工具调用作为大模型能力的延伸在大语言模型(LLM)应用开发领域LangChain已成为事实标准框架之一。其核心价值之一在于通过工具APITools API使大模型能够突破纯文本生成的限制与外部系统、数据和功能进行交互。虽然基础的天气查询、计算器示例广为人知但在实际企业级应用中工具API的深度应用远不止于此。本文将深入探讨LangChain工具API的高级用法包括动态工具选择、复杂工具编排、错误处理机制和性能优化策略并结合新颖的区块链数据分析和多模态处理案例展示如何构建真正实用的智能体系统。一、LangChain工具API的核心架构1.1 工具系统设计哲学LangChain的工具系统基于一个核心理念将大语言模型作为推理引擎而非仅仅是文本生成器。工具API的设计遵循了以下原则标准化接口所有工具统一通过Tool基类或tool装饰器定义声明式描述每个工具必须提供清晰的自然语言描述供模型理解其功能类型安全通过Pydantic模型确保输入输出的结构化可组合性工具可以相互组合形成复杂的工作流1.2 工具注册与发现机制from langchain.tools import BaseTool, StructuredTool, tool from pydantic import BaseModel, Field from typing import Optional, Type, Dict, Any import asyncio class DataAnalysisInput(BaseModel): 区块链数据分析输入参数 contract_address: str Field(description智能合约地址) start_block: int Field(description起始区块号) end_block: Optional[int] Field(defaultNone, description结束区块号) analysis_type: str Field( description分析类型: transactions, token_flows, anomaly_detection ) class BlockchainAnalysisTool(BaseTool): name: str blockchain_data_analyzer description: str 高级区块链数据分析工具支持 1. 智能合约交易模式分析 2. ERC-20代币流向追踪 3. 异常交易检测和模式识别 需要提供合约地址和区块范围 args_schema: Type[BaseModel] DataAnalysisInput return_direct: bool False def _run(self, **kwargs) - str: 同步执行工具逻辑 return self.analyze_blockchain_data(**kwargs) async def _arun(self, **kwargs) - str: 异步执行工具逻辑 return await self.analyze_blockchain_data_async(**kwargs) def analyze_blockchain_data(self, **kwargs) - str: # 实现具体的区块链数据分析逻辑 # 这里简化为模拟实现 from datetime import datetime import random analysis_type kwargs.get(analysis_type, transactions) results { analysis_type: analysis_type, contract_address: kwargs[contract_address], time_executed: datetime.now().isoformat(), total_transactions: random.randint(100, 5000), unique_addresses: random.randint(50, 500), gas_usage_pattern: f{random.uniform(0.1, 5.0):.2f} ETH/day } if analysis_type token_flows: results[token_movements] random.randint(10, 200) results[largest_transfer] f{random.uniform(100, 10000):.2f} tokens return str(results) async def analyze_blockchain_data_async(self, **kwargs) - str: 异步版本的数据分析 await asyncio.sleep(0.5) # 模拟异步操作 return self.analyze_blockchain_data(**kwargs)二、高级工具调用模式2.1 动态工具路由与条件执行在实际应用中智能体需要根据上下文动态选择工具。LangChain通过ToolRouter和ConditionalToolSelector实现这一功能from langchain.agents import Tool, AgentExecutor, create_react_agent from langchain.agents.tools import ToolRouter from langchain_core.prompts import PromptTemplate from langchain_openai import ChatOpenAI import json class DynamicToolRouter: 动态工具路由器基于上下文智能选择工具 def __init__(self, tools: Dict[str, Tool]): self.tools tools self.llm ChatOpenAI(temperature0, modelgpt-4) self.routing_prompt PromptTemplate.from_template( 根据用户查询和上下文选择最合适的工具。 可用工具 {tools_list} 用户查询{query} 当前上下文{context} 请分析 1. 用户的核心需求是什么 2. 需要哪些数据或操作 3. 哪个工具最合适 返回JSON格式 {{ selected_tool: 工具名称, reasoning: 选择理由, parameters: {{param1: value1}} # 根据工具需要的参数填写 }} ) async def route(self, query: str, context: Dict[str, Any]) - Dict[str, Any]: 动态路由到最合适的工具 tools_list \n.join([ f- {name}: {tool.description} for name, tool in self.tools.items() ]) prompt self.routing_prompt.format( tools_listtools_list, queryquery, contextjson.dumps(context, indent2) ) response await self.llm.ainvoke(prompt) try: decision json.loads(response.content) return decision except json.JSONDecodeError: # 如果模型返回的不是有效JSON使用启发式方法选择工具 return self.fallback_route(query, context) def fallback_route(self, query: str, context: Dict[str, Any]) - Dict[str, Any]: 后备路由逻辑 # 基于关键词的简单路由 query_lower query.lower() if any(word in query_lower for word in [blockchain, contract, token]): return { selected_tool: blockchain_data_analyzer, reasoning: 查询涉及区块链相关术语, parameters: {analysis_type: transactions} } elif any(word in query_lower for word in [image, photo, picture]): return { selected_tool: multimodal_processor, reasoning: 查询涉及图像处理, parameters: {} } else: return { selected_tool: general_web_search, reasoning: 通用查询使用网络搜索, parameters: {} }2.2 工具链与工作流编排复杂任务通常需要多个工具协同工作。以下是工具链编排的高级模式from langchain.schema import AgentAction, AgentFinish from langchain.agents import Agent from typing import List, Tuple, Any, Optional, Dict from langchain_core.callbacks import CallbackManagerForChainRun import networkx as nx import matplotlib.pyplot as plt class WorkflowOrchestrator: 工作流编排器支持复杂的工具执行顺序 def __init__(self, tools: List[Tool], max_iterations: int 10): self.tools {tool.name: tool for tool in tools} self.max_iterations max_iterations self.execution_graph nx.DiGraph() async def execute_workflow( self, initial_input: str, workflow_definition: Dict[str, Any] ) - Dict[str, Any]: 执行定义好的工作流 workflow_definition 示例 { steps: [ { tool: blockchain_data_analyzer, params: {analysis_type: transactions}, depends_on: [] }, { tool: data_visualizer, params: {}, depends_on: [blockchain_data_analyzer] } ] } results {} execution_order self._calculate_execution_order(workflow_definition) for step_name in execution_order: step next(s for s in workflow_definition[steps] if s.get(name, s[tool]) step_name) # 收集依赖项的输出作为输入 step_input self._prepare_step_input(step, results, initial_input) # 执行工具 tool self.tools[step[tool]] try: if hasattr(tool, _arun): step_result await tool._arun(**step_input) else: step_result tool._run(**step_input) results[step_name] { output: step_result, status: success, tool: step[tool] } # 记录到执行图 self.execution_graph.add_node(step_name, resultstep_result) for dep in step.get(depends_on, []): self.execution_graph.add_edge(dep, step_name) except Exception as e: results[step_name] { output: str(e), status: failed, tool: step[tool] } # 根据工作流定义决定是否继续 if step.get(fail_fast, True): break return { results: results, success_rate: sum(1 for r in results.values() if r[status] success) / len(results), execution_graph: self.execution_graph } def _calculate_execution_order(self, workflow_def: Dict[str, Any]) - List[str]: 计算工具执行顺序考虑依赖关系 # 使用拓扑排序确定执行顺序 dep_graph nx.DiGraph() for step in workflow_def[steps]: step_name step.get(name, step[tool]) dep_graph.add_node(step_name) for dep in step.get(depends_on, []): dep_graph.add_edge(dep, step_name) try: return list(nx.topological_sort(dep_graph)) except nx.NetworkXUnfeasible: raise ValueError(工作流定义包含循环依赖) def _prepare_step_input(self, step: Dict[str, Any], results: Dict[str, Any], initial_input: str) - Dict[str, Any]: 准备步骤输入合并依赖项的输出 base_params step.get(params, {}).copy() # 将依赖项的输出合并到参数中 for dep in step.get(depends_on, []): if dep in results and results[dep][status] success: # 可以在这里添加复杂的输出转换逻辑 base_params[finput_from_{dep}] results[dep][output] # 如果这是第一步且没有特殊输入使用初始输入 if not step.get(depends_on) and input not in base_params: base_params[input] initial_input return base_params def visualize_workflow(self, filename: str workflow.png): 可视化工作流执行图 plt.figure(figsize(12, 8)) pos nx.spring_layout(self.execution_graph) # 根据执行状态着色 node_colors [] for node in self.execution_graph.nodes(): if result in self.execution_graph.nodes[node]: node_colors.append(lightgreen) else: node_colors.append(lightcoral) nx.draw(self.execution_graph, pos, with_labelsTrue, node_colornode_colors, node_size3000, font_size10, font_weightbold, arrowsize20) plt.title(Workflow Execution Graph) plt.savefig(filename, dpi300, bbox_inchestight) plt.close()三、新颖案例多模态智能合约审计系统让我们构建一个结合多种工具的实际系统多模态智能合约审计助手。from langchain.tools import BaseTool from pydantic import BaseModel, Field from typing import Optional, List import aiohttp import base64 from PIL import Image import io class SmartContractAuditInput(BaseModel): 智能合约审计输入参数 contract_code: Optional[str] Field( None, descriptionSolidity合约源代码如果提供则直接分析 ) contract_address: Optional[str] Field( None, description已部署合约地址如果提供则从区块链获取代码 ) audit_focus: List[str] Field( default[security, gas_optimization, logic], description审计重点领域 ) include_visual: bool Field( defaultTrue, description是否生成可视化分析报告 ) class MultimodalContractAuditor(BaseTool): 多模态智能合约审计工具 结合代码分析、区块链数据查询和可视化生成 name: str multimodal_contract_auditor description: str 高级智能合约审计系统提供 1. Solidity代码安全分析 2. 已部署合约的链上行为分析 3. 交互式可视化报告生成 4. 多维度风险评估 args_schema: type[BaseModel] SmartContractAuditInput def _run(self, **kwargs) - dict: # 综合多种分析工具的结果 audit_results { security_analysis: self._analyze_security(**kwargs), gas_analysis: self._analyze_gas_usage(**kwargs), code_quality: self._analyze_code_quality(**kwargs), visual_report: self._generate_visual_report(**kwargs) if kwargs.get(include_visual) else None } # 计算综合风险评分 audit_results[risk_score] self._calculate_risk_score(audit_results) return audit_results async def _arun(self, **kwargs) - dict: 异步执行所有分析 import asyncio # 并行执行多个分析任务 tasks [ self._analyze_security_async(**kwargs), self._analyze_gas_usage_async(**kwargs), self._analyze_code_quality_async(**kwargs) ] if kwargs.get(include_visual): tasks.append(self._generate_visual_report_async(**kwargs)) results await asyncio.gather(*tasks, return_exceptionsTrue) audit_results { security_analysis: results[0] if not isinstance(results[0], Exception) else str(results[0]), gas_analysis: results[1] if not isinstance(results[1], Exception) else str(results[1]), code_quality: results[2] if not isinstance(results[2], Exception) else str(results[2]), visual_report: results[3] if len(results) 3 and not isinstance(results[3], Exception) else None } audit_results[risk_score] self._calculate_risk_score(audit_results) return audit_results def _analyze_security(self, **kwargs) - dict: 安全分析实现简化版 # 这里可以集成Slither、Mythril等安全分析工具 contract_code kwargs.get(contract_code, ) vulnerabilities [] # 简单的模式匹配实际应用中应使用专业工具 dangerous_patterns [ (reentrancy, [call.value, send(, transfer(]), (integer_overflow, [uint, , *]), (unchecked_external_call, [call(, delegatecall(]) ]