Chat loop inside graph?

[hardik-vala]

I'm implementing a multi-agent network where some of the agents might have to chat with the user for more information inorder to perform their function. Is it possible to allow an LLM and user to converse back and forth inside a graph?

The Prompt Generation from User Requirements tutorial has an example of a single agent using a dialogue with the user to gather more information, but the chat loop is implemented outside the graph, ie,

while True:
    try:
        user = input("User (q/Q to quit): ")
    except:
        user = cached_human_responses[cached_response_index]
        cached_response_index += 1
    print(f"User (q/Q to quit): {user}")
    if user in {"q", "Q"}:
        print("AI: Byebye")
        break
    output = None
    for output in graph.stream(
        {"messages": [HumanMessage(content=user)]}, config=config, stream_mode="updates"
    ):
        last_message = next(iter(output.values()))["messages"][-1]
        last_message.pretty_print()

I'm looking for a way to give some agents in my graph the ability to have conversations with the user without exiting the graph.

[hwchase17]

yup, you can use a human-in-the-loop setup like: https://langchain-ai.github.io/langgraph/how-tos/human_in_the_loop/wait-user-input/#agent

[gfortaine]

here is an example implementation of Prompt Generation from User Requirements example with up-to-date human-in-the-loop support (interrupt and Command) :

from typing import List, Literal

from pydantic import BaseModel

from langchain_core.messages import SystemMessage
from langchain_openai import ChatOpenAI
from langchain_core.messages import AIMessage, ToolMessage
from langgraph.graph import MessagesState, StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command, interrupt


class PromptInstructions(BaseModel):
    """Instructions on how to prompt the LLM."""

    objective: str
    variables: List[str]
    constraints: List[str]
    requirements: List[str]


template = """Your job is to get information from a user about what type of prompt template they want to create.

You should get the following information from them:

- What the objective of the prompt is
- What variables will be passed into the prompt template
- Any constraints for what the output should NOT do
- Any requirements that the output MUST adhere to

If you are not able to discern this info, ask them to clarify! Do not attempt to wildly guess.

After you are able to discern all the information, call the relevant tool."""


def get_messages_info(messages):
    return [SystemMessage(content=template)] + messages


llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tool = llm.bind_tools([PromptInstructions])


# https://langchain-ai.github.io/langgraph/how-tos/human_in_the_loop/wait-user-input/#agent
# Define the function that determines whether to continue or not
def get_state(last_message):
    if not last_message.tool_calls:
        return "human_info"
    tool_name = last_message.tool_calls[0]["name"]
    match tool_name:
        case "PromptInstructions":
            return "add_tool_message"
        case _:
            return END


def info_chain(state) -> Command[Literal["add_tool_message", "human_info", END]]:
    messages = get_messages_info(state["messages"])
    response = llm_with_tool.invoke(messages)
    goto = get_state(response)

    return Command(
        goto=goto,
        update={"messages": [response]},
    )


def human_info_feedback(state: MessagesState) -> Command[Literal["info"]]:
    """A node for collecting user input."""
    print("Waiting for user input...")
    user_input = interrupt(value="Ready for user input.")

    print("user input:", user_input)

    return Command(
        goto="info",
        update={
            "messages": [
                {
                    "role": "human",
                    "content": user_input,
                }
            ]
        },
    )


# New system prompt
prompt_system = """Based on the following requirements, write a good prompt template:

{reqs}"""


# Function to get the messages for the prompt
# Will only get messages AFTER the tool call
def get_prompt_messages(messages: list):
    tool_call = None
    other_msgs = []
    for m in messages:
        if isinstance(m, AIMessage) and m.tool_calls:
            tool_call = m.tool_calls[0]["args"]
        elif isinstance(m, ToolMessage):
            continue
        elif tool_call is not None:
            other_msgs.append(m)
    return [SystemMessage(content=prompt_system.format(reqs=tool_call))] + other_msgs


def prompt_gen_chain(state) -> Command[Literal[END]]:
    messages = get_prompt_messages(state["messages"])
    response = llm.invoke(messages)

    return Command(
        goto=END,
        update={"messages": [response]},
    )


memory = MemorySaver()
workflow = StateGraph(MessagesState)
workflow.add_node("info", info_chain)
workflow.add_node("human_info", human_info_feedback)
workflow.add_node("prompt", prompt_gen_chain)


@workflow.add_node
def add_tool_message(state: MessagesState) -> Command[Literal["prompt"]]:
    return Command(
        goto="prompt",
        update={
            "messages": [
                ToolMessage(
                    content="Prompt generated!",
                    tool_call_id=state["messages"][-1].tool_calls[0]["id"],
                )
            ]
        },
    )


workflow.set_entry_point("info")

graph = workflow.compile(checkpointer=memory)

[estherfc00]

Hello, thanks for sharing. Is this available in production? For example, when the user makes a query using streamlit or gradio, how does the "interrupt" message come into the frontend? thanks!

[jomyhuang]

Add a question, may I ask：
def human_info_feedback(state: MessagesState) -> Command[Literal["info"]]:
"""A node for collecting user input."""
print("Waiting for user input...")
user_input = interrupt(value="Ready for user input.")

print("user input:", user_input)

return Command(
    goto="info",
    update={
        "messages": [
            {
                "role": "human",
                "content": user_input,
            }
        ]
    },
)

state = invoke(....)
how to get value(text "Ready for user input." or dict) after interrupt from "human_info_feedback"?

print value or something..

thanks!