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

python REPL tool not working the for code execution #275

New Issue
Closed
opened 2026-02-20 17:35:02 -05:00 by yindo · 12 comments
yindo commented 2026-02-20 17:35:02 -05:00
Owner
Copy Link
Copy Source

Originally created by @pradeepdev-1995 on GitHub (Oct 10, 2024).

Checked other resources

  • I added a very descriptive title to this issue.
  • I searched the LangGraph/LangChain documentation with the integrated search.
  • I used the GitHub search to find a similar question and didn't find it.
  • I am sure that this is a bug in LangGraph/LangChain rather than my code.
  • I am sure this is better as an issue rather than a GitHub discussion, since this is a LangGraph bug and not a design question.

Example Code

This is the requirement.

How to run/execute Python codes using the pythonRepl tool?
The tool should be one node in the graph.
The just above node creates the Python code and pass to the node that contains the pythonRepl tool
code I tried

from langchain_experimental.utilities import PythonREPL
from langchain_core.tools import tool
repl = PythonREPL()

@tool
def python_repl(code):
    """
    Use this to execute python code without any kind of error.

    """
    try:
        print("Result of code execution here")
        result = repl.run(code)
        print("Result of code execution",result)
    except BaseException as e:
        return f"Failed to execute. Error :  {repr(e)}"

    return f"{result}"


def codeExecution(cls, state: State) -> State:
        codeToExecute = state['codeToExecute']
        print("codeToExecute")
        print(codeToExecute)
        llm_with_tools = llm.bind_tools([python_repl])
        messages = [
            SystemMessage(
                content=""" You have got the task to execute code. Use the python_repl tool to execute it. I will a message and your task is to detect if it was successfully run or produced an error.
                    If the code produced an error just return "True". If it was successfully executed, return "False" """
            ),
            HumanMessage(content=codeToExecute),
        ]
        ai_msg = llm_with_tools.invoke(messages)
        messages.append(ai_msg)
        for tool_call in ai_msg.tool_calls:
            selected_tool = {"python_repl": python_repl}[tool_call["name"].lower()]
            tool_output = selected_tool.invoke(tool_call["args"])
            state["error_message"] = tool_output
            messages.append(ToolMessage(tool_output, tool_call_id=tool_call["id"]))

        result = llm_with_tools.invoke(messages)
        state["error"] = result.content
        return state

Error Message and Stack Trace (if applicable)

File "<path>.local/lib/python3.9/site-packages/streamlit/runtime/scriptrunner/exec_code.py", line 88, in exec_func_with_error_handling
    result = func()
  File "<path>.local/lib/python3.9/site-packages/streamlit/runtime/scriptrunner/script_runner.py", line 579, in code_to_exec
    exec(code, module.__dict__)
  File "<path>file", line 85, in <module>
    executeGraph(taskSpecificGraph,query,st.session_state.datasetColumnNames,st.session_state.datasetPath)
  File "<path>file.py", line 368, in executeGraph
    graphExecutionResult = taskSpecificGraph.invoke({"query": query,"datasetColumnNames":datasetColumnNames,"datasetPath":datasetPath,"iterations": 1})
  File "<path>.local/lib/python3.9/site-packages/langgraph/pregel/__init__.py", line 1263, in invoke
    for chunk in self.stream(
  File "<path>.local/lib/python3.9/site-packages/langgraph/pregel/__init__.py", line 948, in stream
    _panic_or_proceed(done, inflight, loop.step)
  File "<path>.local/lib/python3.9/site-packages/langgraph/pregel/__init__.py", line 1349, in _panic_or_proceed
    raise exc
  File "<path>.local/lib/python3.9/site-packages/langgraph/pregel/executor.py", line 60, in done
    task.result()
  File "/usr/lib/python3.9/concurrent/futures/_base.py", line 439, in result
    return self.__get_result()
  File "/usr/lib/python3.9/concurrent/futures/_base.py", line 391, in __get_result
    raise self._exception
  File "/usr/lib/python3.9/concurrent/futures/thread.py", line 58, in run
    result = self.fn(*self.args, **self.kwargs)
  File "<path>.local/lib/python3.9/site-packages/langgraph/pregel/retry.py", line 25, in run_with_retry
    task.proc.invoke(task.input, task.config)
  File "<path>.local/lib/python3.9/site-packages/langchain_core/runnables/base.py", line 2878, in invoke
    input = context.run(step.invoke, input, config)
  File "<path>.local/lib/python3.9/site-packages/langgraph/utils.py", line 102, in invoke
    ret = context.run(self.func, input, **kwargs)
  File "<path>.local/lib/python3.9/site-packages/langgraph/graph/graph.py", line 82, in _route
    return self._finish(writer, input, result)
  File "<path>.local/lib/python3.9/site-packages/langgraph/graph/graph.py", line 109, in _finish
    destinations = [r if isinstance(r, Send) else self.ends[r] for r in result]
  File "<path>.local/lib/python3.9/site-packages/langgraph/graph/graph.py", line 109, in <listcomp>
    destinations = [r if isinstance(r, Send) else self.ends[r] for r in result]
KeyError: None



### Description

How to run/execute Python codes using the pythonRepl tool?
The tool should be one node in the graph.
The just above node creates the Python code and pass to the node that contains the pythonRepl tool
If any error occurs, we should get the error message also from the tool

### System Info

ubuntu OS
langchain==0.2.16
langchain-community==0.2.16
langchain-core==0.2.39
langchain-experimental==0.0.64
langgraph==0.1.14
langgraph-checkpoint==1.0.11
Originally created by @pradeepdev-1995 on GitHub (Oct 10, 2024). ### Checked other resources - [X] I added a very descriptive title to this issue. - [X] I searched the [LangGraph](https://langchain-ai.github.io/langgraph/)/LangChain documentation with the integrated search. - [X] I used the GitHub search to find a similar question and didn't find it. - [X] I am sure that this is a bug in LangGraph/LangChain rather than my code. - [X] I am sure this is better as an issue [rather than a GitHub discussion](https://github.com/langchain-ai/langgraph/discussions/new/choose), since this is a LangGraph bug and not a design question. ### Example Code This is the requirement. How to run/execute Python codes using the pythonRepl tool? The tool should be one node in the graph. The just above node creates the Python code and pass to the node that contains the pythonRepl tool code I tried ``` from langchain_experimental.utilities import PythonREPL from langchain_core.tools import tool repl = PythonREPL() @tool def python_repl(code): """ Use this to execute python code without any kind of error. """ try: print("Result of code execution here") result = repl.run(code) print("Result of code execution",result) except BaseException as e: return f"Failed to execute. Error : {repr(e)}" return f"{result}" def codeExecution(cls, state: State) -> State: codeToExecute = state['codeToExecute'] print("codeToExecute") print(codeToExecute) llm_with_tools = llm.bind_tools([python_repl]) messages = [ SystemMessage( content=""" You have got the task to execute code. Use the python_repl tool to execute it. I will a message and your task is to detect if it was successfully run or produced an error. If the code produced an error just return "True". If it was successfully executed, return "False" """ ), HumanMessage(content=codeToExecute), ] ai_msg = llm_with_tools.invoke(messages) messages.append(ai_msg) for tool_call in ai_msg.tool_calls: selected_tool = {"python_repl": python_repl}[tool_call["name"].lower()] tool_output = selected_tool.invoke(tool_call["args"]) state["error_message"] = tool_output messages.append(ToolMessage(tool_output, tool_call_id=tool_call["id"])) result = llm_with_tools.invoke(messages) state["error"] = result.content return state ``` ``` ``` ### Error Message and Stack Trace (if applicable) ```shell File "<path>.local/lib/python3.9/site-packages/streamlit/runtime/scriptrunner/exec_code.py", line 88, in exec_func_with_error_handling result = func() File "<path>.local/lib/python3.9/site-packages/streamlit/runtime/scriptrunner/script_runner.py", line 579, in code_to_exec exec(code, module.__dict__) File "<path>file", line 85, in <module> executeGraph(taskSpecificGraph,query,st.session_state.datasetColumnNames,st.session_state.datasetPath) File "<path>file.py", line 368, in executeGraph graphExecutionResult = taskSpecificGraph.invoke({"query": query,"datasetColumnNames":datasetColumnNames,"datasetPath":datasetPath,"iterations": 1}) File "<path>.local/lib/python3.9/site-packages/langgraph/pregel/__init__.py", line 1263, in invoke for chunk in self.stream( File "<path>.local/lib/python3.9/site-packages/langgraph/pregel/__init__.py", line 948, in stream _panic_or_proceed(done, inflight, loop.step) File "<path>.local/lib/python3.9/site-packages/langgraph/pregel/__init__.py", line 1349, in _panic_or_proceed raise exc File "<path>.local/lib/python3.9/site-packages/langgraph/pregel/executor.py", line 60, in done task.result() File "/usr/lib/python3.9/concurrent/futures/_base.py", line 439, in result return self.__get_result() File "/usr/lib/python3.9/concurrent/futures/_base.py", line 391, in __get_result raise self._exception File "/usr/lib/python3.9/concurrent/futures/thread.py", line 58, in run result = self.fn(*self.args, **self.kwargs) File "<path>.local/lib/python3.9/site-packages/langgraph/pregel/retry.py", line 25, in run_with_retry task.proc.invoke(task.input, task.config) File "<path>.local/lib/python3.9/site-packages/langchain_core/runnables/base.py", line 2878, in invoke input = context.run(step.invoke, input, config) File "<path>.local/lib/python3.9/site-packages/langgraph/utils.py", line 102, in invoke ret = context.run(self.func, input, **kwargs) File "<path>.local/lib/python3.9/site-packages/langgraph/graph/graph.py", line 82, in _route return self._finish(writer, input, result) File "<path>.local/lib/python3.9/site-packages/langgraph/graph/graph.py", line 109, in _finish destinations = [r if isinstance(r, Send) else self.ends[r] for r in result] File "<path>.local/lib/python3.9/site-packages/langgraph/graph/graph.py", line 109, in <listcomp> destinations = [r if isinstance(r, Send) else self.ends[r] for r in result] KeyError: None ``` ``` ### Description How to run/execute Python codes using the pythonRepl tool? The tool should be one node in the graph. The just above node creates the Python code and pass to the node that contains the pythonRepl tool If any error occurs, we should get the error message also from the tool ### System Info ubuntu OS langchain==0.2.16 langchain-community==0.2.16 langchain-core==0.2.39 langchain-experimental==0.0.64 langgraph==0.1.14 langgraph-checkpoint==1.0.11
yindo closed this issue 2026-02-20 17:35:02 -05:00
yindo commented 2026-02-20 17:35:02 -05:00
Author
Owner
Copy Link
Copy Source

@gbaian10 commented on GitHub (Oct 10, 2024):

It is recommended to first upgrade the package.
And use the prebuilt ToolNode (from langgraph.prebuilt import ToolNode) to handle the ToolMessage part.

@gbaian10 commented on GitHub (Oct 10, 2024): It is recommended to first upgrade the package. And use the prebuilt `ToolNode` (`from langgraph.prebuilt import ToolNode`) to handle the `ToolMessage` part.
yindo commented 2026-02-20 17:35:02 -05:00
Author
Owner
Copy Link
Copy Source

@pradeepdev-1995 commented on GitHub (Oct 10, 2024):

@gbaian10 upgrade to which version? the latest langgraph versions are seems unstable

@pradeepdev-1995 commented on GitHub (Oct 10, 2024): @gbaian10 upgrade to which version? the latest langgraph versions are seems unstable
yindo commented 2026-02-20 17:35:02 -05:00
Author
Owner
Copy Link
Copy Source

@gbaian10 commented on GitHub (Oct 10, 2024):

@gbaian10 upgrade to which version? the latest langgraph versions are seems unstable

Version 0.2.x has been out for a while and should be stable.
Additionally, langchain_core has reached version 0.3.x.

@gbaian10 commented on GitHub (Oct 10, 2024): > @gbaian10 upgrade to which version? the latest langgraph versions are seems unstable Version 0.2.x has been out for a while and should be stable. Additionally, `langchain_core` has reached version 0.3.x.
yindo commented 2026-02-20 17:35:02 -05:00
Author
Owner
Copy Link
Copy Source

@pradeepdev-1995 commented on GitHub (Oct 10, 2024):

any additional documentation for repl tool integration with langgraph?
Now getting the below error

openai.BadRequestError: Error code: 400 - {'error': {'message': "An assistant message with 'tool_calls' must be followed by tool messages responding to each 'tool_call_id'. The following tool_call_ids did not have response messages: call_jP44yo4aHaQfhC3VAeSd4GOj", 'type': 'invalid_request_error', 'param': 'messages', 'code': None}}
@pradeepdev-1995 commented on GitHub (Oct 10, 2024): any additional documentation for repl tool integration with langgraph? Now getting the below error ``` openai.BadRequestError: Error code: 400 - {'error': {'message': "An assistant message with 'tool_calls' must be followed by tool messages responding to each 'tool_call_id'. The following tool_call_ids did not have response messages: call_jP44yo4aHaQfhC3VAeSd4GOj", 'type': 'invalid_request_error', 'param': 'messages', 'code': None}} ```
yindo commented 2026-02-20 17:35:02 -05:00
Author
Owner
Copy Link
Copy Source

@pradeepdev-1995 commented on GitHub (Oct 10, 2024):

Here is the code i tried

from langchain_core.messages import HumanMessage, SystemMessage,ToolMessage
def read_code_from_file(file_path: str) -> str:
    with open(file_path, "r") as file:
        code = file.read()
    return code


def write_code_to_file(file_path: str, code: str):
    with open(file_path, "w") as file:
        file.write(code)


from typing import Annotated
from langchain_experimental.utilities import PythonREPL
from langchain_core.tools import tool

repl = PythonREPL()


@tool
def python_repl(code: Annotated[str, "filename to read the code from"]):
    """Use this to execute python code read from a file. If you want to see the output of a value,
    Make sure that you read the code from correctly
    you should print it out with `print(...)`. This is visible to the user."""

    try:
        result = repl.run(code)
        print("RESULT CODE EXECUTION:", result)
    except BaseException as e:
        return f"Failed to execute. Error: {repr(e)}"
    return f"Executed:\n```python\n{code}\n```\nStdout: {result}"

from langchain_openai import AzureChatOpenAI
model = AzureChatOpenAI(
)
model_with_tools = model.bind_tools([python_repl])



from typing import TypedDict


class AgentState(TypedDict):
    error: bool
    error_message: str
    file_path: str
    code: str
    iterations: int


def identify_filepath(state: AgentState):
    state["file_path"] = 'testscript.py'
    return state


def execute_code_with_model(state: AgentState):
    code = read_code_from_file(state["file_path"])
    print("code iss")
    print(code)
    model_with_tools = model.bind_tools([python_repl])

    messages = [
        SystemMessage(
            content=""" You have got the task to execute code. Use the python_repl tool to execute it. I will give the message and your task is to detect if it was successfully run or produced an error.
            If the code produced an error just return 'True'. If it was sucessfully executed, return 'False'"""
        ),
        HumanMessage(content=code),
    ]

    ai_msg = model_with_tools.invoke(messages)
    messages.append(ai_msg)
    for tool_call in ai_msg.tool_calls:
        print("tool_call")
        print(tool_call)
        selected_tool = {"python_repl": python_repl}[tool_call["name"].lower()]
        tool_output = selected_tool.invoke(tool_call["args"])
        state["error_message"] = tool_output
        messages.append(ToolMessage(tool_output, tool_call_id=tool_call["id"]))

    result = model_with_tools.invoke(messages)
    print("EVALUATION RESULT:", result)
    state["error"] = result.content
    return state


def rewrite_code(state: AgentState):

    code = state["code"]
    error = state["error_message"]
    state["iterations"] += 1
    messages = [
        SystemMessage(
            content="You can to analyze the following code and error provided in the usermessage. Your task is to fix that code and provide the user the correct new code. VERY IMPORTANT: ONLY RETURN THE UPDATED CODE, NOTHING ELSE! Dont use a markdown style, just the code as Text"
        ),
        HumanMessage(content=f"Code: {code} | Error: {error}"),
    ]
    ai_msg = model.invoke(messages)
    print("NEW SUGGESTED CODE:", ai_msg.content)
    write_code_to_file(file_path=f'{state["file_path"]}', code=ai_msg.content)
    state["code"] = ai_msg.content
    return state


def next_step(state: AgentState):
    if state["iterations"] > 3:
        print("Max Iterations done.... Exit Agent")
        return "max_iterations"
    if state["error"] == "True":
        print(f"Error in {state['file_path']}. {state['iterations']} tries done")
        return "error"
    if state["error"] == "False":
        print(
            f"Code was probably fixed... check out {state['file_path']} if it is correct"
        )
        return "ok"



from langgraph.graph import END, StateGraph

workflow = StateGraph(AgentState)

workflow.add_node("identify_filepath", identify_filepath)
workflow.add_node("execute_code_with_model", execute_code_with_model)
workflow.add_node("rewrite_code", rewrite_code)

workflow.set_entry_point("identify_filepath")
workflow.add_edge("identify_filepath", "execute_code_with_model")

workflow.add_conditional_edges(
    "execute_code_with_model",
    next_step,
    {"error": "rewrite_code", "ok": END, "max_iterations": END},
)
workflow.add_edge("rewrite_code", "execute_code_with_model")
app = workflow.compile()

app.invoke({"iterations": 1})
@pradeepdev-1995 commented on GitHub (Oct 10, 2024): Here is the code i tried ``` from langchain_core.messages import HumanMessage, SystemMessage,ToolMessage def read_code_from_file(file_path: str) -> str: with open(file_path, "r") as file: code = file.read() return code def write_code_to_file(file_path: str, code: str): with open(file_path, "w") as file: file.write(code) from typing import Annotated from langchain_experimental.utilities import PythonREPL from langchain_core.tools import tool repl = PythonREPL() @tool def python_repl(code: Annotated[str, "filename to read the code from"]): """Use this to execute python code read from a file. If you want to see the output of a value, Make sure that you read the code from correctly you should print it out with `print(...)`. This is visible to the user.""" try: result = repl.run(code) print("RESULT CODE EXECUTION:", result) except BaseException as e: return f"Failed to execute. Error: {repr(e)}" return f"Executed:\n```python\n{code}\n```\nStdout: {result}" from langchain_openai import AzureChatOpenAI model = AzureChatOpenAI( ) model_with_tools = model.bind_tools([python_repl]) from typing import TypedDict class AgentState(TypedDict): error: bool error_message: str file_path: str code: str iterations: int def identify_filepath(state: AgentState): state["file_path"] = 'testscript.py' return state def execute_code_with_model(state: AgentState): code = read_code_from_file(state["file_path"]) print("code iss") print(code) model_with_tools = model.bind_tools([python_repl]) messages = [ SystemMessage( content=""" You have got the task to execute code. Use the python_repl tool to execute it. I will give the message and your task is to detect if it was successfully run or produced an error. If the code produced an error just return 'True'. If it was sucessfully executed, return 'False'""" ), HumanMessage(content=code), ] ai_msg = model_with_tools.invoke(messages) messages.append(ai_msg) for tool_call in ai_msg.tool_calls: print("tool_call") print(tool_call) selected_tool = {"python_repl": python_repl}[tool_call["name"].lower()] tool_output = selected_tool.invoke(tool_call["args"]) state["error_message"] = tool_output messages.append(ToolMessage(tool_output, tool_call_id=tool_call["id"])) result = model_with_tools.invoke(messages) print("EVALUATION RESULT:", result) state["error"] = result.content return state def rewrite_code(state: AgentState): code = state["code"] error = state["error_message"] state["iterations"] += 1 messages = [ SystemMessage( content="You can to analyze the following code and error provided in the usermessage. Your task is to fix that code and provide the user the correct new code. VERY IMPORTANT: ONLY RETURN THE UPDATED CODE, NOTHING ELSE! Dont use a markdown style, just the code as Text" ), HumanMessage(content=f"Code: {code} | Error: {error}"), ] ai_msg = model.invoke(messages) print("NEW SUGGESTED CODE:", ai_msg.content) write_code_to_file(file_path=f'{state["file_path"]}', code=ai_msg.content) state["code"] = ai_msg.content return state def next_step(state: AgentState): if state["iterations"] > 3: print("Max Iterations done.... Exit Agent") return "max_iterations" if state["error"] == "True": print(f"Error in {state['file_path']}. {state['iterations']} tries done") return "error" if state["error"] == "False": print( f"Code was probably fixed... check out {state['file_path']} if it is correct" ) return "ok" from langgraph.graph import END, StateGraph workflow = StateGraph(AgentState) workflow.add_node("identify_filepath", identify_filepath) workflow.add_node("execute_code_with_model", execute_code_with_model) workflow.add_node("rewrite_code", rewrite_code) workflow.set_entry_point("identify_filepath") workflow.add_edge("identify_filepath", "execute_code_with_model") workflow.add_conditional_edges( "execute_code_with_model", next_step, {"error": "rewrite_code", "ok": END, "max_iterations": END}, ) workflow.add_edge("rewrite_code", "execute_code_with_model") app = workflow.compile() app.invoke({"iterations": 1}) ```
yindo commented 2026-02-20 17:35:02 -05:00
Author
Owner
Copy Link
Copy Source

@gbaian10 commented on GitHub (Oct 10, 2024):

I haven't looked at the content above yet, but I just wrote one.
You can try it to see if it's the result you want.

from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain_experimental.utilities import PythonREPL
from langchain_openai import ChatOpenAI
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition

load_dotenv()
repl = PythonREPL()


@tool
def python_repl(code: str) -> str:
    """Use this to execute python code."""
    try:
        print("Result of code execution here")
        result = repl.run(code)
        print(result)
    except BaseException as e:
        return f"Failed to execute. Error :  {e!r}"
    else:
        return f"Result of code execution {result}"


system_content = """
You have got the task to execute code.
Use the python_repl tool to execute it.
I will a message and your task is to detect if it was successfully run or produced an error.
If the code produced an error just return "True" (Example: SyntaxError, ZeroDivisionError, ...).
If it was successfully executed, return "False"
"""
prompt = ChatPromptTemplate([("system", system_content), ("placeholder", "{messages}")])
tools = [python_repl]
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)
chain = prompt | llm_with_tools


def code_to_execution(state: MessagesState) -> MessagesState:
    outputs = chain.invoke({"messages": state["messages"]})
    return {"messages": outputs}


workflow = StateGraph(MessagesState)
workflow.add_node("has_error", code_to_execution)
workflow.add_node("tools", ToolNode(tools=tools))

workflow.add_edge(START, "has_error")
workflow.add_conditional_edges("has_error", tools_condition)
workflow.add_edge("tools", "has_error")

app = workflow.compile()

if __name__ == "__main__":
    outputs = app.invoke({"messages": [("human", "print('Hello World')")]})
    print(outputs["messages"][-1].content)

    outputs = app.invoke({"messages": [("human", "1 + ")]})
    print(outputs["messages"][-1].content)

    outputs = app.invoke({"messages": [("human", "1 + '1'")]})
    print(outputs["messages"][-1].content)

    outputs = app.invoke({"messages": [("human", "1 / 0")]})
    print(outputs["messages"][-1].content)
@gbaian10 commented on GitHub (Oct 10, 2024): I haven't looked at the content above yet, but I just wrote one. You can try it to see if it's the result you want. ```py from dotenv import load_dotenv from langchain_core.prompts import ChatPromptTemplate from langchain_core.tools import tool from langchain_experimental.utilities import PythonREPL from langchain_openai import ChatOpenAI from langgraph.graph import START, MessagesState, StateGraph from langgraph.prebuilt import ToolNode, tools_condition load_dotenv() repl = PythonREPL() @tool def python_repl(code: str) -> str: """Use this to execute python code.""" try: print("Result of code execution here") result = repl.run(code) print(result) except BaseException as e: return f"Failed to execute. Error : {e!r}" else: return f"Result of code execution {result}" system_content = """ You have got the task to execute code. Use the python_repl tool to execute it. I will a message and your task is to detect if it was successfully run or produced an error. If the code produced an error just return "True" (Example: SyntaxError, ZeroDivisionError, ...). If it was successfully executed, return "False" """ prompt = ChatPromptTemplate([("system", system_content), ("placeholder", "{messages}")]) tools = [python_repl] llm = ChatOpenAI(model="gpt-4o-mini") llm_with_tools = llm.bind_tools(tools) chain = prompt | llm_with_tools def code_to_execution(state: MessagesState) -> MessagesState: outputs = chain.invoke({"messages": state["messages"]}) return {"messages": outputs} workflow = StateGraph(MessagesState) workflow.add_node("has_error", code_to_execution) workflow.add_node("tools", ToolNode(tools=tools)) workflow.add_edge(START, "has_error") workflow.add_conditional_edges("has_error", tools_condition) workflow.add_edge("tools", "has_error") app = workflow.compile() if __name__ == "__main__": outputs = app.invoke({"messages": [("human", "print('Hello World')")]}) print(outputs["messages"][-1].content) outputs = app.invoke({"messages": [("human", "1 + ")]}) print(outputs["messages"][-1].content) outputs = app.invoke({"messages": [("human", "1 + '1'")]}) print(outputs["messages"][-1].content) outputs = app.invoke({"messages": [("human", "1 / 0")]}) print(outputs["messages"][-1].content) ```
yindo commented 2026-02-20 17:35:02 -05:00
Author
Owner
Copy Link
Copy Source

@pradeepdev-1995 commented on GitHub (Oct 10, 2024):

@gbaian10 Nothing happened.
I put a sample Python program to save a txt file

import os
my_string = "This is a sample string to be saved in a file."
file_name = "output.txt"
with open(file_name, "w") as file:
    file.write(my_string)
print(f"String saved in {os.path.join(os.getcwd(), file_name)}")

And passed to the app.

current_program = """
import os
my_string = "This is a sample string to be saved in a file."
file_name = "output.txt"
with open(file_name, "w") as file:
    file.write(my_string)
print(f"String saved in {os.path.join(os.getcwd(), file_name)}")
"""

outputs = app.invoke({"messages": [("human", current_program)]})
print(outputs["messages"][-1].content)

This is the response

content='' additional_kwargs={'tool_calls': [{'id': 'call_LgBNd46hctrnufN1RagRaAEL', 'function': {'arguments': 'from typing import Any\nfrom functions import python_repl\n\ndef execute_code() -> Any:\n    code = """\nimport os\nmy_string = "This is a sample string to be saved in a file."\nfile_name = "output.txt"\nwith open(file_name, "w") as file:\n    file.write(my_string)\nprint(f"String saved in {os.path.join(os.getcwd(), file_name)}")\n    """\n    try:\n        python_repl(code=code)\n    except Exception as e:\n        return str(e)\n    else:\n        return True\n\n\nexecute_code()', 'name': 'python'}, 'type': 'function'}], 'refusal': None} response_metadata={'token_usage': {'completion_tokens': 121, 'prompt_tokens': 197, 'total_tokens': 318, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-35-turbo', 'system_fingerprint': None, 'finish_reason': 'tool_calls', 'logprobs': None, 'content_filter_results': {}} id='run-3bfde0a4-baa1-4b87-bde1-1831f640d9cc-0' invalid_tool_calls=[{'name': 'python', 'args': 'from typing import Any\nfrom functions import python_repl\n\ndef execute_code() -> Any:\n    code = """\nimport os\nmy_string = "This is a sample string to be saved in a file."\nfile_name = "output.txt"\nwith open(file_name, "w") as file:\n    file.write(my_string)\nprint(f"String saved in {os.path.join(os.getcwd(), file_name)}")\n    """\n    try:\n        python_repl(code=code)\n    except Exception as e:\n        return str(e)\n    else:\n        return True\n\n\nexecute_code()', 'id': 'call_LgBNd46hctrnufN1RagRaAEL', 'error': 'Function python arguments:\n\nfrom typing import Any\nfrom functions import python_repl\n\ndef execute_code() -> Any:\n    code = """\nimport os\nmy_string = "This is a sample string to be saved in a file."\nfile_name = "output.txt"\nwith open(file_name, "w") as file:\n    file.write(my_string)\nprint(f"String saved in {os.path.join(os.getcwd(), file_name)}")\n    """\n    try:\n        python_repl(code=code)\n    except Exception as e:\n        return str(e)\n    else:\n        return True\n\n\nexecute_code()\n\nare not valid JSON. Received JSONDecodeError Expecting value: line 1 column 1 (char 0)', 'type': 'invalid_tool_call'}] usage_metadata={'input_tokens': 197, 'output_tokens': 121, 'total_tokens': 318, 'input_token_details': {}, 'output_token_details': {}}

And could not see any saved file in my local directory.

@pradeepdev-1995 commented on GitHub (Oct 10, 2024): @gbaian10 Nothing happened. I put a sample Python program to save a txt file ``` import os my_string = "This is a sample string to be saved in a file." file_name = "output.txt" with open(file_name, "w") as file: file.write(my_string) print(f"String saved in {os.path.join(os.getcwd(), file_name)}") ``` And passed to the app. ``` current_program = """ import os my_string = "This is a sample string to be saved in a file." file_name = "output.txt" with open(file_name, "w") as file: file.write(my_string) print(f"String saved in {os.path.join(os.getcwd(), file_name)}") """ outputs = app.invoke({"messages": [("human", current_program)]}) print(outputs["messages"][-1].content) ``` This is the response ``` content='' additional_kwargs={'tool_calls': [{'id': 'call_LgBNd46hctrnufN1RagRaAEL', 'function': {'arguments': 'from typing import Any\nfrom functions import python_repl\n\ndef execute_code() -> Any:\n code = """\nimport os\nmy_string = "This is a sample string to be saved in a file."\nfile_name = "output.txt"\nwith open(file_name, "w") as file:\n file.write(my_string)\nprint(f"String saved in {os.path.join(os.getcwd(), file_name)}")\n """\n try:\n python_repl(code=code)\n except Exception as e:\n return str(e)\n else:\n return True\n\n\nexecute_code()', 'name': 'python'}, 'type': 'function'}], 'refusal': None} response_metadata={'token_usage': {'completion_tokens': 121, 'prompt_tokens': 197, 'total_tokens': 318, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-35-turbo', 'system_fingerprint': None, 'finish_reason': 'tool_calls', 'logprobs': None, 'content_filter_results': {}} id='run-3bfde0a4-baa1-4b87-bde1-1831f640d9cc-0' invalid_tool_calls=[{'name': 'python', 'args': 'from typing import Any\nfrom functions import python_repl\n\ndef execute_code() -> Any:\n code = """\nimport os\nmy_string = "This is a sample string to be saved in a file."\nfile_name = "output.txt"\nwith open(file_name, "w") as file:\n file.write(my_string)\nprint(f"String saved in {os.path.join(os.getcwd(), file_name)}")\n """\n try:\n python_repl(code=code)\n except Exception as e:\n return str(e)\n else:\n return True\n\n\nexecute_code()', 'id': 'call_LgBNd46hctrnufN1RagRaAEL', 'error': 'Function python arguments:\n\nfrom typing import Any\nfrom functions import python_repl\n\ndef execute_code() -> Any:\n code = """\nimport os\nmy_string = "This is a sample string to be saved in a file."\nfile_name = "output.txt"\nwith open(file_name, "w") as file:\n file.write(my_string)\nprint(f"String saved in {os.path.join(os.getcwd(), file_name)}")\n """\n try:\n python_repl(code=code)\n except Exception as e:\n return str(e)\n else:\n return True\n\n\nexecute_code()\n\nare not valid JSON. Received JSONDecodeError Expecting value: line 1 column 1 (char 0)', 'type': 'invalid_tool_call'}] usage_metadata={'input_tokens': 197, 'output_tokens': 121, 'total_tokens': 318, 'input_token_details': {}, 'output_token_details': {}} ``` And could not see any saved file in my local directory.
yindo commented 2026-02-20 17:35:02 -05:00
Author
Owner
Copy Link
Copy Source

@gbaian10 commented on GitHub (Oct 10, 2024):

I executed it normally, but I see you're using a different model.I think the model might have changed the code.
You should check LangSmith to see what happened.

However, I see your code.
You might execute some private behavior code, which shouldn't be passed to the LLM, and there are also concerns about the model changing the code.
So I suggest using InjectedState to avoid passing the code to the LLM.
I adjusted the code as follows; you can try it out.

from typing import Annotated

from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain_experimental.utilities import PythonREPL
from langchain_openai import ChatOpenAI
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.prebuilt import InjectedState, ToolNode, tools_condition

load_dotenv()
repl = PythonREPL()


class State(MessagesState):
    code: str


@tool
def python_repl(state: Annotated[State, InjectedState]) -> str:
    """Use this to execute python code."""
    try:
        print("Result of code execution here")
        result = repl.run(state["code"])
        print(result)
    except BaseException as e:
        return f"Failed to execute. Error :  {e!r}"
    else:
        return f"Result of code execution {result}"


system_content = """If the user requests to execute code to call 'python_repl' tool

If the code produced an error just return "True" (Example: SyntaxError, ZeroDivisionError, ...).
If it was successfully executed, return "False"
"""
prompt = ChatPromptTemplate([("system", system_content), ("placeholder", "{messages}")])
tools = [python_repl]
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)
chain = prompt | llm_with_tools


def code_to_execution(state: State) -> State:
    outputs = chain.invoke({"messages": state["messages"]})
    return {"messages": outputs}


workflow = StateGraph(State)
workflow.add_node("has_error", code_to_execution)
workflow.add_node("tools", ToolNode(tools=tools))

workflow.add_edge(START, "has_error")
workflow.add_conditional_edges("has_error", tools_condition)
workflow.add_edge("tools", "has_error")

app = workflow.compile()

current_program = """
import os
my_string = "This is a sample string to be saved in a file."
file_name = "output.txt"
with open(file_name, "w") as file:
    file.write(my_string)
print(f"String saved in {os.path.join(os.getcwd(), file_name)}")
"""

if __name__ == "__main__":
    outputs = app.invoke(
        {"messages": [("human", "Run this code.")], "code": current_program}
    )
    print(outputs["messages"][-1].content)
@gbaian10 commented on GitHub (Oct 10, 2024): I executed it normally, but I see you're using a different model.I think the model might have changed the code. You should check LangSmith to see what happened. However, I see your code. You might execute some private behavior code, which shouldn't be passed to the LLM, and there are also concerns about the model changing the code. So I suggest using `InjectedState` to avoid passing the code to the LLM. I adjusted the code as follows; you can try it out. ```py from typing import Annotated from dotenv import load_dotenv from langchain_core.prompts import ChatPromptTemplate from langchain_core.tools import tool from langchain_experimental.utilities import PythonREPL from langchain_openai import ChatOpenAI from langgraph.graph import START, MessagesState, StateGraph from langgraph.prebuilt import InjectedState, ToolNode, tools_condition load_dotenv() repl = PythonREPL() class State(MessagesState): code: str @tool def python_repl(state: Annotated[State, InjectedState]) -> str: """Use this to execute python code.""" try: print("Result of code execution here") result = repl.run(state["code"]) print(result) except BaseException as e: return f"Failed to execute. Error : {e!r}" else: return f"Result of code execution {result}" system_content = """If the user requests to execute code to call 'python_repl' tool If the code produced an error just return "True" (Example: SyntaxError, ZeroDivisionError, ...). If it was successfully executed, return "False" """ prompt = ChatPromptTemplate([("system", system_content), ("placeholder", "{messages}")]) tools = [python_repl] llm = ChatOpenAI(model="gpt-4o-mini") llm_with_tools = llm.bind_tools(tools) chain = prompt | llm_with_tools def code_to_execution(state: State) -> State: outputs = chain.invoke({"messages": state["messages"]}) return {"messages": outputs} workflow = StateGraph(State) workflow.add_node("has_error", code_to_execution) workflow.add_node("tools", ToolNode(tools=tools)) workflow.add_edge(START, "has_error") workflow.add_conditional_edges("has_error", tools_condition) workflow.add_edge("tools", "has_error") app = workflow.compile() current_program = """ import os my_string = "This is a sample string to be saved in a file." file_name = "output.txt" with open(file_name, "w") as file: file.write(my_string) print(f"String saved in {os.path.join(os.getcwd(), file_name)}") """ if __name__ == "__main__": outputs = app.invoke( {"messages": [("human", "Run this code.")], "code": current_program} ) print(outputs["messages"][-1].content) ```
yindo commented 2026-02-20 17:35:02 -05:00
Author
Owner
Copy Link
Copy Source

@gbaian10 commented on GitHub (Oct 10, 2024):

However, I believe the normal use case should be that you use natural language to make requests to the LLM, which then converts them into code.
Then execute this code through the REPL to verify its effectiveness.

@gbaian10 commented on GitHub (Oct 10, 2024): However, I believe the normal use case should be that you use natural language to make requests to the LLM, which then converts them into code. Then execute this code through the REPL to verify its effectiveness.
yindo commented 2026-02-20 17:35:02 -05:00
Author
Owner
Copy Link
Copy Source

@pradeepdev-1995 commented on GitHub (Oct 10, 2024):

@gbaian10
I am using gpt3.5 via AzureChatOpenAI and in your code I can see it is gpt-4o-mini via ChatOpenAI.
Will it cause any issue?

@pradeepdev-1995 commented on GitHub (Oct 10, 2024): @gbaian10 I am using **gpt3.5 via AzureChatOpenAI** and in your code I can see it is **gpt-4o-mini via ChatOpenAI.** Will it cause any issue?
yindo commented 2026-02-20 17:35:02 -05:00
Author
Owner
Copy Link
Copy Source

@gbaian10 commented on GitHub (Oct 10, 2024):

gpt-3.5 isn't smart enough; in the original logic, it needs to echo the code from your request.
During this output process, the code might become different from what you originally intended.
I suggest checking the process in LangSmith.

In your example, you've already prepared the correct or incorrect code. We shouldn't need the model to repeat the code, as this might cause errors and make the response lengthy.
The model only needs to know when to call the tool and when it has executed and obtained the results to provide an answer.

@gbaian10 commented on GitHub (Oct 10, 2024): gpt-3.5 isn't smart enough; in the original logic, it needs to `echo` the code from your request. During this output process, the code might become different from what you originally intended. I suggest checking the process in LangSmith. In your example, you've already prepared the correct or incorrect code. We shouldn't need the model to repeat the code, as this might cause errors and make the response lengthy. The model only needs to know when to call the tool and when it has executed and obtained the results to provide an answer.
yindo commented 2026-02-20 17:35:02 -05:00
Author
Owner
Copy Link
Copy Source

@vbarda commented on GitHub (Oct 11, 2024):

Thanks @gbaian10! Closing since this is not a langgraph issue

@vbarda commented on GitHub (Oct 11, 2024): Thanks @gbaian10! Closing since this is not a langgraph issue
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#275
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?