diff --git a/agenthub/monologue_agent/agent.py b/agenthub/monologue_agent/agent.py index a08b5e119cf..62974cc2eb9 100644 --- a/agenthub/monologue_agent/agent.py +++ b/agenthub/monologue_agent/agent.py @@ -1,3 +1,5 @@ +from litellm.exceptions import ContextWindowExceededError + import agenthub.monologue_agent.utils.prompts as prompts from opendevin.controller.agent import Agent from opendevin.controller.state.state import State @@ -30,7 +32,6 @@ if config.agent.memory_enabled: from opendevin.memory.memory import LongTermMemory -MAX_TOKEN_COUNT_PADDING = 512 MAX_OUTPUT_LENGTH = 5000 INITIAL_THOUGHTS = [ @@ -100,15 +101,29 @@ def __init__(self, llm: LLM): """ super().__init__(llm) + def _add_default_event(self, event_dict: dict): + """ + Adds a default event to the agent's monologue and memory. + + Default events are not condensed and are used to give the LLM context and examples. + + Parameters: + - event_dict: The event that will be added to monologue and memory + """ + self.monologue.add_default_event(event_dict) + if self.memory is not None: + self.memory.add_event(event_dict) + def _add_event(self, event_dict: dict): """ Adds a new event to the agent's monologue and memory. Monologue automatically condenses when it gets too large. Parameters: - - event (dict): The event that will be added to monologue and memory + - event_dict: The event that will be added to monologue and memory """ + # truncate output if it's too long if ( 'args' in event_dict and 'output' in event_dict['args'] @@ -118,26 +133,13 @@ def _add_event(self, event_dict: dict): event_dict['args']['output'][:MAX_OUTPUT_LENGTH] + '...' ) + # add event to short term history self.monologue.add_event(event_dict) + + # add event to long term memory if self.memory is not None: self.memory.add_event(event_dict) - # Test monologue token length - prompt = prompts.get_request_action_prompt( - '', - self.monologue.get_events(), - [], - ) - messages = [{'content': prompt, 'role': 'user'}] - token_count = self.llm.get_token_count(messages) - - if token_count + MAX_TOKEN_COUNT_PADDING > self.llm.max_input_tokens: - prompt = prompts.get_summarize_monologue_prompt(self.monologue.events) - summary_response = self.memory_condenser.condense( - summarize_prompt=prompt, llm=self.llm - ) - self.monologue.events = prompts.parse_summary_response(summary_response) - def _initialize(self, task: str): """ Utilizes the INITIAL_THOUGHTS list to give the agent a context for its capabilities @@ -146,7 +148,7 @@ def _initialize(self, task: str): Will execute again when called after reset. Parameters: - - task (str): The initial goal statement provided by the user + - task: The initial goal statement provided by the user Raises: - AgentNoInstructionError: If task is not provided @@ -164,7 +166,7 @@ def _initialize(self, task: str): else: self.memory = None - self.memory_condenser = MemoryCondenser() + self.memory_condenser = MemoryCondenser(action_prompt=prompts.get_action_prompt) self._add_initial_thoughts(task) self._initialized = True @@ -187,7 +189,7 @@ def _add_initial_thoughts(self, task): observation = BrowserOutputObservation( content=thought, url='', screenshot='' ) - self._add_event(event_to_memory(observation)) + self._add_default_event(event_to_memory(observation)) previous_action = '' else: action: Action = NullAction() @@ -214,34 +216,72 @@ def _add_initial_thoughts(self, task): previous_action = ActionType.BROWSE else: action = MessageAction(thought) - self._add_event(event_to_memory(action)) + self._add_default_event(event_to_memory(action)) def step(self, state: State) -> Action: """ Modifies the current state by adding the most recent actions and observations, then prompts the model to think about it's next action to take using monologue, memory, and hint. Parameters: - - state (State): The current state based on previous steps taken + - state: The current state based on previous steps taken Returns: - - Action: The next action to take based on LLM response + - The next action to take based on LLM response """ goal = state.get_current_user_intent() self._initialize(goal) + + # add the most recent actions and observations to the agent's memory for prev_action, obs in state.updated_info: self._add_event(event_to_memory(prev_action)) self._add_event(event_to_memory(obs)) + # clean info for this step state.updated_info = [] - prompt = prompts.get_request_action_prompt( + prompt = prompts.get_action_prompt( goal, - self.monologue.get_events(), + self.monologue.get_default_events(), + self.monologue.get_recent_events(), state.background_commands_obs, + # FIXME: are background_commands_obs included in recent_events? ) messages = [{'content': prompt, 'role': 'user'}] - resp = self.llm.completion(messages=messages) + + try: + resp = self.llm.completion(messages=messages) + except (ContextWindowExceededError, Exception) as e: + if ( + isinstance(e, ContextWindowExceededError) + or 'context window' in str(e) + or 'tokens' in str(e) + ): + # we need to condense the recent events + action_prompt_template = prompts.get_action_prompt_template( + task=goal, + default_events=self.monologue.get_default_events(), + background_commands_obs=state.background_commands_obs, + ) + + self.memory_condenser.action_prompt = lambda: action_prompt_template + + condensed_events, _ = self.memory_condenser.condense( + llm=self.llm, + recent_events=self.monologue.get_recent_events(), + ) + self.monologue.recent_events = condensed_events.copy() + + # try again + prompt = prompts.get_action_prompt( + goal, + self.monologue.get_default_events(), + self.monologue.get_recent_events(), + state.background_commands_obs, + ) + messages = [{'content': prompt, 'role': 'user'}] + resp = self.llm.completion(messages=messages) + action_resp = resp['choices'][0]['message']['content'] state.num_of_chars += len(prompt) + len(action_resp) action = prompts.parse_action_response(action_resp) @@ -254,10 +294,10 @@ def search_memory(self, query: str) -> list[str]: Uses search to produce top 10 results. Parameters: - - query (str): The query that we want to find related memories for + - The query that we want to find related memories for Returns: - - list[str]: A list of top 10 text results that matched the query + - A list of top 10 text results that matched the query """ if self.memory is None: return [] diff --git a/agenthub/monologue_agent/utils/prompts.py b/agenthub/monologue_agent/utils/prompts.py index d1b37a062dd..6cb9ed9d4fd 100644 --- a/agenthub/monologue_agent/utils/prompts.py +++ b/agenthub/monologue_agent/utils/prompts.py @@ -73,58 +73,58 @@ Below is the internal monologue of an automated LLM agent. Each thought is an item in a JSON array. The thoughts may be memories, actions taken by the agent, or outputs from those actions. -Please return a new, smaller JSON array, which summarizes the -internal monologue. You can summarize individual thoughts, and -you can condense related thoughts together with a description -of their content. + +The monologue has two parts: the default memories, which you must not change, +they are provided to you only for context, and the recent monologue. + +Please return a new, much smaller JSON array that summarizes the recent monologue. +When summarizing, you should condense the events that appear earlier +in the recent monologue list more aggressively, while preserving more details +for the events that appear later in the list. + +You can summarize individual thoughts, and you can condense related thoughts +together with a description of their content. %(monologue)s -Make the summaries as pithy and informative as possible. +Make the summaries as pithy and informative as possible, especially for the earlier events +in the old monologue. + Be specific about what happened and what was learned. The summary will be used as keywords for searching for the original memory. Be sure to preserve any key words or important information. -Your response must be in JSON format. It must be an object with the -key `new_monologue`, which is a JSON array containing the summarized monologue. -Each entry in the array must have an `action` key, and an `args` key. -The action key may be `summarize`, and `args.summary` should contain the summary. -You can also use the same action and args from the source monologue. -""" - - -def get_summarize_monologue_prompt(thoughts: list[dict]): - """ - Gets the prompt for summarizing the monologue +Your response must be in JSON format. It must be an object with the key `new_monologue`, +which must be a smaller JSON array containing the summarized monologue. +Each entry in the new monologue must have an `action` key, and an `args` key. +You can add a summarized entry with `action` set to "summarize" and a concise summary +in `args.summary`. You can also use the source recent event if relevant, with its original `action` and `args`. - Returns: - - str: A formatted string with the current monologue within the prompt - """ - return MONOLOGUE_SUMMARY_PROMPT % { - 'monologue': json.dumps({'old_monologue': thoughts}, indent=2), - } +Remember you must only summarize the old monologue, not the default memories. +""" -def get_request_action_prompt( +def get_action_prompt( task: str, - thoughts: list[dict], - background_commands_obs: list[CmdOutputObservation] = [], + default_events: list[dict], + recent_events: list[dict], + background_commands_obs: list[CmdOutputObservation], ): """ Gets the action prompt formatted with appropriate values. Parameters: - - task (str): The current task the agent is trying to accomplish - - thoughts (list[dict]): The agent's current thoughts - - background_commands_obs (list[CmdOutputObservation]): list of all observed background commands running + - task: The current task the agent is trying to accomplish + - thoughts: The agent's current thoughts + - background_commands_obs: list of all observed background commands running Returns: - str: Formatted prompt string with hint, task, monologue, and background included """ hint = '' - if len(thoughts) > 0: - latest_thought = thoughts[-1] + if recent_events is not None and len(recent_events) > 0: + latest_thought = recent_events[-1] if 'action' in latest_thought: if latest_thought['action'] == 'message': if latest_thought['args']['content'].startswith('OK so my task is'): @@ -134,37 +134,115 @@ def get_request_action_prompt( elif latest_thought['action'] == 'error': hint = 'Looks like that last command failed. Maybe you need to fix it, or try something else.' - bg_commands_message = '' - if len(background_commands_obs) > 0: - bg_commands_message = 'The following commands are running in the background:' - for command_obs in background_commands_obs: - bg_commands_message += ( - f'\n`{command_obs.command_id}`: {command_obs.command}' - ) - bg_commands_message += '\nYou can end any process by sending a `kill` action with the numerical `command_id` above.' + # FIXME are background commands already in recent events? + bg_commands_message = format_background_commands(background_commands_obs) user = 'opendevin' if config.run_as_devin else 'root' return ACTION_PROMPT % { 'task': task, - 'monologue': json.dumps(thoughts, indent=2), + 'monologue': json.dumps(default_events + recent_events, indent=2), 'background_commands': bg_commands_message, 'hint': hint, 'user': user, 'timeout': config.sandbox_timeout, - 'WORKSPACE_MOUNT_PATH_IN_SANDBOX': config.workspace_mount_path_in_sandbox, + # unused 'workspace_mount_path_in_sandbox': config.workspace_mount_path_in_sandbox, + } + + +def get_action_prompt_template( + task: str, + default_events: list[dict], + background_commands_obs: list[CmdOutputObservation], +) -> str: + """ + Gets the action prompt template with default_events pre-filled and a placeholder for recent_events and hint. + + Parameters: + - task: The current task the agent is trying to accomplish + - default_events: The default events to include in the prompt + - background_commands_obs: list of all observed background commands running + + Returns: + - str: Formatted prompt template string with default_events pre-filled and placeholders for recent_events and hint + """ + bg_commands_message = format_background_commands(background_commands_obs) + user = 'opendevin' if config.run_as_devin else 'root' + + return ACTION_PROMPT % { + 'task': task, + 'monologue': json.dumps(default_events, indent=2) + '%(recent_events)s', + 'background_commands': bg_commands_message, + 'hint': '%(hint)s', + 'user': user, + 'timeout': config.sandbox_timeout, + } + + +def get_action_prompt_for_summarization( + prompt_template: str, + recent_events: list[dict], +) -> str: + """ + Gets the action prompt formatted with recent_events and a generated hint. + + Parameters: + - prompt_template: The prompt template with placeholders for recent_events and hint + - recent_events: The recent events to include in the prompt + + Returns: + - str: Formatted prompt string with recent_events and a generated hint included + """ + hint = '' + if recent_events is not None and len(recent_events) > 0: + latest_thought = recent_events[-1] + if 'action' in latest_thought: + if latest_thought['action'] == 'message': + if latest_thought['args']['content'].startswith('OK so my task is'): + hint = "You're just getting started! What should you do first?" + else: + hint = "You've been thinking a lot lately. Maybe it's time to take action?" + elif latest_thought['action'] == 'error': + hint = 'Looks like that last command failed. Maybe you need to fix it, or try something else.' + + return prompt_template % { + 'recent_events': json.dumps(recent_events, indent=2), + 'hint': hint, } +def format_background_commands( + background_commands_obs: list[CmdOutputObservation] | None, +) -> str: + """ + Formats the background commands for sending in the prompt + + Parameters: + - background_commands_obs: list of all background commands running + + Returns: + - Formatted string with all background commands + """ + if background_commands_obs is None or len(background_commands_obs) == 0: + return '' + + bg_commands_message = 'The following commands are running in the background:' + for obs in background_commands_obs: + bg_commands_message += f'\n`{obs.command_id}`: {obs.command}' + bg_commands_message += '\nYou can end any process by sending a `kill` action with the numerical `command_id` above.' + + return bg_commands_message + + def parse_action_response(orig_response: str) -> Action: """ Parses a string to find an action within it Parameters: - - response (str): The string to be parsed + - orig_response: The string to be parsed Returns: - - Action: The action that was found in the response string + - The action that was found in the response string """ # attempt to load the JSON dict from the response action_dict = json.loads(orig_response) @@ -176,15 +254,30 @@ def parse_action_response(orig_response: str) -> Action: return action_from_dict(action_dict) +def get_summarize_prompt(default_events: list[dict], recent_events: list[dict]): + """ + Gets the prompt for summarizing the monologue + + Returns: + - A formatted string with the current monologue within the prompt + """ + return MONOLOGUE_SUMMARY_PROMPT % { + 'monologue': json.dumps( + {'default_memories': default_events, 'old_monologue': recent_events}, + indent=2, + ), + } + + def parse_summary_response(response: str) -> list[dict]: """ Parses a summary of the monologue Parameters: - - response (str): The response string to be parsed + - response: The response string to be parsed Returns: - - list[dict]: The list of summaries output by the model + - The list of summaries output by the model """ parsed = json.loads(response) return parsed['new_monologue'] diff --git a/opendevin/memory/__init__.py b/opendevin/memory/__init__.py index 9a705a24ac7..86d8fe89c95 100644 --- a/opendevin/memory/__init__.py +++ b/opendevin/memory/__init__.py @@ -1,5 +1,12 @@ +from .prompts import get_summarize_prompt, parse_summary_response from .condenser import MemoryCondenser from .history import ShortTermHistory from .memory import LongTermMemory -__all__ = ['LongTermMemory', 'ShortTermHistory', 'MemoryCondenser'] +__all__ = [ + 'get_summarize_prompt', + 'parse_summary_response', + 'LongTermMemory', + 'ShortTermHistory', + 'MemoryCondenser', +] diff --git a/opendevin/memory/condenser.py b/opendevin/memory/condenser.py index b8b1842dc7e..1fdd4b36bfe 100644 --- a/opendevin/memory/condenser.py +++ b/opendevin/memory/condenser.py @@ -1,26 +1,160 @@ +from typing import Callable + from opendevin.core.logger import opendevin_logger as logger from opendevin.llm.llm import LLM +from . import get_summarize_prompt, parse_summary_response + +MAX_TOKEN_COUNT_PADDING = ( + 512 # estimation of tokens to add to the prompt for the max token count +) + class MemoryCondenser: - def condense(self, summarize_prompt: str, llm: LLM): + """ + Condenses the prompt with a call to the LLM. + """ + + def __init__( + self, + action_prompt: Callable[..., str], + summarize_prompt: Callable[[list[dict]], str] = get_summarize_prompt, + ): """ - Attempts to condense the monologue by using the llm + Initialize the MemoryCondenser with the action and summarize prompts. + + action_prompt is a callable that returns the prompt that is about to be sent to the LLM. + The prompt callable will be called with recent events as arguments. + summarize_prompt is a callable that returns a specific prompt that tells the LLM to summarize the recent events. + The prompt callable will be called with events as argument. + If not provided, the default summarize prompt will be used. Parameters: - - llm (LLM): llm to be used for summarization + - action_prompt: The function to generate an action prompt. The function should accept recent events as arguments. + - summarize_prompt: The function to generate a summarize prompt. The function should accept srecent events as arguments. + """ + self.action_prompt = action_prompt + self.summarize_prompt = summarize_prompt - Raises: - - Exception: the same exception as it got from the llm or processing the response + def condense( + self, + llm: LLM, + recent_events: list[dict], + ) -> tuple[list[dict], bool]: """ + Attempts to condense the recent events of the monologue by using the llm. Returns the condensed recent events if successful, or False if not. + It includes default events in the prompt for context, but does not alter them. + Condenses the events using a summary prompt. + + Parameters: + - llm: LLM to be used for summarization. + - recent_events: List of recent events that may be condensed. + + Returns: + - The condensed recent events if successful, or False if condensation failed. + """ try: - messages = [{'content': summarize_prompt, 'role': 'user'}] - resp = llm.completion(messages=messages) - summary_response = resp['choices'][0]['message']['content'] - return summary_response + logger.debug('Condensing recent events') + + # attempt to condense the recent events + new_recent_events = self._attempt_condense(llm, recent_events) + + if not new_recent_events or len(new_recent_events) == 0: + logger.debug('Condensation failed: new_recent_events is empty') + return [], False + + # re-generate the action prompt with the condensed events + new_action_prompt = self.action_prompt(new_recent_events) + + # hope for the best + return new_recent_events, True + except Exception as e: - logger.error('Error condensing thoughts: %s', str(e), exc_info=False) + logger.error('Condensation failed: %s', str(e), exc_info=False) + return [], False + + def _attempt_condense( + self, + llm: LLM, + recent_events: list[dict], + ) -> list[dict] | None: + """ + Attempts to condense the recent events by splitting them in half and summarizing the first half. + + Parameters: + - llm: The llm to use for summarization. + - default_events: The list of default events to include in the prompt. + - recent_events: The list of recent events to include in the prompt. + + Returns: + - The condensed recent events if successful, None otherwise. + """ - # TODO If the llm fails with ContextWindowExceededError, we can try to condense the monologue chunk by chunk - raise + # Split events + midpoint = len(recent_events) // 2 + first_half = recent_events[:midpoint].copy() + second_half = recent_events[midpoint:].copy() + + # attempt to condense the first half of the recent events + summarize_prompt = self.summarize_prompt(first_half) + + # send the summarize prompt to the LLM + messages = [{'content': summarize_prompt, 'role': 'user'}] + response = llm.completion(messages=messages) + response_content = response['choices'][0]['message']['content'] + + # the new list of recent events will be source events or summarize actions + condensed_events = parse_summary_response(response_content) + + # new recent events list + if ( + not condensed_events + or not isinstance(condensed_events, list) + or len(condensed_events) == 0 + ): + return None + + condensed_events.extend(second_half) + return condensed_events + + def _needs_condense(self, **kwargs): + """ + Checks if the prompt needs to be condensed based on the token count against the limits of the llm passed in the call. + + Parameters: + - llm: The llm to use for checking the token count. + - action_prompt: The prompt to check for token count. If not provided, it will attempt to generate it using the available arguments. + - default_events: The list of default events to include in the prompt. + - recent_events: The list of recent events to include in the prompt. + + Returns: + - True if the prompt needs to be condensed, False otherwise. + """ + llm = kwargs.get('llm') + action_prompt = kwargs.get('action_prompt') + + if not llm: + logger.warning("Missing argument 'llm', cannot check token count.") + return False + + if not action_prompt: + # Attempt to generate the action_prompt using the available arguments + recent_events = kwargs.get('recent_events', []) + + action_prompt = self.action_prompt(recent_events) + + token_count = llm.get_token_count([{'content': action_prompt, 'role': 'user'}]) + return token_count >= self.get_token_limit(llm) + + def get_token_limit(self, llm: LLM) -> int: + """ + Returns the token limit to use for the llm passed in the call. + + Parameters: + - llm: The llm to get the token limit from. + + Returns: + - The token limit of the llm. + """ + return llm.max_input_tokens - MAX_TOKEN_COUNT_PADDING diff --git a/opendevin/memory/history.py b/opendevin/memory/history.py index e9d20b5f22b..d140d7f9479 100644 --- a/opendevin/memory/history.py +++ b/opendevin/memory/history.py @@ -6,14 +6,35 @@ class ShortTermHistory: """ The short term history is the most recent series of events. + + The short term history includes core events, which the agent learned in the initial prompt, and recent events of interest from the event stream. An agent can send this in the prompt or use it for other purpose. + The list of recent events may be condensed when its too long, if the agent uses the memory condenser. """ def __init__(self): """ - Initialize the empty list of events + Initialize the empty lists of events + """ + # the list of events that the agent had in this session + self.recent_events = [] + # default events are events that the agent learned in the initial prompt + self.default_events = [] + + def add_default_event(self, event_dict: dict): + """ + Adds an event to the default memory (sent in every prompt), if it is a valid event. + + Parameters: + - event_dict (dict): The event that we want to add to memory + + Raises: + - AgentEventTypeError: If event_dict is not a dict """ - self.events = [] + if not isinstance(event_dict, dict): + raise AgentEventTypeError() + + self.default_events.append(event_dict) def add_event(self, event_dict: dict): """ @@ -27,18 +48,46 @@ def add_event(self, event_dict: dict): """ if not isinstance(event_dict, dict): raise AgentEventTypeError() - self.events.append(event_dict) - def get_events(self): + # add to the list of recent events + self.recent_events.append(event_dict) + + def get_events(self) -> list[dict]: """ - Get the events in the agent's recent history. + Get the events in the agent's recent history, including core knowledge (the events it learned in the initial prompt). Returns: - List: The list of events that the agent remembers easily. """ - return self.events + return self.recent_events + self.default_events + + def get_default_events(self) -> list[dict]: + """ + Get the events in the agent's initial prompt. + + Returns: + - List: The list of core events. + """ + return self.default_events + + def get_recent_events(self, num_events=None) -> list[dict]: + """ + Get the most recent events in the agent's short term history. + + Will not return default events. + + Parameters: + - num_events (int): The number of recent events to return, defaults to all events. + + Returns: + - List: The list of the most recent events. + """ + if num_events is None: + return self.recent_events + else: + return self.recent_events[-num_events:] - def get_total_length(self): + def get_total_length(self) -> int: """ Gives the total number of characters in all history @@ -46,7 +95,7 @@ def get_total_length(self): - Int: Total number of characters of the recent history. """ total_length = 0 - for t in self.events: + for t in self.recent_events: try: total_length += len(json.dumps(t)) except TypeError as e: diff --git a/opendevin/memory/prompts.py b/opendevin/memory/prompts.py new file mode 100644 index 00000000000..e7b5b5100e3 --- /dev/null +++ b/opendevin/memory/prompts.py @@ -0,0 +1,49 @@ +from opendevin.core.utils import json + +SUMMARY_PROMPT = """ +Below is a list of events representing the history of an automated agent. Each event is an item in a JSON array. +The events may be memories, actions taken by the agent, or outputs from those actions. + +Please return a new, much smaller JSON array that summarizes the events. When summarizing, you should condense the events that appear +earlier in the list more aggressively, while preserving more details for the events that appear later in the list. + +You can summarize individual events, and you can condense related events together with a description of their content. + +%(events)s + +Make the summaries as concise and informative as possible, especially for the earlier events in the list. +Be specific about what happened and what was learned. The summary will be used as keywords for searching for the original event. +Be sure to preserve any key words or important information. + +Your response must be in JSON format. Each entry in the new monologue must have an `action` key, and an `args` key. +You can add a summarized entry with `action` set to "summarize" and a concise summary in `args.summary`. +You can also use the source event if relevant, with its original `action` and `args`. + +It must be an object with the key `summarized_events`, which must be a smaller JSON array containing the summarized events. +""" + + +def get_summarize_prompt(events: list[dict]) -> str: + """ + Gets the prompt for summarizing the events + + Returns: + - A formatted string with the current events within the prompt + """ + return SUMMARY_PROMPT % { + 'events': json.dumps(events, indent=2), + } + + +def parse_summary_response(response: str) -> list[dict]: + """ + Parses a summary of the events + + Parameters: + - response: The response string to be parsed + + Returns: + - The list of summarized events output by the model + """ + parsed = json.loads(response) + return parsed['summarized_events'] diff --git a/tests/unit/test_prompts.py b/tests/unit/test_prompts.py new file mode 100644 index 00000000000..13bb8ec4831 --- /dev/null +++ b/tests/unit/test_prompts.py @@ -0,0 +1,30 @@ +from agenthub.monologue_agent.utils.prompts import ( + format_background_commands, +) +from opendevin.core.schema.observation import ObservationType +from opendevin.events.observation.commands import CmdOutputObservation + + +def test_format_background_commands(): + background_commands_obs = [ + CmdOutputObservation( + command_id='1', + command='python server.py', + observation=ObservationType.RUN, + exit_code=0, + content='some content', + ), + CmdOutputObservation( + command_id='2', + command='npm start', + observation=ObservationType.RUN, + exit_code=0, + content='some content', + ), + ] + + formatted_commands = format_background_commands(background_commands_obs) + + assert 'python server.py' in formatted_commands + assert 'npm start' in formatted_commands + assert 'The following commands are running in the background:' in formatted_commands