How to write a function that save the content of the last response to a file

I am trying to write a chatbot to help me write RPG lore.
I already wrote a function that send back a piece of lore as a json.

Now I want to be able to tell the model that this lore is fine by me and to save it.
So I wrote an other function:

def save_content(content: str):
“”"
Save the content to a file when the user asks to do so.

Args:
    content (str): The content to be saved.

Returns:
    str: A message indicating that the content has been saved.
"""
with open('content.txt', 'w') as file:
    file.write(content)
return "Content saved!\n"

I added this function to the list of function that can be called and I updated the prompt to tell the model that if asked to save, it should use this function.

But this does not seem to work.
I am using gpt-3.5-turbo-0613. Any ideas ?

Hi @julien.thielleux

I suggest you look at the Assistant API. Look for documentation here:
Assistants overview - OpenAI API

Add the following function:

{
      "name": "writeFile",
      "description": "Write the contents of a file",
      "parameters": {
        "type": "object",
        "properties": {
          "path": {
            "type": "string",
            "description": "The file path to write into"
          },
          "content": {
            "type": "string",
            "description": "The content to write"
          }
        },
        "required": ["path","content"]
      }
    }

You must handle the required_action state using this function:

// Write the contents of a specified file
function writeFile(args) {
    try {
        const parsedArgs = JSON.parse(args);
        console.log(parsedArgs);
        const { path: filePath, content } = parsedArgs;

        // Validate content is a non-empty string
        if (typeof content !== 'string' || content.trim() === '') {
            return { success: false, message: 'Content to be written is invalid or empty.' };
        }

        // Ensure that the directory exists (create if not)
        const directory = path.dirname(filePath);
        fs.mkdirSync(directory, { recursive: true });

        // Write the file
        fs.writeFileSync(filePath, content, { encoding: 'utf-8' });
        console.log({ success: true, message: 'File written successfully.' });
        return { success: true, message: 'File written successfully.' };
    } catch (error) {
        console.log(error)

        // Handle specific errors
        switch (error.code) {
            case 'EACCES':
                return { success: false, message: 'You do not have permission to write to this file.' };
            case 'ENOSPC':
                return { success: false, message: 'There is not enough space on the disk to save the file.' };
            case 'ENOENT':
                return { success: false, message: 'The specified file path could not be found.' };
            default:
                return { success: false, message: 'An unexpected error occurred while writing the file. Please try again.' };
        }
    }
}

Hope it helps :slight_smile:
Mart

Hi @mouimet

Thanks for the answer.
That’s already more or less what I did but in python.

I think the problem is more that the function is never called.
I am new to this so I may be doing something wrong, but I asked the model in the system prompt to specifically call this function when I ask it to save the created content:

You are a DM helper. You help dungeon master to create interesting pieces of lore for places. When asked to create a place you call the place_extract_function and you provide a name for the place if none is provided, the region in which this place exists, the population number, a detailed description of the place and a summary of this lore. You only give those informations. You don’t great the user, only the place information. If asked to modify the lore, you try to modify the minimum amount to include the modifications. If asked to save the newly generated content you call the save_content function.

the function is minimal for now:

def save_content(content: str):
“”"
Save the content to a file when the user asks to do so.

Args:
    content (str): The content to be saved.

Returns:
    str: A message indicating that the content has been saved.
"""
with open('content.txt', 'w') as file:
    file.write(content)
print("save_content function here")
return "Content saved!\n"

custom_functions.append(
{
“name”: “save_content”,
“description”: “Write the contents of a file”,
“parameters”: {
“type”: “object”,
“properties”: {
“content”: {
“type”: “string”,
“description”: “The content to write”
}
},
“required”: [“content”]
}
}

1 Like

@julien.thielleux

I understand however there is a very important step in between. You need to poll the status when you runstep. This is how I handle it:

  1. Run the step
    handleRunStep = async () => {
        this.openai.beta.threads.runs
            .create(this.thread.id, { assistant_id: IBRAIN })
            .then((r) => {
                this.pollAnswers(this.thread.id, r.id)

            });
    }

  1. Poll Status and Process requires_action
async pollAnswers(threadId, runId) {
        return new Promise(async (resolve, reject) => {
            let completed = false;
            while (!completed) {
                try {
                    const response = await this.openai.beta.threads.runs.retrieve(
                        threadId,
                        runId
                    );

                    bstack.emit('status.update', { status: response.status, id: response.id, created_at: response.created_at, completed_at: response.completed_at })

                    if (response.status === "completed") {
                        console.log(`Run completed!`);
                        const respmsg = await this.openai.beta.threads.messages.list(threadId);
                        console.log(respmsg.data[0].content[0].text.value);

                        const outgoingMessages = respmsg.data[0].content[0].text.value
                        completed = true;
                        bstack.emit('chat.message', { message: outgoingMessages })
                        resolve();
                    }

                    if (response.status === "expired") {
                        console.log(`Task expired!`);
                        const respmsg = await this.openai.beta.threads.messages.list(threadId);
                        console.log(respmsg.data[0].content[0].text.value);

                        const outgoingMessages = respmsg.data[0].content[0].text.value
                        completed = true;
                        bstack.emit('chat.message', { message: outgoingMessages })
                        resolve();
                    }

                    if (response.status === "requires_action") {
                        console.log(`Run Action Required!`);

                        console.log(
                            response["required_action"]["submit_tool_outputs"]["tool_calls"]
                        );
                        const tool_outputs = await Promise.all(response['required_action']['submit_tool_outputs']['tool_calls'].map(async (ra) => {

                            if (ra.type === 'function' && ra.function.name in functions) {
                                console.log(`Running ${ra.function.name} with arguments ${ra.function.arguments}`)
                                const ret = await Promise.resolve(functions[ra.function.name](ra.function.arguments))
                                console.log(`Returned `, ret)
                                return {
                                    tool_call_id: ra.id,
                                    output: JSON.stringify(ret)
                                }

                            }

                            return {
                                tool_call_id: ra.id,
                                output: "There is a problem, function not found!"
                            }
                        }));

                        const run = await this.openai.beta.threads.runs.submitToolOutputs(
                            threadId,
                            runId,
                            {
                                tool_outputs
                            }
                        );
                    }

                    if (response.status === 'failed' || response.status === 'cancelled' || response.status === 'expired' || response.status === 'cancelling') {
                        completed = true
                        console.log(`Completed ${response.status}`);
                    }

                    if (
                        response.status === "in_progress" ||
                        response.status === "queued"
                    ) {
                        console.log(`Run ${response.status}`);
                        await new Promise((resolve) => setTimeout(resolve, 2000)); // Poll every 2 seconds
                    }
                } catch (error) {
                    console.error("Error in polling (retrying):", err
                }
            }
        });
    }

Do you understand? Let me know if that helps. If you need I can migrate my class to Python. It takes 2 minutes.

Mart

1 Like

@PaulBellow has a lot of expertise in this particular domain so I’m sure he might have some advice as well.

1 Like

Thanks, but I’ve not worked with Assistants API… yet…

Oh, you meant the RPG lore bit! Hrm…

Sounds like a good idea, though. I think I’m going to do an API for my SaaS then hook it up to a Custom GPT or Assistant eventually, so things might be a bit different. Also sounds like you’re on the right track.

So it turned out that I simply wasn’t handling the tools_call properly.
I wrote a method to check if a function was called:

def handleAssistantResponse(response, messages,client):
assistant_message = response.choices[0].message
if assistant_message.tool_calls:
assistant_message.content = str(assistant_message.tool_calls[0].function)

# Call a function if there is one
messages.append({"role": assistant_message.role, "content": assistant_message.content})
if assistant_message.tool_calls:
    results = execute_function_call(assistant_message,client)
    messages.append({"role": "function", "tool_call_id": assistant_message.tool_calls[0].id, "name": assistant_message.tool_calls[0].function.name, "content": results})

return messages

And a method to execute the called function:

def execute_function_call(message,client):
function_name = message.tool_calls[0].function.name
arguments = message.tool_calls[0].function.arguments

match function_name:
    case "save_place":
        results = save_place(arguments)
    case "create_place":
        results = create_place(arguments,client)
    case "modify_place":
        results = modify_place(arguments,client)
    case "create_character":
        results = create_character(arguments,client)
    case "modify_character":
        results = modify_character(arguments,client)
    case "create_other":
        results = create_other(arguments,client)
    case "modify_other":
        results = modify_other(arguments,client)
    case _:
        results = f"Error: function {function_name} does not exist"

return results

And it’s working now

Thank you for your answers

I saw this could use some different techniques.

  • less redundant tedious code
  • more safety about ensuring the function to run exists and is accessible
  • more ability to dynamically adjust functions available to AI corresponding with sent API specification
  • error handling
def execute_function_call(message, client):
    function_name = message.tool_calls[0].function.name
    arguments = message.tool_calls[0].function.arguments

    allowed_functions = {"save_place", "create_place", "modify_place", "create_character", "modify_character", "create_other", "modify_other"}

    if function_name in allowed_functions:
        try:
            function_to_execute = getattr(sys.modules[__name__], function_name)
            results = function_to_execute(arguments, client)
        except AttributeError:
            results = f"Error: function {function_name} does not exist"
    else:
        results = f"Error: function {function_name} is not allowed"

    return results

Ideally you’ve already defined that somewhere when setting things up, no?

Ideally you put your list in a [list], also…derp. That’s a set set.

1 Like

Aha, I missed that (my local brain LLM accepted your syntax without grumble!)

You are right, that’s way better like that.
I’m not so experienced with Python yet so I didn’t know the getattr method.