autogen_ext.tools.langchain

class LangChainToolAdapter(langchain_tool: LangChainTool)

基类： BaseTool[BaseModel, Any]

允许您包装 LangChain 工具并使其可用于 AutoGen。

注意

此类需要 autogen-ext 包的 langchain 额外依赖。

pip install -U "autogen-ext[langchain]"

参数:

langchain_tool (LangChainTool) – 要包装的 LangChain 工具

示例

使用 langchain_experimental 包中的 PythonAstREPLTool 创建一个允许您与 Pandas DataFrame 交互的工具。

import asyncio
import pandas as pd
from langchain_experimental.tools.python.tool import PythonAstREPLTool
from autogen_ext.tools.langchain import LangChainToolAdapter
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.messages import TextMessage
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_core import CancellationToken


async def main() -> None:
    df = pd.read_csv("https://raw.githubusercontent.com/pandas-dev/pandas/main/doc/data/titanic.csv")  # type: ignore
    tool = LangChainToolAdapter(PythonAstREPLTool(locals={"df": df}))
    model_client = OpenAIChatCompletionClient(model="gpt-4o")
    agent = AssistantAgent(
        "assistant",
        tools=[tool],
        model_client=model_client,
        system_message="Use the `df` variable to access the dataset.",
    )
    await Console(
        agent.on_messages_stream(
            [TextMessage(content="What's the average age of the passengers?", source="user")], CancellationToken()
        )
    )


asyncio.run(main())

此示例演示如何使用 langchain_community 包中的 SQLDatabaseToolkit 与 SQLite 数据库交互。它使用 RoundRobinGroupChat 在多个步骤中迭代单个代理。如果您想一次执行一步，您可以直接调用 AssistantAgent 类的 run_stream 方法。

import asyncio
import sqlite3

import requests
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.langchain import LangChainToolAdapter
from langchain_community.agent_toolkits.sql.toolkit import SQLDatabaseToolkit
from langchain_community.utilities.sql_database import SQLDatabase
from langchain_openai import ChatOpenAI
from sqlalchemy import Engine, create_engine
from sqlalchemy.pool import StaticPool


def get_engine_for_chinook_db() -> Engine:
    url = "https://raw.githubusercontent.com/lerocha/chinook-database/master/ChinookDatabase/DataSources/Chinook_Sqlite.sql"
    response = requests.get(url)
    sql_script = response.text
    connection = sqlite3.connect(":memory:", check_same_thread=False)
    connection.executescript(sql_script)
    return create_engine(
        "sqlite://",
        creator=lambda: connection,
        poolclass=StaticPool,
        connect_args={"check_same_thread": False},
    )


async def main() -> None:
    # Create the engine and database wrapper.
    engine = get_engine_for_chinook_db()
    db = SQLDatabase(engine)

    # Create the toolkit.
    llm = ChatOpenAI(temperature=0)
    toolkit = SQLDatabaseToolkit(db=db, llm=llm)

    # Create the LangChain tool adapter for every tool in the toolkit.
    tools = [LangChainToolAdapter(tool) for tool in toolkit.get_tools()]

    # Create the chat completion client.
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    # Create the assistant agent.
    agent = AssistantAgent(
        "assistant",
        model_client=model_client,
        tools=tools,  # type: ignore
        model_client_stream=True,
        system_message="Respond with 'TERMINATE' if the task is completed.",
    )

    # Create termination condition.
    termination = TextMentionTermination("TERMINATE")

    # Create a round-robin group chat to iterate the single agent over multiple steps.
    chat = RoundRobinGroupChat([agent], termination_condition=termination)

    # Run the chat.
    await Console(chat.run_stream(task="Show some tables in the database"))


if __name__ == "__main__":
    asyncio.run(main())

async run(args: BaseModel, cancellation_token: CancellationToken) → Any