Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions comps/agent/src/requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ pandas
prometheus_fastapi_instrumentator
pyarrow
pydantic #==1.10.13
qwen-agent[python_executor]
rank_bm25

# used by document loader
Expand Down
42 changes: 42 additions & 0 deletions comps/agent/src/tools/custom_tools.py
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

please add a UT for python code execution tool.

Original file line number Diff line number Diff line change
@@ -1,6 +1,14 @@
# Copyright (C) 2024 Intel Corporation
# SPDX-License-Identifier: Apache-2.0

try:
from qwen_agent.tools.python_executor import PythonExecutor
except ImportError as e:
raise ImportError(
"The dependency of PythonExecutor is not installed. "
'Please install the required dependency by running: pip install "qwen-agent[python_executor]"'
) from e


# tool for unit test
def search_web(query: str) -> str:
Expand All @@ -18,3 +26,37 @@ def search_weather(query: str) -> str:
It's clear.
"""
return ret_text


def python_executor(
code: str, get_answer_from_stdout=True, get_answer_expr=None, get_answer_symbol=None, timeout_length=20
) -> list:
"""Execute the given python code and return the result of code execution.

Args:
code: A string contains code to execute, format should be like ```\n<code>\n```.
get_answer_from_stdout: A bool variable to specify whether to get execution result from stdout.
get_answer_expr: A string variable to specify the python expression for getting execution result.
get_answer_symbol: A string variable to specify the name of global variable which contains execution result.
timeout_length: A int variable to specify timeout length for code execution in seconds.
Returns:
A tuple of code execution result, format as (execution result string, execution status string).
"""
executor = PythonExecutor(
{
"get_answer_from_stdout": get_answer_from_stdout,
"get_answer_expr": get_answer_expr,
"get_answer_symbol": get_answer_symbol,
"timeout_length": timeout_length,
}
)
return executor.call(code)


if __name__ == "__main__":
# python_executor get_answer_from_stdout test
print(python_executor("```\nprint('a')\nprint('b')\n```"))
# python_executor get_answer_expr test
print(python_executor("```\nl=2;r=3\n```", get_answer_from_stdout=False, get_answer_expr="l*r"))
# python_executor get_answer_symbol test
print(python_executor("```\nvar=1\n```", get_answer_from_stdout=False, get_answer_symbol="var"))