6.2 异步工具和上下文

异步上下文管理器

🎯 小白理解指南：什么是"上下文管理器"？

还记得 with open("file.txt") as f: 吗？这就是上下文管理器。

它帮你自动处理资源的打开和关闭：

• 进入 with 块时，自动打开资源
• 离开 with 块时，自动关闭资源（即使出错也会关闭）

异步上下文管理器是它的异步版本：

• 用 async with 代替 with
• 用 __aenter__ 代替 __enter__
• 用 __aexit__ 代替 __exit__

适用于数据库连接、HTTP 会话等需要异步打开/关闭的资源。

import asyncio

class AsyncDatabaseConnection:
    """异步数据库连接"""

    async def __aenter__(self):
        """进入上下文——连接数据库"""
        print("连接数据库...")
        await asyncio.sleep(0.1)  # 模拟异步连接
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """退出上下文——关闭数据库（自动执行，即使出错）"""
        print("关闭数据库...")
        await asyncio.sleep(0.1)

    async def query(self, sql: str) -> list:
        """执行查询"""
        await asyncio.sleep(0.1)
        return [{"result": "data"}]

async def main():
    # 🎯 async with 会自动调用 __aenter__ 和 __aexit__
    async with AsyncDatabaseConnection() as db:
        results = await db.query("SELECT * FROM agents")
        print(results)
    # 离开 with 块后，数据库连接自动关闭

asyncio.run(main())

异步 LangChain Tool

🎯 小白理解指南：为什么 Tool 需要异步版本？

LangChain 的 Tool（工具）经常需要调用外部 API：

• 搜索工具调用 Google API
• 天气工具调用天气 API
• 数据库工具查询数据库

这些操作都需要等待网络响应。如果用同步方式，Agent 调用 5 个工具就要等 5 次。

用异步工具的 arun() 方法，可以同时调用多个工具，大大加速！

LangChain 的命名规则：

• run() = 同步方法
• arun() = 异步方法（前面加个 a，代表 async）

from typing import Optional
import asyncio

class AsyncTool:
    """异步工具基类"""

    async def arun(self, query: str) -> str:
        """异步运行（子类必须实现）"""
        raise NotImplementedError

class AsyncSearchTool(AsyncTool):
    """异步搜索工具"""

    async def arun(self, query: str) -> str:
        await asyncio.sleep(0.5)  # 模拟 API 调用（实际会是网络请求）
        return f"搜索结果: {query}"

async def main():
    tool = AsyncSearchTool()
    result = await tool.arun("Python")
    print(result)

    # 🎯 异步工具的威力：同时调用多个工具
    tools = [AsyncSearchTool(), AsyncSearchTool(), AsyncSearchTool()]
    queries = ["Python", "LangChain", "异步编程"]

    # 同时发起 3 个搜索，只需要 0.5 秒而不是 1.5 秒
    results = await asyncio.gather(*[
        tool.arun(query) for tool, query in zip(tools, queries)
    ])
    print(results)

asyncio.run(main())

---

下一节：6.3 实战：异步 Agent