langchain-ai/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-07-15 12:35:43 -04:00
Code Issues 314 Packages Projects Releases 17 Wiki Activity

Execution goes into infinite loop #392

New Issue
Closed
opened 2026-02-20 17:39:53 -05:00 by yindo · 1 comment
yindo commented 2026-02-20 17:39:53 -05:00
Owner
Copy Link
Copy Source

Originally created by @asrays on GitHub (Jan 9, 2025).

Checked other resources

  • This is a bug, not a usage question. For questions, please use GitHub Discussions.
  • I added a clear and detailed title that summarizes the issue.
  • I read what a minimal reproducible example is (https://stackoverflow.com/help/minimal-reproducible-example).
  • I included a self-contained, minimal example that demonstrates the issue INCLUDING all the relevant imports. The code run AS IS to reproduce the issue.

Example Code

import os
os.environ['ANTHROPIC_API_KEY'] = 'key'

from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.messages import SystemMessage
from typing import List, Callable
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain import hub
from typing import Annotated
from langgraph.graph.message import add_messages
import requests, json
from langgraph.types import Command
from langgraph.prebuilt import create_react_agent
from langchain_core.messages import HumanMessage, AIMessage
from typing import Literal, TypedDict, Union

llm = ChatAnthropic(model="claude-3-5-haiku-20241022")

# Define graph state
class AgentState(MessagesState):
    messages: Annotated[list, add_messages]


def push_into_db(brand_payload):

    headers = {'Content-Type': 'application/json'}
    
    response = requests.request('POST', 'my_api', data=json.dumps(brand_payload), headers=headers, timeout=180)
    response_content = json.loads(response.content)
    
    return response_content

@tool
def insert_brand_tool(brand_name: str):
    '''This will insert the brand data into database and return object id'''
    try:
        brand_payload = {"name": brand_name, "journey_type": "social"}
        response = push_into_db(brand_payload)

        if response["success"] == True:
            brandId = response["data"]["brandId"]
            
            return f"Brand name inserted with object id {brandId}"
        else:
            return "Could not insert brand into database.Please try again."
            
    except Exception as e:
        return "Could not insert brand into database.Please try again."

@tool
def get_image(img_path: str):
    '''This will read the given web url image path and download it'''
    try:
        img_response = requests.get(img_path)

        img_name = os.path.basename(img_path)    
        with open(img_name, "wb") as f:
            f.write(img_response.content)
        
        return f"Image has been downloaded, status is successfull. No worker needed."
    except Exception as e:
        return "Cannot read image."

members = ["INSERT_DATABASE_AGENT", "UPLOAD_IMAGE_AGENT", "FINISH"]

class Router(TypedDict):
    """Worker to route to next. If no workers needed, route to FINISH."""

    next: Union[Literal["INSERT_DATABASE_AGENT"], Literal["UPLOAD_IMAGE_AGENT"], Literal["FINISH"]]

system_prompt = (
    "You are a supervisor tasked with managing a conversation between the"
    f" following workers: {members}. Given the following user request,"
    " respond with the worker to act next. Each worker will perform a"
    " task and respond with their results and status. WHEN SUCCESSFULL, RESPOND MESSAGE WITH FINISH"
)

def greet_node(state: AgentState):
    messages = [
        {"role": "system", "content": system_prompt},
    ] + [state["messages"][-1]]
    response = llm.with_structured_output(Router).invoke(messages)
    
    goto = response["next"]
    if goto == "FINISH":
        goto = END

    return Command(goto="UPLOAD_IMAGE_AGENT")

insert_database_agent = create_react_agent(llm, tools=[insert_brand_tool])

def router_function_to_db(state: MessagesState):
    result = insert_database_agent.invoke(state)
    return Command(
        update={
            "messages": [
                HumanMessage(content=result["messages"][-1].content, name="INSERT_DATABASE_AGENT")
            ]
        },
        goto="GREETING_AGENT",
    )
    
upload_image_agent = create_react_agent(llm, tools=[get_image], state_modifier="Response with FINISH if image is downloaded successfully.")

def router_function_to_image_upload(state: MessagesState):
    result = upload_image_agent.invoke(state)
    return Command(
        update={
            "messages": [
                HumanMessage(content=result["messages"][-1].content, name="UPLOAD_IMAGE_AGENT")
            ]
        },
        goto="GREETING_AGENT",
    )

builder = StateGraph(MessagesState)

builder.add_edge(START, "GREETING_AGENT")
builder.add_node("GREETING_AGENT", greet_node)
builder.add_node("INSERT_DATABASE_AGENT", router_function_to_db)
builder.add_node("UPLOAD_IMAGE_AGENT", router_function_to_image_upload)

graph = builder.compile()

img_bytes = graph.get_graph().draw_mermaid_png()

with open("graph_mermaid.png", "wb") as f:
     f.write(img_bytes)
    

input={"messages": ["Hi how are you? Can you read the image url(A valid s3 url)"]}

for output in graph.stream(input):
    print(output)

Error Message and Stack Trace (if applicable)

No response

Description

This flow is the improvisation from the langgraph multi agent supervisor docs.
This workflow could not reach to "FINISH" state (only 1 out of 20 time if I run the code). Please note everything else is working fine; only the issue is when i give the input- download the image, it gets downloaded by "upload_image_agent" but my "greeting_agent" could not able to generate "FINISH" respose in its stack so it is not able to exit. What best should be the best promt to get into "FINISH" state?

Please can anyone help?

System Info

versions-
langgraph- 0.2.61
langgraph-checkpoint- 2.0.9
langgraph-sdk- 0.1.48

Originally created by @asrays on GitHub (Jan 9, 2025). ### Checked other resources - [X] This is a bug, not a usage question. For questions, please use GitHub Discussions. - [X] I added a clear and detailed title that summarizes the issue. - [X] I read what a minimal reproducible example is (https://stackoverflow.com/help/minimal-reproducible-example). - [X] I included a self-contained, minimal example that demonstrates the issue INCLUDING all the relevant imports. The code run AS IS to reproduce the issue. ### Example Code ```python import os os.environ['ANTHROPIC_API_KEY'] = 'key' from langchain_anthropic import ChatAnthropic from langchain_core.tools import tool from langgraph.graph import StateGraph, MessagesState, START, END from langgraph.prebuilt import ToolNode, tools_condition from langchain_core.messages import SystemMessage from typing import List, Callable from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain import hub from typing import Annotated from langgraph.graph.message import add_messages import requests, json from langgraph.types import Command from langgraph.prebuilt import create_react_agent from langchain_core.messages import HumanMessage, AIMessage from typing import Literal, TypedDict, Union llm = ChatAnthropic(model="claude-3-5-haiku-20241022") # Define graph state class AgentState(MessagesState): messages: Annotated[list, add_messages] def push_into_db(brand_payload): headers = {'Content-Type': 'application/json'} response = requests.request('POST', 'my_api', data=json.dumps(brand_payload), headers=headers, timeout=180) response_content = json.loads(response.content) return response_content @tool def insert_brand_tool(brand_name: str): '''This will insert the brand data into database and return object id''' try: brand_payload = {"name": brand_name, "journey_type": "social"} response = push_into_db(brand_payload) if response["success"] == True: brandId = response["data"]["brandId"] return f"Brand name inserted with object id {brandId}" else: return "Could not insert brand into database.Please try again." except Exception as e: return "Could not insert brand into database.Please try again." @tool def get_image(img_path: str): '''This will read the given web url image path and download it''' try: img_response = requests.get(img_path) img_name = os.path.basename(img_path) with open(img_name, "wb") as f: f.write(img_response.content) return f"Image has been downloaded, status is successfull. No worker needed." except Exception as e: return "Cannot read image." members = ["INSERT_DATABASE_AGENT", "UPLOAD_IMAGE_AGENT", "FINISH"] class Router(TypedDict): """Worker to route to next. If no workers needed, route to FINISH.""" next: Union[Literal["INSERT_DATABASE_AGENT"], Literal["UPLOAD_IMAGE_AGENT"], Literal["FINISH"]] system_prompt = ( "You are a supervisor tasked with managing a conversation between the" f" following workers: {members}. Given the following user request," " respond with the worker to act next. Each worker will perform a" " task and respond with their results and status. WHEN SUCCESSFULL, RESPOND MESSAGE WITH FINISH" ) def greet_node(state: AgentState): messages = [ {"role": "system", "content": system_prompt}, ] + [state["messages"][-1]] response = llm.with_structured_output(Router).invoke(messages) goto = response["next"] if goto == "FINISH": goto = END return Command(goto="UPLOAD_IMAGE_AGENT") insert_database_agent = create_react_agent(llm, tools=[insert_brand_tool]) def router_function_to_db(state: MessagesState): result = insert_database_agent.invoke(state) return Command( update={ "messages": [ HumanMessage(content=result["messages"][-1].content, name="INSERT_DATABASE_AGENT") ] }, goto="GREETING_AGENT", ) upload_image_agent = create_react_agent(llm, tools=[get_image], state_modifier="Response with FINISH if image is downloaded successfully.") def router_function_to_image_upload(state: MessagesState): result = upload_image_agent.invoke(state) return Command( update={ "messages": [ HumanMessage(content=result["messages"][-1].content, name="UPLOAD_IMAGE_AGENT") ] }, goto="GREETING_AGENT", ) builder = StateGraph(MessagesState) builder.add_edge(START, "GREETING_AGENT") builder.add_node("GREETING_AGENT", greet_node) builder.add_node("INSERT_DATABASE_AGENT", router_function_to_db) builder.add_node("UPLOAD_IMAGE_AGENT", router_function_to_image_upload) graph = builder.compile() img_bytes = graph.get_graph().draw_mermaid_png() with open("graph_mermaid.png", "wb") as f: f.write(img_bytes) input={"messages": ["Hi how are you? Can you read the image url(A valid s3 url)"]} for output in graph.stream(input): print(output) ``` ### Error Message and Stack Trace (if applicable) _No response_ ### Description This flow is the improvisation from the langgraph multi agent supervisor docs. This workflow could not reach to "FINISH" state (only 1 out of 20 time if I run the code). Please note everything else is working fine; only the issue is when i give the input- download the image, it gets downloaded by "upload_image_agent" but my "greeting_agent" could not able to generate "FINISH" respose in its stack so it is not able to exit. What best should be the best promt to get into "FINISH" state? Please can anyone help? ### System Info versions- langgraph- 0.2.61 langgraph-checkpoint- 2.0.9 langgraph-sdk- 0.1.48
yindo closed this issue 2026-02-20 17:39:53 -05:00
yindo commented 2026-02-20 17:39:53 -05:00
Author
Owner
Copy Link
Copy Source

@vbarda commented on GitHub (Jan 9, 2025):

could you try changing the router to use a single literal instead of unions, e.g.

class Router(TypedDict):
    """Worker to route to next. If no workers needed, route to FINISH."""

    next: Literal["INSERT_DATABASE_AGENT", "UPLOAD_IMAGE_AGENT", "FINISH"]

this should make it easier for the model to produce structured output. also, you might also want to modify the prompt to something like

WHEN SUCCESSFULL, RESPOND MESSAGE WITH FINISH to when all workers are finished and the task is completed, respond with "FINISH"

will also turn this into a discussion, since it's not an issue with LangGraph framework

@vbarda commented on GitHub (Jan 9, 2025): could you try changing the router to use a single literal instead of unions, e.g. ```python class Router(TypedDict): """Worker to route to next. If no workers needed, route to FINISH.""" next: Literal["INSERT_DATABASE_AGENT", "UPLOAD_IMAGE_AGENT", "FINISH"] ``` this should make it easier for the model to produce structured output. also, you might also want to modify the prompt to something like `WHEN SUCCESSFULL, RESPOND MESSAGE WITH FINISH` to `when all workers are finished and the task is completed, respond with "FINISH"` will also turn this into a discussion, since it's not an issue with LangGraph framework
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
agents
awaiting author revision
bug
change
chat models
checkpointer
ci
code quality
code quality
dependencies
documentation
drawing
duplicate
enhancement
good first issue
help wanted
invalid
maintainer
meta
missing information
not maintained by langchain
open-swe-auto
open-swe-max
open-swe-max-auto
pending
performance
pull-request
Mirrored from GitHub Pull Request
 question
requires design
sdk
server
stale
studio
tests
type checking
usage question
v0.5
v1.1
No Label
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: langchain-ai/langgraph#392
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?