LangChain工具系统：大模型与外部世界的桥梁

1. LangChain工具系统概述

LangChain 1.0的工具系统是大语言模型(LLM)开发中一个革命性的功能模块，它为AI智能体提供了与外部世界交互的标准接口。这个系统本质上是一套精心设计的抽象层，让开发者能够将各种外部功能无缝集成到LLM应用中。

1.1 工具系统的核心价值

工具系统解决了LLM开发中的几个关键痛点：

• 能力边界突破：大语言模型本身是封闭的知识系统，工具调用让它能够获取实时信息（如天气、股价）、执行复杂计算、操作系统资源
• 工程化落地：通过标准化接口，将业务API、数据库、文件系统等企业资源安全地暴露给LLM使用
• 模块化开发：工具可以像乐高积木一样自由组合，构建出适应不同场景的智能体

在实际项目中，我们经常遇到这样的场景：客户需要一个能查询内部CRM系统的聊天机器人。没有工具系统时，我们不得不通过复杂的提示词工程来解析用户意图，再用传统代码处理请求。现在通过工具系统，我们可以将CRM查询封装成一个标准工具，让LLM自主决定何时以及如何使用它。

1.2 系统架构解析

LangChain工具系统的架构设计体现了良好的软件工程原则：

code复制┌───────────────────────────────────────┐
│            Application Layer          │
│  (Agents, Chains, LangGraph Workflow) │
└──────────────────────┬────────────────┘
                       │
┌──────────────────────▼────────────────┐
│             Integration Layer         │
│  (Toolkits: GitHub, Slack, Gmail etc) │
└──────────────────────┬────────────────┘
                       │
┌──────────────────────▼────────────────┐
│             Implementation Layer      │
│  (Custom Tools via @tool/BaseTool)    │
└──────────────────────┬────────────────┘
                       │
┌──────────────────────▼────────────────┐
│              Abstraction Layer        │
│         (BaseTool, Tool Interface)    │
└───────────────────────────────────────┘

这种分层设计带来的好处是：

1. 可扩展性：可以在不修改上层应用的情况下添加新工具
2. 复用性：通用工具可以被不同智能体共享使用
3. 安全性：通过抽象层可以统一实施权限控制和输入验证

1.3 工具调用工作机制

当用户向集成了工具系统的LLM应用提问时，完整的处理流程如下：

1. 意图识别：模型首先分析用户问题，判断是否需要调用工具
2. 工具选择：如果需要工具，模型会根据工具描述选择最合适的工具
3. 参数提取：模型从用户输入中提取工具所需的参数
4. 执行调用：系统执行工具并获取结果
5. 结果整合：模型将工具返回的结果整合到最终回复中

这个过程中最精妙的部分是"工具选择"环节。LangChain采用了一种自描述机制——每个工具都必须提供清晰的name和description，这些描述会被自动注入到模型的系统提示词中，帮助模型理解何时使用这个工具。

实践建议：工具描述应该采用"动词+宾语"的格式，如"查询指定城市的实时天气数据"，避免使用模糊的描述。

2. 内置工具深度解析

LangChain 1.0提供了超过100种开箱即用的内置工具，覆盖了大多数常见场景。这些工具经过精心设计和优化，可以直接投入生产环境使用。

2.1 核心工具分类与应用

2.2 内置工具实战技巧

以最常用的PythonREPLTool为例，在实际使用中有几个关键注意事项：

python复制from langchain.tools import PythonREPLTool

# 正确初始化方式
python_repl = PythonREPLTool(
    timeout=30,  # 设置超时防止无限执行
    sanitize_input=True,  # 启用输入净化
    restricted_imports=["os", "subprocess"]  # 限制危险模块
)

# 危险示例（绝对避免）
dangerous_code = """
import os
os.system('rm -rf /')  # 这将导致灾难性后果
"""

# 安全的使用方式
safe_code = """
import numpy as np
arr = np.array([1,2,3])
print(arr.mean())
"""

result = python_repl.run(safe_code)

安全警示：任何代码执行工具都必须配置沙箱环境。在生产环境中，建议使用Docker容器隔离执行，并设置严格的资源限制(CPU、内存、网络)。

2.3 工具组合应用案例

内置工具的真正威力在于组合使用。下面是一个股票分析工作流的示例：

python复制from langchain.tools import GoogleSearchTool, PythonREPLTool
from langchain.agents import initialize_agent

# 工具初始化
search = GoogleSearchTool(api_key="your_api_key")
calculator = CalculatorTool()
python_repl = PythonREPLTool()

tools = [search, calculator, python_repl]

# 创建智能体
agent = initialize_agent(
    tools=tools,
    llm=ChatOpenAI(temperature=0),
    agent_type="structured-chat"
)

# 执行复杂查询
query = """获取苹果公司(AAPL)最新股价，
计算如果我现在投资1万美元能买多少股，
然后用Python绘制最近一年的股价趋势图"""

result = agent.run(query)

这个例子展示了如何通过工具组合实现复杂的工作流：搜索获取实时数据 → 计算器进行投资计算 → Python工具进行数据可视化。

3. 自定义工具开发指南

虽然内置工具很强大，但实际业务中我们经常需要开发自定义工具来对接内部系统。LangChain提供了三种主要的工具开发方式，各有适用场景。

3.1 @tool装饰器：快速原型开发

@tool装饰器是最简单的工具创建方式，适合快速验证想法：

python复制from langchain_core.tools import tool
import requests

@tool
def query_employee_info(employee_id: str) -> dict:
    """查询公司内部员工详细信息，包括部门、职位和联系方式
    
    Args:
        employee_id: 员工工号，如'EMP20230001'
    
    Returns:
        包含员工详细信息的字典
    
    Raises:
        ValueError: 当员工不存在时抛出
    """
    # 这里是模拟实现，实际应该调用HR系统API
    mock_data = {
        "EMP20230001": {
            "name": "张三",
            "department": "研发部",
            "position": "高级工程师",
            "email": "zhangsan@company.com"
        }
    }
    
    if employee_id not in mock_data:
        raise ValueError(f"员工{employee_id}不存在")
    
    return mock_data[employee_id]

# 使用示例
try:
    info = query_employee_info.invoke({"employee_id": "EMP20230001"})
    print(info)
except ValueError as e:
    print(f"查询失败: {str(e)}")

关键设计要点：

1. 类型注解：参数和返回值都要有明确的类型提示
2. 文档字符串：详细描述工具功能、参数和可能的异常
3. 错误处理：对非法输入要有清晰的错误反馈
4. 返回结构：尽量返回结构化数据方便模型解析

3.2 BaseTool类：企业级工具开发

对于需要更复杂控制的工具，应该继承BaseTool类：

python复制from langchain_core.tools import BaseTool
from typing import Optional, Type
from pydantic import BaseModel, Field

class InventoryQueryInput(BaseModel):
    product_id: str = Field(..., description="产品唯一标识符")
    warehouse_id: Optional[str] = Field(None, description="仓库ID，不指定则查询所有仓库")

class InventoryTool(BaseTool):
    name = "inventory_query"
    description = "查询产品库存信息，包括可用数量和所在仓库"
    args_schema: Type[BaseModel] = InventoryQueryInput
    
    def _run(self, product_id: str, warehouse_id: Optional[str] = None, **kwargs):
        """实际库存查询逻辑"""
        # 这里应该是调用WMS系统的API
        mock_data = {
            "P1001": {
                "total": 150,
                "details": [
                    {"warehouse": "WH01", "quantity": 80},
                    {"warehouse": "WH02", "quantity": 70}
                ]
            }
        }
        
        if product_id not in mock_data:
            raise ValueError(f"产品{product_id}不存在")
            
        result = mock_data[product_id]
        
        if warehouse_id:
            for item in result["details"]:
                if item["warehouse"] == warehouse_id:
                    return {**item, "product_id": product_id}
            raise ValueError(f"产品{product_id}在仓库{warehouse_id}无库存")
            
        return result

# 使用示例
inventory_tool = InventoryTool()
print(inventory_tool.run({"product_id": "P1001"}))
print(inventory_tool.run({"product_id": "P1001", "warehouse_id": "WH01"}))

BaseTool的优势在于：

• 输入验证：通过Pydantic模型定义严格的输入模式
• 细粒度控制：可以重写工具的生命周期方法
• 元数据丰富：支持更详细的工具描述和配置

3.3 ToolNode：工作流集成

当需要将工具集成到LangGraph工作流中时，ToolNode提供了完美的解决方案：

python复制from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage

# 准备工具集
tools = [query_employee_info, inventory_tool]

# 创建工具节点
tool_node = ToolNode(tools)

# 模拟工作流中的消息
messages = [HumanMessage(content="查询员工EMP20230001的信息")]

# 执行工具调用
result = tool_node.invoke({"messages": messages})
print(result)

ToolNode的核心价值在于：

1. 无缝集成：可以轻松插入到任何LangGraph工作流中
2. 自动路由：根据消息内容自动选择正确的工具
3. 状态管理：维护工具调用的上下文信息

4. 高级工具开发技巧

4.1 异步工具实现

对于IO密集型的工具，异步实现可以显著提高性能：

python复制from langchain_core.tools import BaseTool
import aiohttp

class AsyncWeatherTool(BaseTool):
    name = "async_weather"
    description = "异步查询城市天气信息"
    
    async def _arun(self, city: str, **kwargs):
        async with aiohttp.ClientSession() as session:
            url = f"https://api.weather.com/v1/city/{city}"
            async with session.get(url) as response:
                if response.status != 200:
                    raise ValueError(f"天气查询失败: {response.status}")
                return await response.json()

# 使用示例
import asyncio

async def main():
    tool = AsyncWeatherTool()
    print(await tool.arun({"city": "北京"}))

asyncio.run(main())

异步工具开发要点：

• 继承BaseTool并实现_arun方法
• 使用async/await语法处理异步操作
• 注意资源清理（如关闭会话）
• 在FastAPI等异步框架中特别有用

4.2 工具版本管理

在企业环境中，工具版本管理至关重要：

python复制from langchain_core.tools import BaseTool
from pydantic import Field

class VersionedTool(BaseTool):
    name = "versioned_example"
    description = "示例工具演示版本管理"
    version: str = Field("1.0.0", description="工具版本号")
    changelog: dict = Field({
        "1.0.0": "初始版本",
        "1.1.0": "添加了缓存支持"
    })
    
    def _run(self, *args, **kwargs):
        return {
            "result": "示例输出",
            "metadata": {
                "tool_version": self.version
            }
        }

# 使用示例
tool_v1 = VersionedTool(version="1.0.0")
tool_v2 = VersionedTool(version="1.1.0")

print(tool_v1.run({}))
print(tool_v2.run({}))

版本管理最佳实践：

1. 遵循语义化版本规范(MAJOR.MINOR.PATCH)
2. 维护详细的变更日志
3. 在返回结果中包含版本信息
4. 考虑提供向后兼容性

4.3 工具性能监控

生产环境中的工具应该内置监控能力：

python复制from langchain_core.tools import BaseTool
import time
from statsd import StatsClient

class MonitoredTool(BaseTool):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.statsd = StatsClient(host='localhost', port=8125)
    
    def _run(self, input_text: str, **kwargs):
        start_time = time.time()
        
        try:
            # 模拟工具处理
            result = input_text.upper()
            elapsed = (time.time() - start_time) * 1000
            
            # 记录指标
            self.statsd.timing(f"tool.{self.name}.latency", elapsed)
            self.statsd.incr(f"tool.{self.name}.success")
            
            return result
        except Exception as e:
            self.statsd.incr(f"tool.{self.name}.error")
            raise

# 使用示例
tool = MonitoredTool(name="monitored_example")
print(tool.run({"input_text": "hello"}))

关键监控指标包括：

• 调用次数（成功/失败）
• 延迟分布
• 资源使用情况（CPU、内存）
• 依赖服务可用性

5. 工具包集成策略

5.1 官方工具包解析

LangChain提供了多个官方维护的工具包，简化了常见平台的集成：

5.2 企业微信集成案例

以下是如何创建自定义的企业微信工具包：

python复制from langchain_core.tools import BaseTool
from langchain_core.toolkits import BaseToolkit
from typing import List
import requests

class WeComMessageTool(BaseTool):
    name = "wecom_message"
    description = "发送企业微信消息"
    
    def __init__(self, corp_id: str, secret: str, agent_id: str):
        self.corp_id = corp_id
        self.secret = secret
        self.agent_id = agent_id
        self.access_token = None
    
    def _get_token(self):
        url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={self.corp_id}&corpsecret={self.secret}"
        response = requests.get(url).json()
        self.access_token = response["access_token"]
    
    def _run(self, user_id: str, content: str, **kwargs):
        if not self.access_token:
            self._get_token()
            
        url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={self.access_token}"
        payload = {
            "touser": user_id,
            "msgtype": "text",
            "agentid": self.agent_id,
            "text": {"content": content},
            "safe": 0
        }
        
        response = requests.post(url, json=payload).json()
        if response["errcode"] != 0:
            raise ValueError(f"发送失败: {response['errmsg']}")
        
        return {"status": "success", "message_id": response["msgid"]}

class WeComToolkit(BaseToolkit):
    def __init__(self, corp_id: str, secret: str, agent_id: str):
        self.message_tool = WeComMessageTool(corp_id, secret, agent_id)
    
    def get_tools(self) -> List[BaseTool]:
        return [self.message_tool]

# 使用示例
toolkit = WeComToolkit(
    corp_id="your_corp_id",
    secret="your_secret",
    agent_id="your_agent_id"
)

tools = toolkit.get_tools()
wecom_tool = tools[0]
print(wecom_tool.run({"user_id": "User1", "content": "测试消息"}))

企业集成关键点：

1. 认证管理：正确处理token获取和刷新
2. 错误处理：处理各种API错误情况
3. 速率限制：遵守平台API调用限制
4. 安全考虑：妥善保管敏感凭证

5.3 工具包设计模式

优秀的工具包应该遵循以下设计原则：

1. 单一职责：每个工具包专注于一个平台或业务领域
2. 统一认证：在工具包级别集中管理认证逻辑
3. 模块化设计：允许用户按需选择工具，而不是全量引入
4. 配置灵活：通过配置文件或环境变量管理各种参数
5. 文档完善：提供清晰的接入文档和使用示例

6. 生产环境最佳实践

6.1 安全防护策略

工具调用是LLM系统中最主要的安全风险点，必须实施多层防护：

防御层级架构：

code复制┌───────────────────────┐
│  输入验证与净化层      │  # 检查输入格式、内容
├───────────────────────┤
│  权限控制系统层        │  # RBAC权限管理
├───────────────────────┤
│  沙箱执行环境层        │  # 容器化/虚拟机隔离
├───────────────────────┤
│  输出过滤与脱敏层      │  # 移除敏感信息
└───────────────────────┘

具体实施示例：

python复制from langchain_core.tools import BaseTool
import re
from typing import List

class SafeSQLTool(BaseTool):
    name = "safe_sql_query"
    description = "执行安全的SQL查询"
    allowed_tables: List[str] = ["products", "customers"]
    
    def _validate_query(self, query: str):
        # 检查是否包含危险操作
        if re.search(r"\b(DROP|DELETE|UPDATE|INSERT|ALTER)\b", query, re.I):
            raise ValueError("查询包含危险操作")
            
        # 检查是否访问了未授权的表
        for table in re.findall(r"\bFROM\s+(\w+)", query, re.I):
            if table.lower() not in self.allowed_tables:
                raise ValueError(f"无权访问表{table}")
    
    def _run(self, query: str, **kwargs):
        self._validate_query(query)
        
        # 实际执行查询
        # ... 这里应该是使用参数化查询 ...
        
        return {"status": "success", "data": "..."}

# 使用示例
tool = SafeSQLTool()
try:
    print(tool.run({"query": "SELECT * FROM products LIMIT 5"}))
    print(tool.run({"query": "DROP TABLE products"}))  # 会抛出异常
except ValueError as e:
    print(f"安全错误: {str(e)}")

6.2 性能优化技巧

工具调用的性能直接影响用户体验，以下是一些实测有效的优化方法：

1. 缓存策略：

python复制from functools import lru_cache
from datetime import timedelta
from langchain_core.tools import BaseTool

class CachedWeatherTool(BaseTool):
    @lru_cache(maxsize=1000)
    def _run(self, city: str, **kwargs):
        # 实际天气查询逻辑
        return {"city": city, "temperature": "25°C"}

2. 批处理模式：

python复制class BatchProcessingTool(BaseTool):
    def _run(self, inputs: List[str], **kwargs):
        # 单条处理函数
        def process_single(item):
            return item.upper()
        
        return [process_single(item) for item in inputs]

3. 异步并行：

python复制import asyncio
from langchain_core.tools import BaseTool

class AsyncBatchTool(BaseTool):
    async def _arun_batch(self, inputs: List[str], **kwargs):
        async def process(item):
            await asyncio.sleep(0.1)  # 模拟IO操作
            return item.upper()
            
        return await asyncio.gather(*[process(item) for item in inputs])

4. 连接池管理：

python复制import psycopg2
from psycopg2 import pool
from langchain_core.tools import BaseTool

class DBTool(BaseTool):
    def __init__(self):
        self.connection_pool = psycopg2.pool.SimpleConnectionPool(
            minconn=1,
            maxconn=10,
            host="localhost",
            database="mydb",
            user="user",
            password="password"
        )
    
    def _run(self, query: str, **kwargs):
        conn = self.connection_pool.getconn()
        try:
            with conn.cursor() as cursor:
                cursor.execute(query)
                return cursor.fetchall()
        finally:
            self.connection_pool.putconn(conn)

6.3 监控与日志

完善的监控是生产系统不可或缺的部分：

python复制from langchain_core.tools import BaseTool
import logging
from datetime import datetime

class MonitoredTool(BaseTool):
    def __init__(self):
        self.logger = logging.getLogger(self.name)
        self.request_count = 0
    
    def _run(self, input_text: str, **kwargs):
        start_time = datetime.now()
        self.request_count += 1
        
        try:
            # 工具逻辑
            result = input_text.upper()
            
            # 记录成功日志
            duration = (datetime.now() - start_time).total_seconds()
            self.logger.info(
                f"Tool {self.name} succeeded - "
                f"Duration: {duration:.2f}s - "
                f"Total requests: {self.request_count}"
            )
            
            return result
        except Exception as e:
            # 记录错误日志
            self.logger.error(
                f"Tool {self.name} failed - "
                f"Error: {str(e)} - "
                f"Input: {input_text}"
            )
            raise

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# 使用示例
tool = MonitoredTool(name="monitored_example")
print(tool.run({"input_text": "test"}))

日志记录最佳实践：

1. 包含足够的上下文信息（时间戳、工具名、输入参数等）
2. 区分不同日志级别（INFO、WARNING、ERROR）
3. 记录性能指标（处理时间、吞吐量）
4. 避免记录敏感信息（密码、令牌等）
5. 使用结构化日志便于分析

7. 典型问题排查指南

7.1 工具调用失败分析

以下是工具调用中常见问题及解决方法：

7.2 调试技巧与工具

LangChain提供了多种调试工具帮助诊断问题：

1. 回调处理器：

python复制from langchain_core.callbacks import StdOutCallbackHandler

handler = StdOutCallbackHandler()
tool.run({"input": "test"}, callbacks=[handler])

2. verbose模式：

python复制agent = initialize_agent(
    tools=tools,
    llm=ChatOpenAI(),
    agent_type="structured-chat",
    verbose=True  # 启用详细日志
)

3. 中间结果检查：

python复制from langchain_core.runnables import RunnableLambda

debug_chain = (
    RunnableLambda(lambda x: print(f"输入: {x}") | x)
    | tool
    | RunnableLambda(lambda x: print(f"输出: {x}") | x)
)

debug_chain.invoke({"input": "test"})

4. LangSmith集成：

python复制import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "My Project"

# 所有调用将自动记录到LangSmith
tool.run({"input": "test"})

7.3 常见错误处理

工具开发中需要特别注意这些边界情况：

1. 网络异常处理：

python复制from requests.exceptions import RequestException
import backoff

@backoff.on_exception(
    backoff.expo,
    RequestException,
    max_tries=3
)
def call_external_api(url):
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

2. 资源清理：

python复制class ResourceIntensiveTool(BaseTool):
    def __init__(self):
        self.resource = initialize_resource()
    
    def __del__(self):
        if hasattr(self, 'resource'):
            cleanup_resource(self.resource)
    
    def _run(self, input_text: str, **kwargs):
        try:
            return process_with_resource(self.resource, input_text)
        except Exception as e:
            cleanup_resource(self.resource)
            self.resource = None
            raise

3. 超时控制：

python复制import signal
from contextlib import contextmanager

class TimeoutException(Exception):
    pass

@contextmanager
def time_limit(seconds):
    def signal_handler(signum, frame):
        raise TimeoutException("操作超时")
    
    signal.signal(signal.SIGALRM, signal_handler)
    signal.alarm(seconds)
    try:
        yield
    finally:
        signal.alarm(0)

class SafeTool(BaseTool):
    def _run(self, input_text: str, **kwargs):
        try:
            with time_limit(5):  # 5秒超时
                return long_running_process(input_text)
        except TimeoutException:
            return {"error": "处理超时"}

8. 架构设计与扩展思路

8.1 大型系统中的工具管理

当系统中有数十个工具时，需要考虑更高级的管理策略：

1. 工具注册中心模式：

python复制from typing import Dict, Type
from langchain_core.tools import BaseTool

class ToolRegistry:
    _instance = None
    _tools: Dict[str, Type[BaseTool]] = {}
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
    
    @classmethod
    def register(cls, name: str):
        def decorator(tool_class: Type[BaseTool]):
            cls._tools[name] = tool_class
            return tool_class
        return decorator
    
    @classmethod
    def get_tool(cls, name: str, **kwargs) -> BaseTool:
        if name not in cls._tools:
            raise ValueError(f"工具{name}未注册")
        return cls._tools[name](**kwargs)

# 注册工具
@ToolRegistry.register("weather")
class WeatherTool(BaseTool):
    pass

# 使用工具
weather_tool = ToolRegistry.get_tool("weather")

2. 按领域分组管理：

python复制from dataclasses import dataclass
from typing import Dict, List

@dataclass
class ToolGroup:
    name: str
    description: str
    tools: List[BaseTool]

class ToolManager:
    def __init__(self):
        self.groups: Dict[str, ToolGroup] = {}
    
    def add_group(self, name: str, description: str):
        self.groups[name] = ToolGroup(name, description, [])
    
    def add_tool(self, group_name: str, tool: BaseTool):
        if group_name not in self.groups:
            raise ValueError(f"分组{group_name}不存在")
        self.groups[group_name].tools.append(tool)
    
    def get_tools_by_group(self, group_name: str) -> List[BaseTool]:
        return self.groups[group_name].tools

# 使用示例
manager = ToolManager()
manager.add_group("search", "搜索相关工具")
manager.add_tool("search", GoogleSearchTool())
manager.add_tool("search", BingSearchTool())

8.2 工具路由优化

对于大型工具集，智能路由可以显著提高效率：

python复制from langchain_core.tools import BaseTool
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

class SmartRouter:
    def __init__(self, tools: List[BaseTool]):
        self.tools = tools
        self.vectorizer = TfidfVectorizer()
        
        # 使用工具描述构建特征矩阵
        descriptions = [f"{t.name}: {t.description}" for t in tools]
        self.matrix = self.vectorizer.fit_transform(descriptions)
    
    def find_best_tool(self, query: str) -> BaseTool:
        # 将查询向量化
        query_vec = self.vectorizer.transform([query])
        
        # 计算相似度
        similarities = cosine_similarity(query_vec, self.matrix)
        best_idx = similarities.argmax()
        
        return self.tools[best_idx]

# 使用示例
tools = [CalculatorTool(), GoogleSearchTool()]
router = SmartRouter(tools)
best_tool = router.find_best_tool("今天的天气怎么样")
print(f"最佳工具: {best_tool.name}")

8.3 工具版本兼容性

处理多版本工具共存的策略：

python复制from langchain_core.tools import BaseTool
from pydantic import Field

class VersionedTool(BaseTool):
    version: str = Field("1.0.0", description="工具版本")
    compatible_versions: list = Field(["1.0.0", "1.1.0"], description="兼容版本")
    
    def is_compatible(self, other_version: str) -> bool:
        return other_version in self.compatible_versions

class ToolAdapter:
    @staticmethod
    def adapt_input(input_data: dict, from_version: str, to_version: str) -> dict:
        # 实现不同版本间的输入转换
        if from_version == "1.0.0" and to_version == "1.1.0":
            if "text" in input_data:
                input_data["content"] = input_data.pop("text")
        return input_data
    
    @staticmethod
    def adapt_output(output_data: dict, from_version: str, to_version: str) -> dict:
        # 实现不同版本间的输出转换
        if from_version == "1.1.0" and to_version == "1.0.0":
            if "result" in output_data:
                output_data["output"] = output_data.pop("result")
        return output_data

8.4 工具市场构想

对于大型组织，可以构建内部工具市场：

python复制from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class ToolMetadata(BaseModel):
    name: str
    description: str
    version: str
    author: str
    endpoint: str

class ToolMarketplace:
    def __init__(self):
        self.tools: Dict[str, ToolMetadata] = {}
    
    def register_tool(self, metadata: ToolMetadata):
        self.tools[metadata.name] = metadata
    
    def discover_tools(self, query: str = None) -> List[ToolMetadata]:
        if query:
            return [t for t in self.tools.values() if query.lower() in t.description.lower()]
        return list(self.tools.values())

marketplace = ToolMarketplace()

@app.post("/register")
async def register(metadata: ToolMetadata):
    marketplace.register_tool(metadata)
    return {"status": "success"}

@app.get("/discover")
async def discover(query: str = None):
    return marketplace.discover_tools(query)

这种架构允许：

• 开发者发布新工具
• 用户发现和评估工具
• 管理员管理工具生命周期
• 系统自动处理依赖和兼容性