如何通过from_function为StructuredTool实现异步方法以适配流式UI？

一、用from_function快速构建异步工具 #

完全可以通过from_function创建异步工具，不用手动继承BaseTool写完整类，这是最简实现方式。只需用async修饰工具函数，创建时指定is_async=True，LangChain会自动生成对应的异步执行方法。

基础异步工具示例 #

import asyncio
from langchain.tools import Tool

async def async_text_processor(input_text: str) -> str:
    # 模拟异步操作（比如调用异步API、数据库查询）
    await asyncio.sleep(1)
    return f"处理完成：{input_text}"

# 创建异步工具
text_tool = Tool.from_function(
    func=async_text_processor,
    name="AsyncTextProcessor",
    description="异步处理文本输入，返回处理后的结果",
    is_async=True  # 关键标记：声明这是异步工具
)

结构化输入的异步工具示例 #

如果需要结构化输入（比如带参数校验），用StructuredTool.from_function同样支持异步：

from langchain.tools import StructuredTool
from pydantic import BaseModel, Field

class QueryInput(BaseModel):
    user_query: str = Field(description="用户需要处理的查询内容")
    priority: str = Field(description="处理优先级，可选值：高/中/低")

async def async_structured_processor(input_data: QueryInput) -> str:
    await asyncio.sleep(1)
    return f"[{input_data.priority}] 处理查询：{input_data.user_query}"

structured_tool = StructuredTool.from_function(
    func=async_structured_processor,
    name="AsyncStructuredProcessor",
    description="异步处理带优先级的结构化查询",
    args_schema=QueryInput,
    is_async=True
)

二、解决"tool does not support async"错误 #

出现这个错误的核心原因是工具没有被标记为异步，或者没有提供异步执行方法。只需确保两点：

• 工具函数用async修饰
• 创建工具时设置is_async=True

这样LangChain会自动为工具生成arun方法，适配异步Agent的调用逻辑。

三、配合结构化聊天Agent实现流式输出 #

要实现类似OpenAIChat的token流式输出，需要用异步Agent执行器+流式模型，步骤如下：

1. 初始化流式异步模型 #

from langchain.chat_models import ChatOpenAI

llm = ChatOpenAI(
    temperature=0,
    model="gpt-3.5-turbo",
    streaming=True,  # 开启流式输出
    verbose=True
)

2. 创建异步结构化聊天Agent #

from langchain.agents import initialize_agent, AgentType
from langchain.agents import AsyncAgentExecutor

# 初始化基础agent
base_agent = initialize_agent(
    tools=[text_tool],  # 加入你的异步工具
    llm=llm,
    agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
    handle_parsing_errors=True  # 容错：处理工具调用解析错误
)

# 转为异步执行器（必须用这个才能支持异步流式）
async_agent = AsyncAgentExecutor.from_agent_and_tools(
    agent=base_agent.agent,
    tools=[text_tool],
    verbose=True
)

3. 流式处理响应并对接UI #

用astream方法获取流式chunk，逐个输出到UI：

# 定义流式响应生成函数
async def stream_response(query: str):
    async for chunk in async_agent.astream({"input": query}):
        # 根据chunk类型提取内容
        if "output" in chunk:
            # 最终输出的token
            yield chunk["output"]
        elif "intermediate_steps" in chunk:
            # 工具调用的中间步骤（可选展示）
            step = chunk["intermediate_steps"][0]
            yield f"→ 调用工具：{step[0].tool}\n工具返回：{step[1]}\n"

# 对接Panel流式UI（适配你原来的示例逻辑）
import panel as pn
pn.extension()

input_box = pn.widgets.TextInput(value="测试异步工具", placeholder="输入查询...")
output_box = pn.widgets.TextAreaInput(value="", rows=10, disabled=True)

async def handle_submit(event):
    output_box.value = ""
    query = input_box.value
    async for content in stream_response(query):
        output_box.value += content

input_box.on_submit(handle_submit)

pn.Column(input_box, output_box).servable()

关键注意事项 #

• 必须使用AsyncAgentExecutor，普通AgentExecutor不支持异步流式
• 所有加入Agent的工具都要标记为is_async=True，否则会触发异步不支持的错误
• 流式chunk的结构会根据Agent的执行阶段变化（思考、工具调用、最终输出），需按需提取内容

内容的提问来源于stack exchange，提问作者lalligagger