Replies: 2 comments 15 replies
-
|
Hey there, @suribe06! I'm here to help you out with any coding conundrums you may have. I can assist with bug squashing, answer your burning questions, and even guide you on the path to becoming a contributor. Let's tackle this together! To prevent the LLM from overwriting the parameters passed to a LangChain tool in a custom agent, you can use the Here is an example of how to do this: from typing import List
from langchain_core.tools import InjectedToolArg, tool
from typing_extensions import Annotated
user_to_pets = {}
@tool(parse_docstring=True)
def update_favorite_pets(
pets: List[str], user_id: Annotated[str, InjectedToolArg]
) -> None:
"""Add the list of favorite pets.
Args:
pets: List of favorite pets to set.
user_id: User's ID.
"""
user_to_pets[user_id] = pets
@tool(parse_docstring=True)
def delete_favorite_pets(user_id: Annotated[str, InjectedToolArg]) -> None:
"""Delete the list of favorite pets.
Args:
user_id: User's ID.
"""
if user_id in user_to_pets:
del user_to_pets[user_id]
@tool(parse_docstring=True)
def list_favorite_pets(user_id: Annotated[str, InjectedToolArg]) -> None:
"""List favorite pets if any.
Args:
user_id: User's ID.
"""
return user_to_pets.get(user_id, [])When you invoke the tool, you need to pass in the user_id = "123"
update_favorite_pets.invoke({"pets": ["lizard", "dog"], "user_id": user_id})
print(user_to_pets)
print(list_favorite_pets.invoke({"user_id": user_id}))When the model calls the tool, no from copy import deepcopy
from langchain_core.runnables import chain
@chain
def inject_user_id(ai_msg):
tool_calls = []
for tool_call in ai_msg.tool_calls:
tool_call_copy = deepcopy(tool_call)
tool_call_copy["args"]["user_id"] = user_id
tool_calls.append(tool_call_copy)
return tool_calls
inject_user_id.invoke(ai_msg)Finally, chain together your model, injection code, and the actual tools to create a tool-executing chain: tool_map = {tool.name: tool for tool in tools}
@chain
def tool_router(tool_call):
return tool_map[tool_call["name"]]
chain = llm_with_tools | inject_user_id | tool_router.map()
chain.invoke("my favorite animals are cats and parrots")This approach ensures that the |
Beta Was this translation helpful? Give feedback.
-
|
To properly structure the chain to integrate the agent, tool, and parameter injection without encountering an "IndexError: list index out of range," you need to ensure that the
By following these steps, you ensure that the parameters are correctly injected into the tool calls at runtime, preventing the "IndexError: list index out of range" error. This approach ensures that the |
Beta Was this translation helpful? Give feedback.
-
|
@dosu The error is raised in the lines: |
Beta Was this translation helpful? Give feedback.
-
|
The "IndexError: list index out of range" error you're encountering is likely due to the To resolve this, you need to ensure that the input dictionary passed to the
Here's the updated code: from copy import deepcopy
from langchain_core.runnables import chain
from langchain_core.prompts import ChatPromptTemplate
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain.chat_models import ChatOpenAI
# Define the tool as you have done
from langchain_core.tools import InjectedToolArg, tool
from typing import Optional, Annotated
@tool(parse_docstring=False)
def api_tool(api_name: Annotated[str, InjectedToolArg],
method: Annotated[str, InjectedToolArg],
endpoint: Annotated[str, InjectedToolArg],
params: Optional[dict] = None, data: Optional[dict] = None) -> dict:
"""
Call an API using the specified parameters.
Args:
api_name (str): The name of the API.
method (str): The HTTP method to use.
endpoint (str): The endpoint path to call.
params (dict, optional): The query parameters to include in the request.
data (dict, optional): The body data to include in the request.
Returns:
dict: The JSON response from the API.
"""
swagger_spec = get_swagger_file(api_name)
paths = swagger_spec.get("paths", {})
if endpoint not in paths:
return {"error": "Path not found"}
if method not in paths[endpoint]:
return {"error": "Method not found"}
server = swagger_spec['servers'][0]['url']
url = f"{server}{endpoint}"
# Make the API call
response = requests.request(method, url, params=params)
return response.json()
# Function to inject parameters at runtime
@chain
def inject_parameters(ai_msg, api_name, method, endpoint):
tool_calls = []
for tool_call in ai_msg.tool_calls:
tool_call_copy = deepcopy(tool_call)
tool_call_copy["args"]["api_name"] = api_name
tool_call_copy["args"]["method"] = method
tool_call_copy["args"]["endpoint"] = endpoint
tool_calls.append(tool_call_copy)
return tool_calls
# Set up the agent and agent executor
prompt = ChatPromptTemplate.from_messages(
[
("system", "You are a helpful assistant"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
]
)
llm = ChatOpenAI(temperature=0.7, model_name="gpt-4")
agent = create_tool_calling_agent(
tools=[api_tool],
llm=llm,
prompt=prompt
)
agent_executor = AgentExecutor(agent=agent, tools=[api_tool], verbose=True)
# Parameters to inject
api_name = "trakt.tv"
method = "get"
endpoint = "/movies/anticipated"
# Create the chain
chain = agent_executor | inject_parameters
# Ensure the input dictionary contains all required keys
inputs = {
"input": "What are the most anticipated movies?",
"api_name": api_name,
"method": method,
"endpoint": endpoint,
"agent_scratchpad": "" # Add any other necessary keys
}
# Invoke the chain
response = chain.invoke(inputs)
print(response)In this setup:
This should prevent the "IndexError: list index out of range" error by ensuring that all required inputs are provided [1]. |
Beta Was this translation helpful? Give feedback.
-
|
@dosu However, I’m getting the following error: It seems like there’s an issue with how I’m passing the input to the |
Beta Was this translation helpful? Give feedback.
-
|
To correctly use
This setup ensures that the Here is the complete code with the necessary adjustments: from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain.chat_models import ChatOpenAI
from copy import deepcopy
from langchain_core.tools import InjectedToolArg, tool
from typing import Optional, Annotated
# Define the tool
@tool(parse_docstring=False)
def api_tool(api_name: Annotated[str, InjectedToolArg],
method: Annotated[str, InjectedToolArg],
endpoint: Annotated[str, InjectedToolArg],
params: Optional[dict] = None, data: Optional[dict] = None) -> dict:
"""
Call an API using the specified parameters.
Args:
api_name (str): The name of the API.
method (str): The HTTP method to use.
endpoint (str): The endpoint path to call.
params (dict, optional): The query parameters to include in the request.
data (dict, optional): The body data to include in the request.
Returns:
dict: The JSON response from the API.
"""
swagger_spec = get_swagger_file(api_name)
paths = swagger_spec.get("paths", {})
if endpoint not in paths:
return {"error": "Path not found"}
if method not in paths[endpoint]:
return {"error": "Method not found"}
server = swagger_spec['servers'][0]['url']
url = f"{server}{endpoint}"
# Make the API call
response = requests.request(method, url, params=params)
return response.json()
# Function to inject parameters at runtime
@chain
def inject_parameters(ai_msg, api_name, method, endpoint):
tool_calls = []
for tool_call in ai_msg.tool_calls:
tool_call_copy = deepcopy(tool_call)
tool_call_copy["args"]["api_name"] = api_name
tool_call_copy["args"]["method"] = method
tool_call_copy["args"]["endpoint"] = endpoint
tool_calls.append(tool_call_copy)
return tool_calls
# Set up the agent and agent executor
prompt = ChatPromptTemplate.from_messages(
[
("system", "You are a helpful assistant"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
]
)
llm = ChatOpenAI(temperature=0.7, model_name="gpt-4")
agent = create_tool_calling_agent(
tools=[api_tool],
llm=llm,
prompt=prompt
)
agent_executor = AgentExecutor(agent=agent, tools=[api_tool], verbose=True)
# Parameters to inject
api_name = "trakt.tv"
method = "get"
endpoint = "/movies/anticipated"
# Create the RunnableParallel instance
injection_step = RunnableParallel(
{
"ai_msg": llm_with_tools,
"params": RunnablePassthrough().assign(api_name=api_name, method=method, endpoint=endpoint)
}
) | inject_parameters
# Create the chain
chain = injection_step | api_tool
# Invoke the chain
response = chain.invoke({"input": "What are the most anticipated movies?"})
print(response)This should resolve the |
Beta Was this translation helpful? Give feedback.
-
|
@dosu But the |
Beta Was this translation helpful? Give feedback.
-
|
The Here's how you can modify your
Here's the updated code: from langchain_core.runnables import RunnableLambda
llm_step = llm_with_tools
# Define the parameter injection step
injection_step = inject_parameters
# Define the API tool invocation step
def tool_invocation_step_factory(tool_map):
def invoke_tool(response):
tool_name = response[0]["name"]
selected_tool = tool_map[tool_name]
args = response[0]["args"]
return selected_tool.invoke(input=args) # Pass the args as the input argument
return RunnableLambda(invoke_tool)
def create_custom_chain(tools, api_name, method, endpoint):
tool_map = {tool.name: tool for tool in tools}
chain = llm_step | RunnableLambda(lambda ai_msg: inject_parameters.invoke(ai_msg, api_name=api_name, method=method, endpoint=endpoint)) | tool_invocation_step_factory(tool_map)
return chain
api_name = "trakt.tv"
method = "get"
endpoint = "/movies/anticipated"
tools = [api_tool]
chain = create_custom_chain(tools, api_name, method, endpoint)
response = chain.invoke("What are the most anticipated movies?")
print(response)In this updated code:
This should resolve the |
Beta Was this translation helpful? Give feedback.
-
|
@dosu The idea is to pass both the user query and the response from the API (generated in the tool_invocation_step_factory step) into an LLM to structure the final output. In this new step ( |
Beta Was this translation helpful? Give feedback.
-
|
To add an extra step to your custom chain where an LLM formats the output based on the original user query and the result of the API call, you can use a
Here's the complete implementation: from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_openai import OpenAI
# Define the prompt template for formatting the output
prompt_template = """
Please format the API response based on the user query.
User Query:
{input}
API Response:
{api_response}
Provide a clear and concise summary or structured output.
"""
prompt = PromptTemplate(
input_variables=["input", "api_response"],
template=prompt_template,
)
# Initialize the LLM for formatting
llm = OpenAI()
# Create the LLM chain for formatting
llm_chain = prompt | llm
# Define the API tool invocation step
def tool_invocation_step_factory(tool_map):
def invoke_tool(response):
tool_name = response[0]["name"]
selected_tool = tool_map[tool_name]
args = response[0]["args"]
return selected_tool.invoke(input=args) # Pass the args as the input argument
return RunnableLambda(invoke_tool)
# Define the custom chain creation function
def create_custom_chain(tools, api_name, method, endpoint):
tool_map = {tool.name: tool for tool in tools}
chain = (
llm_with_tools |
RunnableLambda(lambda ai_msg: inject_parameters.invoke(ai_msg, api_name=api_name, method=method, endpoint=endpoint)) |
tool_invocation_step_factory(tool_map) |
RunnableLambda(lambda result: llm_chain.invoke({"input": result["input"], "api_response": result["response"]}))
)
return chain
# Parameters to inject
api_name = "trakt.tv"
method = "get"
endpoint = "/movies/anticipated"
tools = [api_tool]
# Create the custom chain
chain = create_custom_chain(tools, api_name, method, endpoint)
# Invoke the chain with the user query
response = chain.invoke("What are the most anticipated movies?")
print(response)In this setup:
This ensures that the LLM formats the output based on the original user query and the result of the API call using the specified prompt template [1]. |
Beta Was this translation helpful? Give feedback.
-
|
@dosu However, I’m encountering an error when trying to load the specs: The traceback indicates that the error occurs in the OpenAPISpec.parse_obj method when loading the spec: |
Beta Was this translation helpful? Give feedback.
-
|
To correctly load OpenAPI specifications using the Here is the relevant part of the code that shows how to correctly load the OpenAPI specifications: from openapi_pydantic import OpenAPI
import json
import yaml
from pathlib import Path
from typing import Union
import requests
import copy
from pydantic import ValidationError
class OpenAPISpec(OpenAPI):
"""OpenAPI Model that removes mis-formatted parts of the spec."""
openapi: str = "3.1.0" # overriding overly restrictive type from parent class
@classmethod
def parse_obj(cls, obj: dict) -> OpenAPISpec:
try:
cls._alert_unsupported_spec(obj)
return super().parse_obj(obj)
except ValidationError as e:
# We are handling possibly misconfigured specs and
# want to do a best-effort job to get a reasonable interface out of it.
new_obj = copy.deepcopy(obj)
for error in e.errors():
keys = error["loc"]
item = new_obj
for key in keys[:-1]:
item = item[key]
item.pop(keys[-1], None)
return cls.parse_obj(new_obj)
@classmethod
def from_spec_dict(cls, spec_dict: dict) -> OpenAPISpec:
"""Get an OpenAPI spec from a dict."""
return cls.parse_obj(spec_dict)
@classmethod
def from_text(cls, text: str) -> OpenAPISpec:
"""Get an OpenAPI spec from a text."""
try:
spec_dict = json.loads(text)
except json.JSONDecodeError:
spec_dict = yaml.safe_load(text)
return cls.from_spec_dict(spec_dict)
@classmethod
def from_file(cls, path: Union[str, Path]) -> OpenAPISpec:
"""Get an OpenAPI spec from a file path."""
path_ = path if isinstance(path, Path) else Path(path)
if not path_.exists():
raise FileNotFoundError(f"{path} does not exist")
with path_.open("r") as f:
return cls.from_text(f.read())
@classmethod
def from_url(cls, url: str) -> OpenAPISpec:
"""Get an OpenAPI spec from a URL."""
response = requests.get(url)
return cls.from_text(response.text)Ensure that the |
Beta Was this translation helpful? Give feedback.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
-
Checked other resources
Commit to Help
Example Code
Description
I'm working on a custom LangChain agent that uses a tool to make API calls based on parameters passed in the code. However, I'm encountering an issue where the LLM is generating its own parameters instead of using the ones I pass. The problem is that the tool is not using the parameters I provide, but instead, the LLM seems to be generating them. Here’s the output:
As you can see, the api_name has changed to movieDatabase, even though I passed trakt.tv. I want the tool to use the parameters I define in the code, not those generated by the LLM.
There a way to ensure that the parameters I pass are respected?
Any advice or suggestions would be greatly appreciated.
System Info
langchainlangchain-communitylangchain-corelangchain-experimentallangchain-huggingfacelangchain-openailangchain-text-splitterslangcodeslangsmithBeta Was this translation helpful? Give feedback.
All reactions