Expected mapping type as input to PromptTemplate. Received <class 'langchain_core.messages.ai.AIMessage

[intern162]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from src.configs import set_environ, api_key
set_environ()
#%%
from typing import TypedDict, List, Annotated
from langgraph.graph import StateGraph, END, START
import operator
from src.file_manager import get_file_content_as_string
#%%

from langchain_core.messages import BaseMessage, HumanMessage
from typing import Sequence


class CustomState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    next: str
    file_in_focus: str
    root_dir: str
    error_list: List[str]
    iterations: int

#%%
from langchain_community.tools import ShellTool
from langchain_community.agent_toolkits import FileManagementToolkit
file_manager = FileManagementToolkit(
    root_dir='/path/to/root/directory'
).get_tools()
shell_tool = ShellTool()
shell_tool.description = shell_tool.description + f"args {shell_tool.args}".replace(
    "{", "{{"
).replace("}", "}}")
#%%
from langchain_openai import ChatOpenAI
model = my_coder_agent = ChatOpenAI(
        model='your-model-name',
        openai_api_key=api_key,
        openai_api_base='https://api.yourapi.com/v1',
        temperature=1.0,
        max_tokens=8192,
        timeout=None,
        max_retries=2,
        api_key=api_key
    )
#%%
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
class SupervisorResponse(BaseModel):
    message: str= Field(description="The message/task to give to the next agent")
    next: str= Field(description="enum out of FIX, BASH, or END")

#%%
supervisor_parser = PydanticOutputParser(pydantic_object=SupervisorResponse)
supervisor_prompt = PromptTemplate(
    template="Based on the current state and the information provided, determine the next step. You can either provide a message/task for agents.\n\nInstructions:\n{format_instructions}\n\nCurrent Conversation:\n{messages}\n\nList of Errors:\n{error_list}\n\nFile Path Needed for Shell Operations:\n{file_in_focus}\n\nRoot Directory of the Application:\n{root_dir}",
    input_variables=["error_list", "messages", "file_in_focus", "root_dir"],
    partial_variables={"format_instructions": supervisor_parser.get_format_instructions()},
    input_types={"messages": List[BaseMessage],
                 "error_list": List[str],
                 "file_in_focus": str,
                 'root_dir': str}
)
#%%
supervisor_chain = supervisor_prompt | model | supervisor_prompt
#%%
def supervisor_node(state: CustomState) -> CustomState:
    # Append the new message to the messages list
    state['messages'].append(HumanMessage(content="Based on the current state and the information provided, determine the next step."))

    # Invoke the supervisor chain
    supervisor_response = supervisor_chain.invoke({
        "file_in_focus": state['file_in_focus'],
        "error_list": state['error_list'],
        "messages": state['messages'],
        "root_dir": state['root_dir']
    })

    # Append the supervisor's response to the messages list
    state['messages'].append(HumanMessage(content=supervisor_response.message))

    # Return the updated state
    return {
        'messages': state['messages'],
        'next': supervisor_response.next,
        'file_in_focus': state['file_in_focus'],
        'root_dir': state['root_dir'],
        'error_list': state['error_list'],
        'iterations': 0
    }
#%%
workflow = StateGraph(CustomState)
#%%
workflow.add_node("supervisor", supervisor_node)
workflow.add_edge("__start__", "supervisor")
workflow.add_edge("supervisor", "__end__")
#%%
app = workflow.compile()
#%%
result = app.invoke({'messages': [HumanMessage(content="Run the code i tell you in root_dir")],
        'next': "",
        'file_in_focus': "",
        'root_dir': "/path/to/root/directory",
        'error_list': [],
        'iterations': 0})

Description

I am trying to use this graph just to test if the supervisor components works but the promp template keeps failing with type error:
Expected mapping type as input to PromptTemplate. Received <class 'langchain_core.messages.ai.AIMessage
Switching to str instead of HumanMessage, also does not seem to help.

System Info

M1 Pro Mac

[dosubot[bot]]

Hey @intern162! 👋 I'm here to help you tackle any bugs, answer questions, and guide you on how to contribute. If you have any issues or queries, feel free to ask!

To resolve the type error "Expected mapping type as input to PromptTemplate. Received <class 'langchain_core.messages.ai.AIMessage>" when using a LangChain graph with a prompt template, ensure that you are passing a dictionary (mapping type) to the PromptTemplate instead of an AIMessage object.

Here is an example of how to correctly use the ChatPromptTemplate with a dictionary:

from langchain_core.prompts import ChatPromptTemplate

chat_template = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful AI bot. Your name is {name}."),
        ("human", "Hello, how are you doing?"),
        ("ai", "I'm doing well, thanks!"),
        ("human", "{user_input}"),
    ]
)

messages = chat_template.format_messages(name="Bob", user_input="What is your name?")

In your case, ensure that the messages input to the supervisor_chain.invoke method is a list of dictionaries, not AIMessage objects. You can convert your HumanMessage objects to dictionaries before passing them to the invoke method.

Here's how you can modify your supervisor_node function:

def supervisor_node(state: CustomState) -> CustomState:
    # Append the new message to the messages list
    state['messages'].append(HumanMessage(content="Based on the current state and the information provided, determine the next step."))

    # Convert messages to dictionaries
    messages_dict = [{"content": msg.content, "type": msg.type} for msg in state['messages']]

    # Invoke the supervisor chain
    supervisor_response = supervisor_chain.invoke({
        "file_in_focus": state['file_in_focus'],
        "error_list": state['error_list'],
        "messages": messages_dict,
        "root_dir": state['root_dir']
    })

    # Append the supervisor's response to the messages list
    state['messages'].append(HumanMessage(content=supervisor_response.message))

    # Return the updated state
    return {
        'messages': state['messages'],
        'next': supervisor_response.next,
        'file_in_focus': state['file_in_focus'],
        'root_dir': state['root_dir'],
        'error_list': state['error_list'],
        'iterations': 0
    }

This should resolve the type error you are encountering ^[1][2].

To continue talking to Dosu, mention @dosu.

[intern162]

@dosu
I have changed all my messages from human messages to
class CustomState(TypedDict):
messages: Annotated[List[Dict[str, Any]], operator.add]
next: str
file_in_focus: str
root_dir: str
error_list: List[str]
iterations: int
Still facing the same issue despite making appropriate changes for that.

[dosubot[bot]]

Ensure that the input to the PromptTemplate is a dictionary and not an instance of AIMessage. The _validate_input method in the BasePromptTemplate class expects the input to be a dictionary. If the input is not a dictionary, it raises a TypeError.

Here's how you can modify your supervisor_node function to ensure the input is correctly formatted:

def supervisor_node(state: CustomState) -> CustomState:
    # Append the new message to the messages list
    state['messages'].append({"content": "Based on the current state and the information provided, determine the next step.", "type": "human"})

    # Invoke the supervisor chain
    supervisor_response = supervisor_chain.invoke({
        "file_in_focus": state['file_in_focus'],
        "error_list": state['error_list'],
        "messages": state['messages'],
        "root_dir": state['root_dir']
    })

    # Append the supervisor's response to the messages list
    state['messages'].append({"content": supervisor_response["message"], "type": "human"})

    # Return the updated state
    return {
        'messages': state['messages'],
        'next': supervisor_response["next"],
        'file_in_focus': state['file_in_focus'],
        'root_dir': state['root_dir'],
        'error_list': state['error_list'],
        'iterations': 0
    }

Ensure that the messages input to the supervisor_chain.invoke method is a list of dictionaries, not AIMessage objects. This should resolve the type error you are encountering ^[1][2].

To continue talking to Dosu, mention @dosu.

[intern162]

@dosu

from src.configs import set_environ, api_key
set_environ()

#%%
from typing import TypedDict, List, Annotated
from langgraph.graph import StateGraph
import operator

#%%
from typing import Dict, Any

class CustomState(TypedDict):
    messages: Annotated[List[Dict[str, Any]], operator.add]
    next: str
    file_in_focus: str
    root_dir: str
    error_list: List[str]
    iterations: int

#%%
from langchain_community.tools import ShellTool
from langchain_community.agent_toolkits import FileManagementToolkit
file_manager = FileManagementToolkit(
    root_dir='/path/to/your/project'
).get_tools()
shell_tool = ShellTool()
shell_tool.description = shell_tool.description + f"args {shell_tool.args}".replace(
    "{", "{{"
).replace("}", "}}")

#%%
from langchain_openai import ChatOpenAI
model = deepseek_coder_beta_agent = ChatOpenAI(
        model='deepseek-coder',
        openai_api_key=api_key,
        openai_api_base='https://api.yourapi.com/beta',
        temperature=1.0,
        max_tokens=8192,
        timeout=None,
        max_retries=2,
        api_key=api_key
    )

#%%
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import PydanticOutputParser

class SupervisorResponse(BaseModel):
    message: str = Field(description="The message/task to give to the next agent")
    next: str = Field(description="enum out of FIX, BASH, or END")

#%%
supervisor_parser = PydanticOutputParser(pydantic_object=SupervisorResponse)
supervisor_prompt = PromptTemplate(
    template=("Answer the following.\n{format_instructions}\n After analyzing the received information, "
              "choose the message/task to give to the next agent."
              "This is the conversation until now: {messages}\n These are the errors: {error_list}\n"
              "and the file path needed according to bash is: {file_in_focus}.\n"
              "This is the root directory holding the application: {root_dir}"),
    input_variables=["error_list", "messages", "file_in_focus", "root_dir"],
    partial_variables={"format_instructions": supervisor_parser.get_format_instructions()}
)

#%%
supervisor_chain = supervisor_prompt | model | supervisor_prompt

#%%
def supervisor_node(state: CustomState) -> CustomState:
    # Append the new message to the messages list
    state['messages'].append({"content": "Use this conversation to determine whether to change/fix code in the provided file path with FIX or run some other command in the shell with BASH.", "type": "human"})

    # Invoke the supervisor chain
    supervisor_response = supervisor_chain.invoke({
        "file_in_focus": state['file_in_focus'],
        "error_list": state['error_list'],
        "messages": state['messages'],
        "root_dir": state['root_dir']
    })

    # Append the supervisor's response to the messages list
    state['messages'].append({"content": str(supervisor_response.message), "type": "supervisor"})

    # Return the updated state
    return {
        'messages': state['messages'],
        'next': supervisor_response.next,
        'file_in_focus': state['file_in_focus'],
        'root_dir': state['root_dir'],
        'error_list': state['error_list'],
        'iterations': state['iterations'] + 1
    }

#%%
workflow = StateGraph(CustomState)

#%%
workflow.add_node("supervisor", supervisor_node)
workflow.add_edge("__start__", "supervisor")
workflow.add_edge("supervisor", "__end__")

#%%
app = workflow.compile()

#%%
result = app.invoke({'messages': [{"content": "Try to build the code in the root directory.", "type": "human"}],
        'next': "",
        'file_in_focus': "",
        'root_dir': "/path/to/your/project",
        'error_list': [],
        'iterations': 0})

With this new updated code too, I am getting the same error

[dosubot[bot]]

To resolve the type error "Expected mapping type as input to PromptTemplate. Received <class 'langchain_core.messages.ai.AIMessage>", ensure that the messages input to the supervisor_chain.invoke method is a list of dictionaries, not AIMessage objects. Your current approach seems correct, but let's ensure that the supervisor_chain is correctly set up to handle the input.

Here is a refined version of your code with additional checks:

1. Ensure that the supervisor_chain is correctly set up to handle the input.
2. Ensure that the messages are correctly formatted as dictionaries.

Updated Code

from src.configs import set_environ, api_key
set_environ()

#%%

from typing import TypedDict, List, Annotated, Dict, Any
from langgraph.graph import StateGraph
import operator

class CustomState(TypedDict):
    messages: Annotated[List[Dict[str, Any]], operator.add]
    next: str
    file_in_focus: str
    root_dir: str
    error_list: List[str]
    iterations: int

#%%

from langchain_community.tools import ShellTool
from langchain_community.agent_toolkits import FileManagementToolkit
file_manager = FileManagementToolkit(
    root_dir='/path/to/your/project'
).get_tools()
shell_tool = ShellTool()
shell_tool.description = shell_tool.description + f"args {shell_tool.args}".replace(
    "{", "{{"
).replace("}", "}}")

#%%

from langchain_openai import ChatOpenAI
model = deepseek_coder_beta_agent = ChatOpenAI(
        model='deepseek-coder',
        openai_api_key=api_key,
        openai_api_base='https://api.yourapi.com/beta',
        temperature=1.0,
        max_tokens=8192,
        timeout=None,
        max_retries=2,
        api_key=api_key
    )

#%%

from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import PydanticOutputParser

class SupervisorResponse(BaseModel):
    message: str = Field(description="The message/task to give to the next agent")
    next: str = Field(description="enum out of FIX, BASH, or END")

#%%

supervisor_parser = PydanticOutputParser(pydantic_object=SupervisorResponse)
supervisor_prompt = PromptTemplate(
    template=("Answer the following.\n{format_instructions}\n After analyzing the received information, "
              "choose the message/task to give to the next agent and what agent to give it to. "
              "Use FIX for code fixer agent, BASH to make the shell agent run more commands again, and END to end the loop.\n"
              "This is the conversation until now: {messages}\n These are the errors: {error_list}\n"
              "and the file path needed according to bash is: {file_in_focus}.\n"
              "This is the root directory holding the application: {root_dir}"),
    input_variables=["error_list", "messages", "file_in_focus", "root_dir"],
    partial_variables={"format_instructions": supervisor_parser.get_format_instructions()}
)

#%%

supervisor_chain = supervisor_prompt | model | supervisor_parser

#%%

def supervisor_node(state: CustomState) -> CustomState:
    # Append the new message to the messages list
    state['messages'].append({"content": "Use this conversation to determine whether to change/fix code in the provided file path with FIX or run some other command in the shell with BASH.", "type": "human"})

    # Ensure messages are dictionaries
    messages_dict = [{"content": msg["content"], "type": msg["type"]} for msg in state['messages']]

    # Invoke the supervisor chain
    supervisor_response = supervisor_chain.invoke({
        "file_in_focus": state['file_in_focus'],
        "error_list": state['error_list'],
        "messages": messages_dict,
        "root_dir": state['root_dir']
    })

    # Append the supervisor's response to the messages list
    state['messages'].append({"content": supervisor_response["message"], "type": "supervisor"})

    # Return the updated state
    return {
        'messages': state['messages'],
        'next': supervisor_response["next"],
        'file_in_focus': state['file_in_focus'],
        'root_dir': state['root_dir'],
        'error_list': state['error_list'],
        'iterations': state['iterations'] + 1
    }

#%%

workflow = StateGraph(CustomState)

#%%

workflow.add_node("supervisor", supervisor_node)
workflow.add_edge("__start__", "supervisor")
workflow.add_edge("supervisor", "__end__")

#%%

app = workflow.compile()

#%%

result = app.invoke({
    'messages': [{"content": "Try to build the code in the root directory.", "type": "human"}],
    'next': "",
    'file_in_focus': "",
    'root_dir': "/path/to/your/project",
    'error_list': [],
    'iterations': 0
})

Ensure that the supervisor_chain is correctly set up to handle the input and that the messages are correctly formatted as dictionaries. This should resolve the type error you are encountering ^[1][2].

To continue talking to Dosu, mention @dosu.

[intern162]

Fixed, I had a typo in there. Sorry