langgraph-bigtool

repository·main·Indexed 19 days ago

https://github.com/langchain-ai/langgraph-bigtool

A Python library for building LangGraph agents capable of interacting with massive toolsets. It uses a retrieval-based approach via LangGraph's persistence layer (such as InMemoryStore or PostgresStore) and semantic search to index tool metadata, allowing agents to dynamically discover and execute relevant tools from a registry without exceeding LLM context limits.

Tokens
3.3K
Snippets
10
Records
12
Agent score
68%

What's inside langgraph-bigtool

  1. How langgraph-bigtool works: Scalable tool access via semantic search

    main

    Instead of providing hundreds or thousands of tools directly to an LLM (which can exceed context limits or degrade performance), langgraph-bigtool uses a retrieval-based approach:

    1. Tool Registry: A dictionary mapping unique identifiers to tool instances.
    2. LangGraph Store: Tool metadata (names and descriptions) is indexed in a LangGraph Store (e.g., InMemoryStore or PostgresStore) using embeddings.
    3. Retrieval Tool: The agent is equipped with a specialized tool that searches the Store for relevant tool IDs based on the user's query.
    4. Execution: The agent retrieves the relevant tool IDs, identifies the corresponding tools from the registry, and then executes them.
  2. Quickstart: Create an agent with many tools

    main

    This example demonstrates how to equip an agent with all functions from Python's math library by indexing them in an InMemoryStore.

    To run this, you will need langgraph-bigtool, langchain[openai], and an OPENAI_API_KEY environment variable set.

    import math
    import types
    import uuid
    
    from langchain.chat_models import init_chat_model
    from langchain.embeddings import init_embeddings
    from langgraph.store.memory import InMemoryStore
    
    from langgraph_bigtool import create_agent
    from langgraph_bigtool.utils import (
        convert_positional_only_function_to_tool
    )
    
    # 1. Collect functions from `math` built-in
    all_tools = []
    for function_name in dir(math):
        function = getattr(math, function_name)
        if not isinstance(
            function, types.BuiltinFunctionType
        ):
            continue
        # Handle positional-only functions common in math library
        if tool := convert_positional_only_function_to_tool(
            function
        ):
            all_tools.append(tool)
    
    # 2. Create registry of tools (dict mapping identifiers to tool instances)
    tool_registry = {
        str(uuid.uuid4()): tool
        for tool in all_tools
    }
    
    # 3. Index tool names and descriptions in the LangGraph Store
    embeddings = init_embeddings("openai:text-embedding-3-small")
    
    store = InMemoryStore(
        index={
            "embed": embeddings,
            "dims": 1536,
            "fields": ["description"],
        }
    )
    for tool_id, tool in tool_registry.items():
        store.put(
            ("tools",),
            tool_id,
            {
                "description": f"{tool.name}: {tool.description}",
            },
        )
    
    # 4. Initialize agent
    llm = init_chat_model("openai:gpt-4o-mini")
    
    builder = create_agent(llm, tool_registry)
    agent = builder.compile(store=store)
    
    # 5. Test it out
    query = "Use available tools to calculate arc cosine of 0.5."
    
    for step in agent.stream(
        {"messages": query},
        stream_mode="updates",
    ):
        for _, update in step.items():
            for message in update.get("messages", []):
                message.pretty_print()
    import math
    import types
    import uuid
    
    from langchain.chat_models import init_chat_model
    from langchain.embeddings import init_embeddings
    from langgraph.store.memory import InMemoryStore
    
    from langgraph_bigtool import create_agent
    from langgraph_bigtool.utils import (
        convert_positional_only_function_to_tool
    )
    
    # 1. Collect functions from `math` built-in
    all_tools = []
    for function_name in dir(math):
        function = getattr(math, function_name)
        if not isinstance(
            function, types.BuiltinFunctionType
        ):
            continue
        # Handle positional-only functions common in math library
        if tool := convert_positional_only_function_to_tool(
            function
        ):
            all_tools.append(tool)
    
    # 2. Create registry of tools (dict mapping identifiers to tool instances)
    tool_registry = {
        str(uuid.uuid4()): tool
        for tool in all_tools
    }
    
    # 3. Index tool names and descriptions in the LangGraph Store
    embeddings = init_embeddings("openai:text-embedding-3-small")
    
    store = InMemoryStore(
        index={
            "embed": embeddings,
            "dims": 1536,
            "fields": ["description"],
        }
    )
    for tool_id, tool in tool_registry.items():
        store.put(
            ("tools",),
            tool_id,
            {
                "description": f"{tool.name}: {tool.description}",
            },
        )
    
    # 4. Initialize agent
    llm = init_chat_model("openai:gpt-4o-mini")
    
    builder = create_agent(llm, tool_registry)
    agent = builder.compile(store=store)
    
    # 5. Test it out
    query = "Use available tools to calculate arc cosine of 0.5."
    
    for step in agent.stream(
        {"messages": query},
        stream_mode="updates",
    ):
        for _, update in step.items():
            for message in update.get("messages", []):
                message.pretty_print()
  3. Implement arbitrary tool retrieval logic

    main

    If you do not want to use semantic search via the LangGraph Store, you can implement custom logic that returns tool IDs based on other criteria (e.g., categories).

    Tip: If you use typing.Literal for function arguments, the LLM will be signaled to provide a categorical value.

    from typing import Literal
    
    tool_registry = {
        "id_1": get_balance,
        "id_2": get_history,
        "id_3": create_ticket,
    }
    
    def retrieve_tools(
        category: Literal["billing", "service"],
    ) -> list[str]:
        """Get tools for a category."""
        if category == "billing":
            return ["id_1", "id_2"]
        else:
            return ["id_3"]
    
    # Use this in create_agent
    builder = create_agent(
        llm, tool_registry, retrieve_tools_function=retrieve_tools
    )
    agent = builder.compile(store=store)
  4. Customize tool retrieval with LangGraph Store

    main

    You can override the default semantic search behavior by providing a custom retrieve_tools_function or retrieve_tools_coroutine to create_agent. The function must return a list of tool IDs.

    To use the LangGraph Store within your custom function, use the InjectedStore annotation.

    from langgraph.prebuilt import InjectedStore
    from langgraph.store.base import BaseStore
    from typing_extensions import Annotated
    
    
    def retrieve_tools(
        query: str,
        *, 
        store: Annotated[BaseStore, InjectedStore],
    ) -> list[str]:
        """Retrieve a tool to use, given a search query."""
        results = store.search(("tools",), query=query, limit=2)
        tool_ids = [result.key for result in results]
        return tool_ids
    
    builder = create_agent(
        llm, tool_registry, retrieve_tools_function=retrieve_tools
    )
    agent = builder.compile(store=store)
    from langgraph.prebuilt import InjectedStore
    from langgraph.store.base import BaseStore
    from typing_extensions import Annotated
    
    
    def retrieve_tools(
        query: str,
        *, 
        store: Annotated[BaseStore, InjectedStore],
    ) -> list[str]:
        """Retrieve a tool to use, given a search query."""
        results = store.search(("tools",), query=query, limit=2)
        tool_ids = [result.key for result in results]
        return tool_ids
    
    builder = create_agent(
        llm, tool_registry, retrieve_tools_function=retrieve_tools
    )
    agent = builder.compile(store=store)
  5. Retrieve tools using retrieve_tools

    main

    Use retrieve_tools to perform a synchronous search for tools within a LangGraph BaseStore based on a query string. This function is designed to be used within a LangGraph node where the store is provided via InjectedStore.

    from typing import Annotated
    from langgraph.prebuilt import InjectedStore
    from langgraph.store.base import BaseStore
    from langgraph_bigtool.tools import retrieve_tools
    
    # Note: In a real LangGraph node, 'store' is injected automatically
    async def my_node(query: str, store: Annotated[BaseStore, InjectedStore]):
        tool_ids = retrieve_tools(query=query, store=store)
        return {"tool_ids": tool_ids}
  6. Retrieve tools asynchronously using aretrieve_tools

    main

    Use aretrieve_tools to perform an asynchronous search for tools within a LangGraph BaseStore based on a query string. This is the preferred method for async LangGraph workflows.

    from typing import Annotated
    from langgraph.prebuilt import InjectedStore
    from langgraph.store.base import BaseStore
    from langgraph_bigtool.tools import aretrieve_tools
    
    # Note: In a real LangGraph node, 'store' is injected automatically
    async def my_node(query: str, store: Annotated[BaseStore, InjectedStore]):
        tool_ids = await aretrieve_tools(query=query, store=store)
        return {"tool_ids": tool_ids}
  7. Identify tool arguments requiring InjectedStore using get_store_arg

    main

    The get_store_arg function inspects a BaseTool's input schema to find the name of the argument annotated with InjectedStore. This is useful for identifying which part of a tool's input schema is intended to receive the LangGraph store.

    from langchain_core.tools import BaseTool
    from langgraph_bigtool.tools import get_store_arg
    
    # Assuming 'my_tool' is a BaseTool where one argument is annotated with InjectedStore
    store_arg_name = get_store_arg(my_tool)
    if store_arg_name:
        print(f"The tool requires the store in argument: {store_arg_name}")
  8. Create a tool-retrieval agent with create_agent()

    main

    Use create_agent() to initialize a ReAct-style agent that starts with only a tool for retrieving other tools from a registry. As the agent runs, it can discover and bind new tools dynamically. The agent uses a StateGraph and relies on a BaseStore to perform semantic searches for tools.

    Arguments

    • llm: A LanguageModelLike instance.
    • tool_registry: A dictionary mapping string IDs to BaseTool instances or callables.
    • limit (int, default 2): Maximum number of tools to retrieve per selection step.
    • filter (dict | None): Key-value pairs to filter tool retrieval results.
    • namespace_prefix (tuple[str, ...], default ("tools",)): The hierarchical path in the BaseStore where tools are indexed.
    • retrieve_tools_function (Callable | None): A function that returns a list of tool IDs. If omitted, the agent uses semantic search against the Store.
    • retrieve_tools_coroutine (Callable | None): An async version of the retrieval function. If omitted, the agent uses semantic search against the Store.
    from langchain_core.language_models import ChatOpenAI
    from langgraph_bigtool.graph import create_agent
    
    llm = ChatOpenAI(model="gpt-4o")
    tool_registry = {
        "math_add": lambda x, y: x + y,
        # ... other tools
    }
    
    agent_graph = create_agent(
        llm=llm,
        tool_registry=tool_registry,
        limit=5,
        namespace_prefix=("my_custom_tools",)
    )
  9. Define the State for the agent

    main

    The agent's state is managed by the State class, which inherits from langgraph.graph.MessagesState. It includes a specialized field selected_tool_ids which uses an _add_new reducer to ensure that newly discovered tool IDs are appended to the existing list without duplicates.

    selected_tool_ids: Annotated[list[str], _add_new]

    from langgraph_bigtool.graph import State
    
    # State contains 'messages' (from MessagesState) 
    # and 'selected_tool_ids' (list of strings)