vikarti.anatra/Skyvern
1
0
Fork 0
mirror of https://github.com/Skyvern-AI/skyvern.git synced 2025-09-02 02:30:07 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

remove unnecessary info in action result history (#248)

Browse source
This commit is contained in:
Kerem Yilmaz 2024-05-07 21:10:36 -07:00 • committed by GitHub
parent 01f834bbbd
commit d7599a21c5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 34 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

47
skyvern/forge/agent.py
View file

@ -616,20 +616,9 @@ class ForgeAgent:
num_elements=len(scraped_page.elements),
url=task.url,
)
# Get action results from the last app.SETTINGS.PROMPT_ACTION_HISTORY_WINDOW steps
steps = await app.DATABASE.get_task_steps(task_id=task.task_id, organization_id=task.organization_id)
window_steps = steps[-1 * SettingsManager.get_settings().PROMPT_ACTION_HISTORY_WINDOW :]
actions_and_results: list[tuple[ActionTypeUnion, list[ActionResult]]] = []
for window_step in window_steps:
if window_step.output and window_step.output.actions_and_results:
actions_and_results.extend(window_step.output.actions_and_results)
actions_and_results_str = json.dumps(
[
{"action": action.model_dump(), "results": [result.model_dump() for result in results]}
for action, results in actions_and_results
]
)
actions_and_results_str = await self._get_action_results(task)
# Generate the extract action prompt
navigation_goal = task.navigation_goal
starting_url = task.url
@ -673,6 +662,38 @@ class ForgeAgent:
return scraped_page, extract_action_prompt
async def _get_action_results(self, task: Task) -> str:
# Get action results from the last app.SETTINGS.PROMPT_ACTION_HISTORY_WINDOW steps
steps = await app.DATABASE.get_task_steps(task_id=task.task_id, organization_id=task.organization_id)
window_steps = steps[-1 * SettingsManager.get_settings().PROMPT_ACTION_HISTORY_WINDOW :]
actions_and_results: list[tuple[ActionTypeUnion, list[ActionResult]]] = []
for window_step in window_steps:
if window_step.output and window_step.output.actions_and_results:
actions_and_results.extend(window_step.output.actions_and_results)
# shall we exclude successful actions?
return json.dumps(
[
{
"action": action.model_dump(exclude_none=True),
"results": [
result.model_dump(
exclude_none=True,
exclude={
"javascript_triggered",
"interacted_with_sibling",
"interacted_with_parent",
"step_retry_number",
"step_order",
},
)
for result in results
],
}
for action, results in actions_and_results
]
)
async def get_extracted_information_for_task(self, task: Task) -> dict[str, Any] | list | str | None:
"""
Find the last successful ScrapeAction for the task and return the extracted information.