Structured Tools

Обновлено 24 мая 2024

from typing import List

from langchain_core.tools import tool


@tool
def get_data(n: int) -> List[dict]:
    """Get n datapoints."""
    return [{"name": "foo", "value": "bar"}] * n


tools = [get_data]

We will use a prompt from the hub - you can inspect the prompt more at https://smith.langchain.com/hub/hwchase17/openai-functions-agent

from langchain import hub
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI

# Get the prompt to use - you can modify this!
# If you want to see the prompt in full, you can at: https://smith.langchain.com/hub/hwchase17/openai-functions-agent
prompt = hub.pull("hwchase17/openai-functions-agent")

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

Stream intermediate steps

Let's look at how to stream intermediate steps. We can do this easily by just using the .stream method on the AgentExecutor

We can then parse the results to get actions (tool inputs) and observtions (tool outputs).

for chunk in agent_executor.stream({"input": "get me three datapoints"}):
    # Agent Action
    if "actions" in chunk:
        for action in chunk["actions"]:
            print(
                f"Calling Tool ```{action.tool}``` with input ```{action.tool_input}```"
            )
    # Observation
    elif "steps" in chunk:
        for step in chunk["steps"]:
            print(f"Got result: ```{step.observation}```")

    Calling Tool ```get_data``` with input ```{'n': 3}```
    Got result: ```[{'name': 'foo', 'value': 'bar'}, {'name': 'foo', 'value': 'bar'}, {'name': 'foo', 'value': 'bar'}]```