Agent framework and applications built upon Qwen, featuring Code Interpreter and Chrome browser extension.
import base64
import io
import json
import logging
import os
import queue
import re
import subprocess
import sys
import time
import traceback
import uuid

import matplotlib
import PIL.Image
from jupyter_client import BlockingKernelClient
from utils.code_utils import extract_code

WORK_DIR = os.getenv('CODE_INTERPRETER_WORK_DIR', '/tmp/workspace')

LAUNCH_KERNEL_PY = """
from ipykernel import kernelapp as app
app.launch_new_instance()
"""

_KERNEL_CLIENTS = {}


# Run this fix before jupyter starts if matplotlib cannot render CJK fonts.
# And we need to additionally run the following lines in the jupyter notebook.
#   ```python
#   import matplotlib.pyplot as plt
#   plt.rcParams['font.sans-serif'] = ['SimHei']
#   plt.rcParams['axes.unicode_minus'] = False
#   ````
def fix_matplotlib_cjk_font_issue():
    local_ttf = os.path.join(
        os.path.abspath(
            os.path.join(matplotlib.matplotlib_fname(), os.path.pardir)),
        'fonts', 'ttf', 'simhei.ttf')
    if not os.path.exists(local_ttf):
        logging.warning(f'Missing font file `{local_ttf}` for matplotlib. It may cause some error when using matplotlib.')


def start_kernel(pid):
    fix_matplotlib_cjk_font_issue()

    connection_file = os.path.join(WORK_DIR,
                                   f'kernel_connection_file_{pid}.json')
    launch_kernel_script = os.path.join(WORK_DIR, f'launch_kernel_{pid}.py')
    for f in [connection_file, launch_kernel_script]:
        if os.path.exists(f):
            logging.warning(f'{f} already exists')
            os.remove(f)

    os.makedirs(WORK_DIR, exist_ok=True)

    with open(launch_kernel_script, 'w') as fout:
        fout.write(LAUNCH_KERNEL_PY)

    kernel_process = subprocess.Popen([
        sys.executable,
        launch_kernel_script,
        '--IPKernelApp.connection_file',
        connection_file,
        '--matplotlib=inline',
        '--quiet',
    ],
                                      cwd=WORK_DIR)
    logging.info(f"INFO: kernel process's PID = {kernel_process.pid}")

    # Wait for kernel connection file to be written
    while True:
        if not os.path.isfile(connection_file):
            time.sleep(0.1)
        else:
            # Keep looping if JSON parsing fails, file may be partially written
            try:
                with open(connection_file, 'r') as fp:
                    json.load(fp)
                break
            except json.JSONDecodeError:
                pass

    # Client
    kc = BlockingKernelClient(connection_file=connection_file)
    kc.load_connection_file()
    kc.start_channels()
    kc.wait_for_ready()
    return kc


def escape_ansi(line):
    ansi_escape = re.compile(r'(?:\x1B[@-_]|[\x80-\x9F])[0-?]*[ -/]*[@-~]')
    return ansi_escape.sub('', line)


def publish_image_to_local(image_base64: str):
    image_file = str(uuid.uuid4()) + '.png'
    local_image_file = os.path.join(WORK_DIR, image_file)

    png_bytes = base64.b64decode(image_base64)
    assert isinstance(png_bytes, bytes)
    bytes_io = io.BytesIO(png_bytes)
    PIL.Image.open(bytes_io).save(local_image_file, 'png')

    return local_image_file


START_CODE = """
import signal
def _m6_code_interpreter_timeout_handler(signum, frame):
    raise TimeoutError("M6_CODE_INTERPRETER_TIMEOUT")
signal.signal(signal.SIGALRM, _m6_code_interpreter_timeout_handler)

def input(*args, **kwargs):
    raise NotImplementedError('Python input() function is disabled.')

import os
if 'upload_file' not in os.getcwd():
    os.chdir("./upload_file/")

import math
import re
import json

import seaborn as sns
sns.set_theme()

import matplotlib
import matplotlib.pyplot as plt
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

import numpy as np
import pandas as pd

from sympy import Eq, symbols, solve
"""


def code_interpreter(action_input_list: list, timeout=30, clear=False):
    code = ''
    for action_input in action_input_list:
        code += (extract_code(action_input) + '\n')
    fixed_code = []
    for line in code.split('\n'):
        fixed_code.append(line)
        if line.startswith('sns.set_theme('):
            fixed_code.append('plt.rcParams["font.sans-serif"] = ["SimHei"]')
            fixed_code.append('plt.rcParams["axes.unicode_minus"] = False')
    fixed_code = '\n'.join(fixed_code)
    if 'def solution()' in fixed_code:
        fixed_code += '\nsolution()'

    return _code_interpreter(fixed_code, timeout, clear)


def _code_interpreter(code: str, timeout, clear=False):
    if not code.strip():
        return ''
    if timeout:
        code = f'signal.alarm({timeout})\n{code}'
    if clear:
        code = "get_ipython().run_line_magic('reset', '-f')\n" + START_CODE + code

    pid = os.getpid()
    if pid not in _KERNEL_CLIENTS:
        _KERNEL_CLIENTS[pid] = start_kernel(pid)
        _code_interpreter(START_CODE, timeout=None)
    kc = _KERNEL_CLIENTS[pid]
    kc.wait_for_ready()
    kc.execute(code)
    result = ''
    image_idx = 0
    while True:
        text = ''
        image = ''
        finished = False
        msg_type = 'error'
        try:
            msg = kc.get_iopub_msg()
            msg_type = msg['msg_type']
            if msg_type == 'status':
                if msg['content'].get('execution_state') == 'idle':
                    finished = True
            elif msg_type == 'execute_result':
                text = msg['content']['data'].get('text/plain', '')
                if 'image/png' in msg['content']['data']:
                    image_b64 = msg['content']['data']['image/png']
                    image_url = publish_image_to_local(image_b64)
                    image_idx += 1
                    image = '![fig-%03d](%s)' % (image_idx, image_url)
            elif msg_type == 'display_data':
                if 'image/png' in msg['content']['data']:
                    image_b64 = msg['content']['data']['image/png']
                    image_url = publish_image_to_local(image_b64)
                    image_idx += 1
                    image = '![fig-%03d](%s)' % (image_idx, image_url)
                else:
                    text = msg['content']['data'].get('text/plain', '')
            elif msg_type == 'stream':
                msg_type = msg['content']['name']  # stdout, stderr
                text = msg['content']['text']
            elif msg_type == 'error':
                text = escape_ansi('\n'.join(msg['content']['traceback']))
                if 'M6_CODE_INTERPRETER_TIMEOUT' in text:
                    text = f'Timeout. No response after {timeout} seconds.'
        except queue.Empty:
            text = f'Timeout. No response after {timeout} seconds.'
            finished = True
        except Exception:
            text = 'The code interpreter encountered an unexpected error.'
            logging.warning(''.join(traceback.format_exception(*sys.exc_info())))
            finished = True
        if text:
            result += f'\n\n{msg_type}:\n\n```\n{text}\n```'
        if image:
            result += f'\n\n{image}'
        if finished:
            break
    result = result.lstrip('\n')
    if timeout:
        _code_interpreter('signal.alarm(0)', timeout=None)
    return result


def get_multiline_input(hint):
    print(hint)
    print('// Press ENTER to make a new line. Press CTRL-D to end input.')
    lines = []
    while True:
        try:
            line = input()
        except EOFError:  # CTRL-D
            break
        lines.append(line)
    print('// Input received.')
    if lines:
        return '\n'.join(lines)
    else:
        return ''


if __name__ == '__main__':
    while True:
        print(code_interpreter([get_multiline_input('Enter python code:')]))

benchmark/code_interpreter.py
QwenLM/Qwen-Agent
from parser import InternLMReActParser, ReActParser

from models import LLM, HFModel, Qwen
from prompt import InternLMReAct, LlamaReAct, QwenReAct

react_prompt_map = {
    'qwen': QwenReAct,
    'llama': LlamaReAct,
    'internlm': InternLMReAct,
}

react_parser_map = {
    'qwen': ReActParser,
    'llama': ReActParser,
    'internlm': InternLMReActParser,
}

model_map = {
    'qwen': Qwen,
    'llama': LLM,
    'internlm': LLM,
    'qwen-vl': HFModel
}

model_type_map = {
    'qwen-14b-chat': 'qwen',
    'qwen-1.8b-chat': 'qwen',
    'qwen-7b-chat': 'qwen',
    'llama-2-7b-chat': 'llama',
    'llama-2-13b-chat': 'llama',
    'codellama-7b-instruct': 'llama',
    'codellama-13b-instruct': 'llama',
    'internlm-7b-chat-1.1': 'internlm',
    'internlm-20b-chat': 'internlm',
    'qwen-vl-chat': 'qwen-vl',
}


model_path_map = {
    'qwen-14b-chat': 'Qwen/Qwen-14B-Chat',
    'qwen-7b-chat': 'Qwen/Qwen-7B-Chat',
    'qwen-1.8b-chat': 'Qwen/Qwen-1.8B-chat',
    'llama-2-7b-chat': 'meta-llama/Llama-2-7b-chat-hf',
    'llama-2-13b-chat': 'meta-llama/Llama-2-13b-chat-hf',
    'codellama-7b-instruct': 'codellama/CodeLlama-7b-Instruct-hf',
    'codellama-13b-instruct': 'codellama/CodeLlama-13b-Instruct-hf',
    'internlm-7b-chat-1.1': 'internlm/internlm-chat-7b-v1_1',
    'internlm-20b-chat': 'internlm/internlm-chat-20b',
    'qwen-vl-chat': 'Qwen/Qwen-VL-Chat',
}


def get_react_prompt(model_name, query, lang, upload_fname_list):
    react_prompt_cls = react_prompt_map.get(model_type_map[model_name], QwenReAct)
    return react_prompt_cls(query, lang, upload_fname_list)


def get_react_parser(model_name):
    react_parser_cls = react_parser_map.get(model_type_map[model_name], ReActParser)
    return react_parser_cls()


def get_model(model_name):
    model_path = model_path_map[model_name]
    model_cls = model_map.get(model_type_map[model_name], LLM)
    return model_cls(model_path)

benchmark/config.py
QwenLM/Qwen-Agent
import argparse
import json
import logging
import os
from parser import ReActParser

import prettytable
import tqdm
from code_interpreter import code_interpreter
from config import (get_model, get_react_parser, get_react_prompt,
                    model_path_map)
from datasets import load_dataset
from metrics.code_execution import eval_code_execution_rate
from metrics.gsm8k import eval_gsm8k_acc, is_correct
from metrics.visualization import eval_visualization_acc
from utils.code_utils import replace_upload_fname
from utils.data_utils import load_jsonl

logging.basicConfig(
    format='%(asctime)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S',
    level=logging.INFO,
)

WORK_DIR = os.getenv('CODE_INTERPRETER_WORK_DIR', '/tmp/workspace')
os.makedirs(WORK_DIR, exist_ok=True)
os.system(f'cp -r upload_file_clean {WORK_DIR}/upload_file')
os.system('cp -r upload_file_clean ./upload_file')


global_eval_result = {
    'code_executability': {
        'math': None,
        'visualization': None,
        'general': None,
    },
    'code_correctness': {
        'math': None,
        'visualization-hard': None,
        'visualization-easy': None,
    }
}


def llm_with_plugin(args, query, item=None, exec_limit=3):
    exec_count = 0

    # Build ReAct prompt
    upload_fname_list = item['input_file_path'] if item and 'input_file_path' in item else []
    lang = item['lang'] if item and 'lang' in item else 'en'
    react_prompt_obj = get_react_prompt(args.model, query, lang, upload_fname_list)
    planning_prompt = react_prompt_obj.build_prompt()

    # Execute the code when providing the first action in the query
    if '<|im_start|>' in query:
        _, prepend_code, __ = ReActParser().parse_latest_plugin_call(query)
        prepend_code = replace_upload_fname(prepend_code, upload_fname_list)
        call_plugin(_, [prepend_code], clear=(exec_count == 0))
        exec_count += 1
        exec_limit += 1

    # Inference and execute
    text = ''
    while exec_count < exec_limit:
        stop_words_list = react_prompt_obj.get_stop_words_list()
        output = text_completion(args.llm, planning_prompt + text, stop_words=stop_words_list)

        if args.gen_only:
            text += output
            break

        react_parser = get_react_parser(args.model)
        action, action_input, output = react_parser.parse_latest_plugin_call(output)
        if action:
            action_input = replace_upload_fname(action_input, upload_fname_list)
            observation = call_plugin(action, [action_input], clear=(exec_count == 0))
            output += react_prompt_obj.build_observation(observation)
            text += output
            exec_count += 1
            if 'error:' in observation or 'Traceback' in observation:
                break
        else:
            text += output
            break
    return text


def text_completion(llm, input_text, stop_words=[]):
    logging.info('Generating'.center(60, '='))
    logging.info('Input'.center(60, '-'))
    logging.info(input_text)

    output = llm.generate(input_text, stop_words)

    logging.info('Output'.center(60, '-'))
    logging.info(output)
    return output


def call_plugin(plugin_name, plugin_args_list, clear=False):
    # Relax constraints on plugin name.
    logging.info('Call code interpreter'.center(60, '='))
    obs = code_interpreter(plugin_args_list, clear=clear)
    logging.info(obs)
    return obs


def process_code_interpreter(item, writer):
    query = item['query']
    exec_limit = 3 if 'visualization' in item['tags'] else 1
    response = llm_with_plugin(args=args, query=query, item=item, exec_limit=exec_limit)
    item['gen'] = response

    writer.write(json.dumps(item, ensure_ascii=False) + '\n')
    writer.flush()


def process_gsm8k(doc, writer):
    context = doc['question']
    completion = llm_with_plugin(args=args, query=context)
    acc = is_correct(completion, doc['answer'])
    doc['completion'] = completion
    doc['acc'] = acc

    writer.write(json.dumps(doc, ensure_ascii=False) + '\n')
    writer.flush()


def sequential_processing(args, data_list, process_func, writer):
    for item in tqdm.tqdm(data_list):
        process_func(item, writer)


process_func_map = {
    'gsm8k': process_gsm8k,
    'visualization': process_code_interpreter
}


def gather_eval_result(model_name):
    for metric in global_eval_result:
        logging.info(metric)
        table = prettytable.PrettyTable()
        table.field_names = ['model'] + list(global_eval_result[metric].keys())
        row_data = [model_name]
        for item in global_eval_result[metric].values():
            item = str(item) if not item else str(round(item, 2))
            row_data.append(item)
        table.add_row(row_data)
        logging.info('\n' + str(table))


def eval_metrics(args, test_set, full_output_fname):
    # metrics
    assert os.path.exists(full_output_fname), f'Not Found File {full_output_fname}.'
    inference_res = load_jsonl(full_output_fname)
    assert len(inference_res) == len(test_set), f'There are still {len(test_set)-len(inference_res)} cases left.'

    abs_output_fname = os.path.join(os.path.dirname(os.path.abspath(__file__)), full_output_fname)
    if args.task == 'gsm8k':
        math_code_correctness = eval_gsm8k_acc(abs_output_fname)
        global_eval_result['code_correctness'].update(math_code_correctness)
    else:
        code_executability = eval_code_execution_rate(abs_output_fname, args.task, args.model)
        global_eval_result['code_executability'].update(code_executability)
        if args.task in ['all_ci', 'visualization'] and not args.eval_code_exec_only:
            visualization_code_correctness = eval_visualization_acc(abs_output_fname, args.model)
            global_eval_result['code_correctness'].update(visualization_code_correctness)


def main(args):
    current_dir = os.getcwd()
    os.makedirs(args.output_path, exist_ok=True)
    full_output_fname = os.path.join(args.output_path, (args.output_fname or f'{args.task}_{args.model}_res.jsonl'))

    if not os.path.exists(full_output_fname):
        with open(full_output_fname, 'w'):
            logging.info(f'Create file {full_output_fname} done.')

    # build data
    if args.task == 'gsm8k':
        dataset = load_dataset('gsm8k', 'main')
        test_set = dataset['test']
    else:
        eval_data_path = os.path.join(args.input_path, args.input_fname)
        test_set = [item for item in load_jsonl(eval_data_path) if args.task in item['tags']]
    logging.info(f'Test set: {len(test_set)}')

    if args.eval_only:
        eval_metrics(args, test_set, full_output_fname)
    else:
        key = 'question' if args.task == 'gsm8k' else 'query'
        cache_question = [item[key] for item in load_jsonl(full_output_fname)] if not args.force else []
        data_list = [item for item in test_set if item[key] not in cache_question]
        logging.info(f'Left cases: {len(data_list)}')

        # inference
        writer_mode = 'w' if args.force else 'a'
        f_output = open(full_output_fname, writer_mode, encoding='utf-8')
        process_func = process_func_map.get(args.task, process_code_interpreter)
        sequential_processing(args, data_list, process_func, f_output)
        f_output.close()

        # evaluate
        if not args.gen_exec_only:
            eval_metrics(args, test_set, full_output_fname)

    os.chdir(current_dir)


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument('--model', type=str, default='qwen-14b-chat', choices=list(model_path_map.keys()))
    parser.add_argument('--task', type=str, default='all', choices=['all', 'gsm8k', 'all_ci', 'visualization', 'math', 'general'])
    parser.add_argument('--output-path', type=str, default='output_data')
    parser.add_argument('--input-path', type=str, default='eval_data')
    parser.add_argument('-o', '--output-fname', type=str, default='')
    parser.add_argument('-i', '--input-fname', type=str, default='eval_code_interpreter_v1.jsonl')
    parser.add_argument('-f', '--force', action='store_true', default=False)
    parser.add_argument('--eval-only', action='store_true', default=False)
    parser.add_argument('--eval-code-exec-only', action='store_true', default=False)
    parser.add_argument('--gen-exec-only', action='store_true', default=False)
    parser.add_argument('--gen-only', action='store_true', default=False)
    args = parser.parse_args()
    return args


if __name__ == '__main__':
    args = parse_args()
    if not args.eval_only:
        args.llm = get_model(args.model)
        logging.info(f'Init {args.model} done.')

    if args.task == 'all':
        for key in ['all_ci', 'gsm8k']:
            args.task = key
            main(args)
    else:
        main(args)
    gather_eval_result(args.model)

benchmark/inference_and_execute.py
QwenLM/Qwen-Agent
import logging
import os

import func_timeout
from config import get_react_parser
from func_timeout import func_set_timeout
from utils.code_utils import extract_code, replace_upload_fname
from utils.data_utils import load_jsonl, save_jsonl

pre_load = """
import os
if 'upload_file' not in os.getcwd():
    os.chdir("./upload_file/")

import seaborn as sns

import matplotlib
# matplotlib.use('Agg')
import matplotlib.pyplot as plt
plt.ion()

import numpy as np
import pandas as pd
from sympy import Eq, symbols, solve
import re
import json
import math
"""


tags_config = {
    'visualization': {
        'timelimit': True,
        'extract_first_code': True,
    },
    'math': {
        'timelimit': True,
        'extract_first_code': False,
    },
    'general': {
        'timelimit': False,
        'extract_first_code': True,
    }
}

code_executability = {
    'math': None,
    'visualization': None,
    'general': None
}


@func_set_timeout(10)
def exec_limit_time(text):
    exec(text, locals())


def exec_code(text, timelimit=False):
    if timelimit:
        exec_limit_time(text)
    else:
        exec(text, locals())


def postprocess_code(gen_code, line):
    if '<|im_start|>' in line['query']:
        first_action_code = get_action_input_code(line['query'])
        gen_code = first_action_code + gen_code

    upload_fname_list = line['input_file_path'] if line and 'input_file_path' in line else []
    gen_code = replace_upload_fname(gen_code, upload_fname_list)

    if 'def solution()' in gen_code:
        gen_code += '\nsolution()\n'

    if 'plt.show()' in gen_code:
        gen_code += "\nplt.pause(1)\nplt.close('all')\n"

    if 'sns.' in gen_code and 'plot' in gen_code:
        gen_code += "\nplt.close('all')\n"

    gen_code = pre_load + gen_code
    return gen_code


def get_action_input_code(text, model_name='qwen-14b-chat', extract_first_code=False):
    action_input_list = []
    tmp = text
    react_parser = get_react_parser(model_name)
    while True:
        action_input = react_parser.get_first_action_input(tmp)
        if not action_input:
            break
        action_input_list.append(action_input)
        tmp = tmp.split(action_input)[1]
        if not tmp or extract_first_code:
            break

    code = ''
    for action_input in action_input_list:
        code = code + '# concat\n' + extract_code(action_input) + '\n'
    return code


def eval_code_execution_rate(output_fname, tag='all_ci', model_name='qwen-14b-chat', timelimit=False, extract_first_code=False):
    data_list = load_jsonl(output_fname)
    pip_package = []

    for line_id, line in enumerate(data_list):
        line['idx'] = line_id
        tags_list = line['tags'].split(',')
        if tag not in tags_list:
            continue

        # update args
        for cur_tag in tags_list:
            if cur_tag != 'all_ci':
                timelimit = tags_config[cur_tag]['timelimit']
                extract_first_code = tags_config[cur_tag]['extract_first_code']

        line['executable_code'] = False
        line['missing_code'] = False
        line['code_error_info'] = ''

        # get Action Input code from response
        gen_code = get_action_input_code(line['gen'], model_name=model_name, extract_first_code=extract_first_code)

        if not gen_code:
            line['missing_code'] = True
            line['code'] = ''
            line['code_error_info'] = 'missing code'
            continue

        line['code'] = gen_code
        gen_code = postprocess_code(gen_code, line)

        while True:
            try:
                exec_code(gen_code, timelimit=timelimit)
                line['executable_code'] = True
                break
            except func_timeout.exceptions.FunctionTimedOut as ex:
                line['code_error_info'] = str(ex)
                break
            except (ImportError, ModuleNotFoundError) as ex:
                try:
                    packege = str(ex).split("'")[1].strip()
                except Exception:
                    packege = ''
                if packege and packege not in pip_package:  # install package
                    pip_package.append(packege)
                    os.system('pip install '+packege)
                    logging.info(f'Automatic installation: {packege}')
                else:
                    line['code_error_info'] = str(ex)
                    break
            except Exception as ex:
                line['code_error_info'] = str(ex)
                break

        # double check
        observation = get_react_parser(model_name).get_first_observation(line['gen'])
        if line['executable_code'] and ('error:' in observation):
            logging.warning('The code executes correctly, but it has an error in IPython!')
            logging.warning(f'Code:\n{gen_code}')
            logging.warning(f'IPython error info:\n{observation}')
            logging.info('='*60)
        elif not line['executable_code'] and not ('error:' in observation):
            logging.warning('The code has an execution error, but it runs correctly in IPython!')
            logging.warning(f'Code:\n{gen_code}')
            logging.warning(f"Exec error info:\n{line['code_error_info']}")
            logging.warning(f'IPython observation:\n{observation}')
            logging.info('='*60)

    # save error data
    error_data_list = [item for item in data_list if not item['executable_code'] or item['missing_code']]
    error_data_output_fname = os.path.splitext(output_fname)[0] + '_exec_error.jsonl'
    save_jsonl(error_data_list, error_data_output_fname)

    log_result(data_list)

    return code_executability


def log_result(data_list, verbose=True):
    if verbose:
        logging.info('*'*60)
        logging.info('{:^60}'.format('Detail'))
        logging.info('*'*60)
        for line_id, line in enumerate(data_list):
            logging.info(f'Question {line_id}'.center(60, '='))
            logging.info(line['query'])

            logging.info(f'Generated {line_id}'.center(60, '-'))
            logging.info('\n' + line['gen'])

            logging.info(f'Code {line_id}'.center(60, '-'))
            logging.info('\n' + line['code'])

            logging.info(f'Exec Result {line_id}'.center(60, '-'))
            prefix_info = 'Exec Success' if line['executable_code'] else 'Exec Error: '
            exec_info = prefix_info + line['code_error_info']
            logging.info(exec_info)

    logging.info('='*60)
    logging.info('{:^60}'.format('Code Execuation Rate'))
    logging.info('='*60)
    involved_tags = []
    for line in data_list:
        involved_tags += line['tags'].split(',')
    involved_tags = list(set(involved_tags))

    for key in involved_tags:
        logging.info(f'task: {key}'.center(60, '='))
        key_item_list = [item for item in data_list if key in item['tags']]
        all_count = len(key_item_list)
        missing_code_count = len([item for item in key_item_list if item['missing_code']])
        executable_code_count = len([item for item in key_item_list if item['executable_code']])

        logging.info(f'All Test: {all_count}')
        logging.info(f'Missing Code: {missing_code_count}')
        logging.info(f'Predict Exec Success: {executable_code_count}')
        logging.info('Codes available && Execution Rate: {:.2f}'.format(executable_code_count/(all_count-missing_code_count)*100))
        logging.info('Execution Rate: {:.2f}'.format(executable_code_count/all_count*100))
        logging.info('Non-executable rate: {:.2f}'.format((all_count-missing_code_count-executable_code_count)/all_count*100))
        logging.info('Missing code rate: {:.2f}'.format(missing_code_count/all_count*100))

        if key != 'all_ci':
            code_executability[key] = executable_code_count/all_count*100

        if verbose:
            logging.info('Error List: ')
            error_list = [(item['idx'], item['code_error_info']) for item in key_item_list if item['code_error_info']]
            error_list.sort(key=lambda x: x[1])
            for x in error_list:
                logging.info(x)

benchmark/metrics/code_execution.py
QwenLM/Qwen-Agent
import logging
import os
import re

import numpy as np
from utils.data_utils import load_jsonl, save_jsonl

INVALID_ANS = '[invalid]'


def extract_answer(completion):
    def _get_last_digit(s):
        _PAT_LAST_DIGIT = re.compile(r'(?<=(\s|[\$%#{]))([+-])?(?=(\S))(0|([1-9](\d*|\d{0,2}(,\d{3})*)))?(\.\d*[1-9])?(?=(\s|[.,}]|$))')
        match = list(_PAT_LAST_DIGIT.finditer(s))
        if match:
            last_digit = match[-1].group().replace(',', '').replace('+', '')
        else:
            last_digit = None
            logging.warning(f'No digits found in {s!r}')
        return last_digit

    job_gen = completion.strip('.').replace('\n', '\\n')
    last_digit = _get_last_digit(job_gen)
    if last_digit:
        return eval(last_digit)
    else:
        return INVALID_ANS


def is_correct(completion, answer):
    gold = extract_answer(answer)
    assert gold != INVALID_ANS, 'No ground truth answer found in the document.'
    return extract_answer(completion) == gold


def eval_gsm8k_acc(output_fname):
    data_list = load_jsonl(output_fname)
    acc_res = [item['acc'] for item in data_list]
    logging.info('='*60)
    logging.info('{:^60}'.format('Math Acc.'))
    logging.info('='*60)
    logging.info('Total num={:.2f}'.format(len(acc_res)))
    logging.info('Right num={:.2f}'.format(np.sum(acc_res)))
    logging.info('Zero-shot Acc={:.2f}'.format(np.mean(acc_res)*100))

    error_data_list = [item for item in data_list if not item['acc']]
    error_data_output_fname = os.path.splitext(output_fname)[0] + '_gsm8k_error.jsonl'
    save_jsonl(error_data_list, error_data_output_fname)

    return {'math': np.mean(acc_res)*100}

benchmark/metrics/gsm8k.py
QwenLM/Qwen-Agent
import logging
import os
import re

import torch
from config import get_model, get_react_parser
from utils.data_utils import load_jsonl, save_jsonl

torch.manual_seed(1234)


EVAL_VISUAL_PROMPT_ZH = """请判断图片是否与下面的[问题]一致，如果一致则回复“right”，不一致则回复“wrong”。
[问题]：{query}
"""

EVAL_VISUAL_PROMPT_EN = """Please judge whether the image is consistent with the [Question] below, if it is consistent then reply "right", if not then reply "wrong".
[Question]: {query}
"""

visualization_code_correctness = {
    'visualization-hard': None,
    'visualization-easy': None,
}


def qwen_vl_inference(qwen_vl, imgs=[], prompt=''):
    inputs = []
    for img in imgs:
        inputs.append({'image': img})

    inputs.append({'text': prompt})
    logging.info('Eval'.center(60, '-'))
    logging.info(inputs)
    query = qwen_vl.tokenizer.from_list_format(inputs)
    response, history = qwen_vl.model.chat(qwen_vl.tokenizer, query=query, history=None)
    logging.info(response)
    logging.info('='*60)
    return response


def extract_images(text):
    regex = re.compile(r'!\[fig-(.+)\]\((.+)\)')
    results = re.findall(regex, text)
    images = []
    for res in results:
        assert len(res) == 2
        if os.path.exists(res[1]):
            images.append(res[1])
    return images


def check_images_observation(text, images, model_name):
    start_flag = get_react_parser(model_name).observation
    for image in images:
        logging.info('Image'.center(60, '-'))
        logging.info(image)

        end_idx = text.find(image)
        tmp_text = text[:end_idx+len(image)]
        start_idx = tmp_text.rfind(start_flag)
        check_text = tmp_text[start_idx + len(start_flag):]

        logging.info('Observation'.center(60, '-'))
        logging.info(check_text)

        # As long as there exists correctly executed observation, we consider `True`
        if 'error:' not in check_text and 'Traceback' not in check_text:
            return True
    return False


eval_visual_prompt = {
    'zh': EVAL_VISUAL_PROMPT_ZH,
    'en': EVAL_VISUAL_PROMPT_EN
}


def eval_visualization_acc(output_fname, model_name):
    qwen_vl = get_model('qwen-vl-chat')

    one_action, one_action_right = 0, 0
    zero_action, zero_action_right = 0, 0

    data_list = load_jsonl(output_fname)
    for item in data_list:
        if 'visualization' not in item['tags']:
            continue

        item['vis_acc'] = False
        if '<|im_end|>' in item['query']:
            one_action += 1
            prompt = item['query'].split('<|im_end|>')[0]
        else:
            zero_action += 1
            prompt = item['query']

        images = extract_images(item['gen'])

        if images and check_images_observation(item['gen'], images, model_name):
            input_prompt = eval_visual_prompt[item.get('lang', 'en')]
            format_prompt = input_prompt.format(query=prompt)
            output = qwen_vl_inference(qwen_vl, images, format_prompt)
            if 'right' in output.lower():
                item['vis_acc'] = True
                if '<|im_end|>' in item['query']:
                    one_action_right += 1
                else:
                    zero_action_right += 1

    logging.info('*'*60)
    logging.info('{:^60}'.format('Visualization Acc.'))
    logging.info('*'*60)
    logging.info('Visualization-Hard count={}, Visualization-Hard right count={}, Visualization-Hard acc={:.2f}'.format(zero_action, zero_action_right, zero_action_right/zero_action*100))
    logging.info('Visualization-Easy count={}, Visualization-Easy right count={}, Visualization-Easy acc={:.2f}'.format(one_action, one_action_right, one_action_right/one_action*100))
    logging.info('all count={}, all right={}, all acc={:.2f}'.format(zero_action+one_action, zero_action_right+one_action_right, (zero_action_right+one_action_right)/(zero_action+one_action)*100))

    visualization_code_correctness['visualization-hard'] = zero_action_right/zero_action*100
    visualization_code_correctness['visualization-easy'] = one_action_right/one_action*100

    error_data_list = [item for item in data_list if 'visualization' in item['tags'] and not item['vis_acc']]
    error_data_output_fname = os.path.splitext(output_fname)[0] + '_vis_error.jsonl'
    save_jsonl(error_data_list, error_data_output_fname)

    return visualization_code_correctness

benchmark/metrics/visualization.py
QwenLM/Qwen-Agent
from models.base import HFModel  # noqa
from models.llm import LLM  # noqa
from models.qwen import Qwen  # noqa

benchmark/models/__init__.py
QwenLM/Qwen-Agent
from transformers import AutoModelForCausalLM, AutoTokenizer
from transformers.generation import GenerationConfig


class HFModel(object):
    def __init__(self, model_path):
        self.tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
        self.model = AutoModelForCausalLM.from_pretrained(model_path, trust_remote_code=True, device_map='auto', low_cpu_mem_usage=True).eval()
        self.model.generation_config = GenerationConfig.from_pretrained(model_path, trust_remote_code=True)
        self.model.generation_config.do_sample = False

benchmark/models/base.py
QwenLM/Qwen-Agent
import torch
from models.base import HFModel


class LLM(HFModel):
    def __init__(self, model_path):
        super().__init__(model_path)

    def generate(self, input_text, stop_words=[], max_new_tokens=512):
        if isinstance(input_text, str):
            input_text = [input_text]

        input_ids = self.tokenizer(input_text)['input_ids']
        input_ids = torch.tensor(input_ids, device=self.model.device)
        gen_kwargs = {'max_new_tokens': max_new_tokens, 'do_sample': False}
        outputs = self.model.generate(input_ids, **gen_kwargs)
        s = outputs[0][input_ids.shape[1]:]
        output = self.tokenizer.decode(s, skip_special_tokens=True)

        for stop_str in stop_words:
            idx = output.find(stop_str)
            if idx != -1:
                output = output[: idx + len(stop_str)]

        return output

benchmark/models/llm.py
QwenLM/Qwen-Agent
import torch
from models.base import HFModel


class Qwen(HFModel):
    def __init__(self, model_path):
        super().__init__(model_path)

    def generate(self, input_text, stop_words=[]):
        im_end = '<|im_end|>'
        if im_end not in stop_words:
            stop_words = stop_words + [im_end]
        stop_words_ids = [self.tokenizer.encode(w) for w in stop_words]

        input_ids = torch.tensor([self.tokenizer.encode(input_text)]).to(self.model.device)
        output = self.model.generate(input_ids, stop_words_ids=stop_words_ids)
        output = output.tolist()[0]
        output = self.tokenizer.decode(output, errors='ignore')
        assert output.startswith(input_text)
        output = output[len(input_text):].replace('<|endoftext|>', '').replace(im_end, '')

        return output

benchmark/models/qwen.py
QwenLM/Qwen-Agent
from parser.internlm_parser import InternLMReActParser  # noqa
from parser.react_parser import ReActParser  # noqa

benchmark/parser/__init__.py
QwenLM/Qwen-Agent
from parser.react_parser import ReActParser


class InternLMReActParser(ReActParser):
    def __init__(self):
        self.action = '\nAction:'
        self.action_input = '\nActionInput:'
        self.action_input_stop = '<eoa>'
        self.observation = '<|System|>:Response:'
        self.observation_stop = '<TOKENS_UNUSED_2>\n<|Bot|>:'

benchmark/parser/internlm_parser.py
QwenLM/Qwen-Agent

class ReActParser(object):
    def __init__(self):
        self.action = '\nAction:'
        self.action_input = '\nAction Input:'
        self.action_input_stop = '\nObservation:'
        self.observation = '\nObservation:'
        self.observation_stop = '\nThought:'

    def parse_latest_plugin_call(self, text):
        action = self.action
        action_input = self.action_input
        observation = self.action_input_stop
        plugin_name, plugin_args = '', ''
        i = text.rfind(action)
        j = text.rfind(action_input)
        k = text.rfind(observation)
        if 0 <= i < j:  # If the text has `Action` and `Action input`,
            if k < j:  # but does not contain `Observation`,
                # then it is likely that `Observation` is ommited by the LLM,
                # because the output text may have discarded the stop word.
                text = text.rstrip() + observation  # Add it back.
            k = text.rfind(observation)
            plugin_name = text[i + len(action): j].strip()
            plugin_args = text[j + len(action_input): k].strip()
            text = text[:k]
        return plugin_name, plugin_args, text

    def _extract_first_target(self, text, start_flag, end_flag):
        target = ''
        i = text.find(start_flag)
        if i != -1:
            j = text.find(end_flag, i)
            if j != -1:
                target = text[i+len(start_flag):j].strip()
            else:
                target = text[i+len(start_flag):].strip()
        return target

    def get_first_observation(self, text):
        return self._extract_first_target(text, self.observation, self.observation_stop)

    def get_first_action_input(self, text):
        return self._extract_first_target(text, self.action_input, self.action_input_stop)

benchmark/parser/react_parser.py
QwenLM/Qwen-Agent
from prompt.internlm_react import InternLMReAct  # noqa
from prompt.llama_react import LlamaReAct  # noqa
from prompt.qwen_react import QwenReAct  # noqa
from prompt.react import ReAct  # noqa

benchmark/prompt/__init__.py
QwenLM/Qwen-Agent
from prompt.react import ReAct

INTERNLM_TOOL_DESCRIPTION = """用来执行Python代码。代码必须是一个函数，
函数名必须得是 'solution'，代码对应你的思考过程。代码实例格式如下：
```python
# import 依赖包
import xxx
def solution():
    # 初始化一些变量
    variable_names_with_real_meaning = xxx
    # 步骤一
    mid_variable = func(variable_names_with_real_meaning)
    # 步骤 x
    mid_variable = func(mid_variable)
    # 最后结果
    final_answer =  func(mid_variable)
    return final_answer
```"""


INTERNLM_TOOL = {'PythonInterpreter': INTERNLM_TOOL_DESCRIPTION}


INTERNLM_REACT_PROMPT_ZH = """<|System|>:你是一个可以调用外部工具的助手，可以使用的工具包括：
{tools_text}
如果使用工具请遵循以下格式回复：
```
Thought:思考你当前步骤需要解决什么问题，是否需要使用工具
Action:工具名称，你的工具必须从 [{tools_name_text}] 选择
ActionInput:工具输入参数
```
工具返回按照以下格式回复：
```
Response:调用工具后的结果
```
如果你已经知道了答案，或者你不需要工具，请遵循以下格式回复
```
Thought:给出最终答案的思考过程
FinalAnswer:最终答案
```
开始!<TOKENS_UNUSED_2>
<|User|>:{query}<eoh>
<|Bot|>:"""


INTERNLM_REACT_PROMPT_EN = """<|System|>:You are a assistant who can utilize external tools.
{tools_text}
To use a tool, please use the following format:
```
Thought: Think what you need to solve, do you need to use tools?
Action: the tool name, should be one of [{tools_name_text}]
ActionInput: the input to the action
```
The response after utilizing tools should using the following format:
```
Response: the results after call the tool.
``
If you already know the answer, or you do not need to use tools,
please using the following format to reply:
```
Thought: the thought process to get the final answer
FinalAnswer: final answer
```
Begin!<TOKENS_UNUSED_2>
<|User|>:{query}<eoh>
<|Bot|>:"""


class InternLMReAct(ReAct):
    def __init__(self, query, lang='en', upload_file_paths=[]):
        super().__init__(query, lang, upload_file_paths)
        self.react_template = INTERNLM_REACT_PROMPT_ZH if self.lang == 'zh' else INTERNLM_REACT_PROMPT_EN

    def build_prompt(self):
        planning_prompt = super().build_prompt()
        if '<|im_end|>' in self.query and planning_prompt.endswith('<eoh>\n<|Bot|>:'):
            planning_prompt = planning_prompt[: -len('<eoh>\n<|Bot|>:')]

        if '<|im_end|>' in self.query:
            planning_prompt = planning_prompt.replace('<|im_end|>\n<|im_start|>assistant\n', '<eoh>\n<|Bot|>:').replace('Observation:', '<eoa>\n<|System|>:Response:').replace('\nAction Input', '\nActionInput').replace('code_interpreter', 'PythonInterpreter')
            assert planning_prompt.endswith('Thought:')
            planning_prompt = planning_prompt[: -len('Thought:')] + '<TOKENS_UNUSED_2>\n<|Bot|>:'

        self.prompt = planning_prompt
        return planning_prompt

    def _build_tools_text(self):
        return INTERNLM_TOOL

    def _build_tools_name_text(self):
        return list(INTERNLM_TOOL.keys())

    def build_observation(self, observation):
        return f'<eoa>\n<|System|>:Response:{observation}\n<TOKENS_UNUSED_2>\n<|Bot|>:'

    def get_stop_words_list(self):
        return ['<eoa>']

benchmark/prompt/internlm_react.py
QwenLM/Qwen-Agent
from prompt.react import ReAct


class LlamaReAct(ReAct):
    def __init__(self, query, lang='en', upload_file_paths=[]):
        super().__init__(query, lang, upload_file_paths)

    def build_prompt(self):
        planning_prompt = super().build_prompt()
        planning_prompt = '[INST] ' + planning_prompt + ' [/INST]'

        if '<|im_end|>' in self.query:
            planning_prompt = planning_prompt.replace('<|im_end|>\n<|im_start|>assistant', ' [/INST] ')
            assert planning_prompt.endswith(' [/INST]')
            planning_prompt = planning_prompt[: -len(' [/INST]')]

        self.prompt = planning_prompt
        return planning_prompt

benchmark/prompt/llama_react.py
QwenLM/Qwen-Agent

import json
import os

from prompt.react import ReAct

QWEN_TOOLS_LIST = [
    {
        'name_for_human': '代码解释器',
        'name_for_model': 'code_interpreter',
        'description_for_model': '代码解释器，可用于执行Python代码。',
        'parameters': [{'name': 'code', 'type': 'string', 'description': '待执行的代码'}],
        'args_format': 'code'
    },
]


TOOL_DESC = """{name_for_model}: Call this tool to interact with the {name_for_human} API. What is the {name_for_human} API useful for? {description_for_model} Parameters: {parameters}"""


class QwenReAct(ReAct):
    def __init__(self, query, lang='en', upload_file_paths=[]):
        super().__init__(query, lang, upload_file_paths)

        self.upload_file_paths = [f'{os.path.basename(fname)}' for fname in upload_file_paths]
        self.list_of_plugin_info = QWEN_TOOLS_LIST
        self.fname_template = {
            'zh': '[上传文件{fname_str}]',
            'en': '[Upload file {fname_str}]',
            'en_multi': '[Upload file {fname_str}]'
        }

    def build_prompt(self):
        im_start = '<|im_start|>'
        im_end = '<|im_end|>'
        prompt = f'{im_start}system\nYou are a helpful assistant.{im_end}'

        query = super().build_prompt()

        query = query.lstrip('\n').rstrip()
        prompt += f'\n{im_start}user\n{query}{im_end}'
        if f'{im_start}assistant' not in query:
            prompt += f'\n{im_start}assistant\n{im_end}'
            assert prompt.endswith(f'\n{im_start}assistant\n{im_end}')

        prompt = prompt[: -len(f'{im_end}')]
        self.prompt = prompt
        return prompt

    def _build_tools_text(self):
        # tool info
        tools_text = []
        for plugin_info in self.list_of_plugin_info:
            tool = TOOL_DESC.format(
                name_for_model=plugin_info['name_for_model'],
                name_for_human=plugin_info['name_for_human'],
                description_for_model=plugin_info['description_for_model'],
                parameters=json.dumps(plugin_info['parameters'], ensure_ascii=False),
            )
            if plugin_info.get('args_format', 'json') == 'json':
                tool += ' Format the arguments as a JSON object.'
            elif plugin_info['args_format'] == 'code':
                tool += ' Enclose the code within triple backticks (`) at the beginning and end of the code.'
            else:
                raise NotImplementedError
            tools_text.append(tool)
        tools_text = '\n\n'.join(tools_text)
        return tools_text

    def _build_tools_name_text(self):
        return ', '.join([plugin_info['name_for_model'] for plugin_info in self.list_of_plugin_info])

benchmark/prompt/qwen_react.py
QwenLM/Qwen-Agent
import os

tools_text = """code_interpreter: Call this tool to interact with the Code Interpreter API.
What is the Code Interpreter API useful for?
Code Interpreter is used to execute Python code to deal with the following tasks:
1. Solving mathematical problems, both quantitative and qualitative
2. Doing data analysis and visualization
3. Converting files between formats
Parameters:
```py
code
```
Enclose the code within triple backticks (```) at the beginning and end of the code.
"""


REACT_PROMPT = """Answer the following questions as best you can. You have access to the following tools:

{tools_text}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tools_name_text}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can be repeated zero or more times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {query}"""


fname_template = {
    'zh': '文件{fname_str}，',
    'en_multi': 'Files {fname_str}. ',
    'en': 'File {fname_str}. ',
}


class ReAct(object):
    def __init__(self, query, lang='en', upload_file_paths=[]):
        self.query = query
        self.lang = lang
        self.upload_file_paths = [f'`{os.path.basename(fname)}`' for fname in upload_file_paths]

        self.fname_template = fname_template
        self.react_template = REACT_PROMPT
        self.prompt = ''

    def build_prompt(self):
        query = self._format_upload_fname() + self.query
        tools_text = self._build_tools_text()
        tools_name_text = self._build_tools_name_text()
        planning_prompt = self.react_template.format(query=query, tools_text=tools_text, tools_name_text=tools_name_text)

        self.prompt = planning_prompt
        return planning_prompt

    def _format_upload_fname(self):
        prefix = ''
        if self.upload_file_paths:
            fname_str = ', '.join(self.upload_file_paths)
            lang_key = 'en_multi' if self.lang == 'en' and len(self.upload_file_paths) > 1 else self.lang
            fname_template = self.fname_template[lang_key]
            prefix = fname_template.format(fname_str=fname_str)
        return prefix

    def _build_tools_text(self):
        return tools_text

    def _build_tools_name_text(self):
        return 'code_interpreter'

    def build_observation(self, observation):
        return f'\nObservation: {observation}\nThought:'

    def get_stop_words_list(self):
        return ['Observation:', 'Observation:\n']

benchmark/prompt/react.py
QwenLM/Qwen-Agent
import os
import re

import json5


def replace_upload_fname(text, upload_fname_list):
    for full_input_fname in upload_fname_list:
        if full_input_fname not in text and os.path.basename(full_input_fname) in text:
            text = text.replace(os.path.basename(full_input_fname), full_input_fname)
    return text


def extract_code(text):
    # Match triple backtick blocks first
    triple_match = re.search(r'```[^\n]*\n(.+?)```', text, re.DOTALL)
    # Match single backtick blocks second
    single_match = re.search(r'`([^`]*)`', text, re.DOTALL)
    if triple_match:
        text = triple_match.group(1)
    elif single_match:
        text = single_match.group(1)
    else:
        try:
            text = json5.loads(text)['code']
        except Exception:
            pass
    # If no code blocks found, return original text
    return text

benchmark/utils/code_utils.py
QwenLM/Qwen-Agent
import json
import logging

import tqdm


def load_jsonl(path):
    data = []
    with open(path, 'r', encoding='utf8') as f:
        for idx, line in enumerate(f, start=1):
            try:
                data.append(json.loads(line))
            except Exception as e:
                logging.info(line)
                logging.warning(f'Error at line {idx}: {e}')
                continue
    return data


def save_jsonl(data, path, progress=False, enabled=True):
    if not enabled:
        return
    with open(path, 'w', encoding='utf-8') as f:
        if progress:
            data = tqdm(data)
        for item in data:
            line = json.dumps(item, ensure_ascii=False)
            print(line, file=f)

benchmark/utils/data_utils.py
QwenLM/Qwen-Agent
from .continuewriting import ContinueWriting
from .eval_correlation import EvalCorr
from .expand import Expand
from .outline import Outline
from .react import ReAct
from .simple import Simple
from .vote import Vote
from .writefromzero import WriteFromZero

__all__ = [
    'Simple', 'ContinueWriting', 'Outline', 'Expand', 'EvalCorr', 'Vote', 'ReAct', 'WriteFromZero'
]

qwen_agent/actions/__init__.py
QwenLM/Qwen-Agent
action_list = {
    'browserqwen': {
        'summarize': 'summarize the main content of reference materials',
        'outline': 'write outline',
        'expand': 'expand text'
    }
}


def get_action_list(task):
    return action_list[task]

qwen_agent/actions/actions.py
QwenLM/Qwen-Agent
from qwen_agent.schema import Message
from qwen_agent.utils.util import count_tokens


class Action:
    def __init__(self, llm=None, stream=False):
        self.llm = llm
        self.stream = stream

    def _run(self, prompt, messages=None):
        return self.llm.chat(prompt, messages=messages, stream=self.stream)

    def _get_history(self, user_request, other_text, input_max_token=6000):
        history = ''
        other_len = count_tokens(other_text)
        print(other_len)
        valid_token_num = input_max_token - 100 - other_len
        for x in user_request[::-1]:
            now_history = '\n'
            now_history += Message('user', x[0]).to_str()
            now_history += '\n'
            now_history += Message('assistant', x[1]).to_str()
            now_token = count_tokens(now_history)
            if now_token <= valid_token_num:
                valid_token_num -= now_token
                history = now_history + history
            else:
                break
            break  # only adding one turn history
        return history

qwen_agent/actions/base.py
QwenLM/Qwen-Agent
from qwen_agent.actions.base import Action

PROMPT_TEMPLATE_CN = """
你是一个写作助手，请依据参考资料，根据给定的前置文本续写合适的内容。
#参考资料：
{ref_doc}

#前置文本：
{user_request}

保证续写内容和前置文本保持连贯，请开始续写：

"""

PROMPT_TEMPLATE_EN_COT = """
You are a writing assistant, please follow the reference materials and continue to write appropriate content based on the given previous text.

# References:
{ref_doc}

# Previous text:
{user_request}

Please start writing directly, output only the continued text, do not repeat the previous text, do not say irrelevant words, and ensure that the continued content and the previous text remain consistent.

Use the following format:
Thought: you should always think about what to do
Summariz: extracted relevant text
Final Answer: the continuation of previous text

"""

PROMPT_TEMPLATE_EN = """
You are a writing assistant, please follow the reference materials and continue to write appropriate content based on the given previous text.

# References:
{ref_doc}

# Previous text:
{user_request}

Please start writing directly, output only the continued text, do not repeat the previous text, do not say irrelevant words, and ensure that the continued content and the previous text remain consistent.
"""


class ContinueWriting(Action):
    def __init__(self, llm=None, stream=False):
        super().__init__(llm=llm, stream=stream)

    def run(self, ref_doc, user_request, messages=None, prompt_lan='CN'):
        if prompt_lan == 'CN':
            prompt = PROMPT_TEMPLATE_CN.format(
                ref_doc=ref_doc,
                user_request=user_request,
            )
        elif prompt_lan == 'EN':
            prompt = PROMPT_TEMPLATE_EN.format(
                ref_doc=ref_doc,
                user_request=user_request,
            )

        return self._run(prompt, messages=messages)

qwen_agent/actions/continuewriting.py
QwenLM/Qwen-Agent
from qwen_agent.actions.base import Action

PROMPT_TEMPLATE_CN = """
判断下列续写内容中，谁和前置文本最相关
#前置文本：
{user_request}

#续写内容：
{content_list}

必须按照如下格式返回：
{{'best_id': '和前置文本最相关的续写内容的序号', 'reason': '原因'}}

"""

PROMPT_TEMPLATE_EN = """
Determine which of the following continuation text is most relevant to the previous text
# Previous text:
{user_request}

# Continuation text:
{content_list}

It must be returned in the following format:
{{'best_id ': 'the sequence number of the continuation text most relevant to the previous text', 'reason': 'the reason'}}

"""


class EvalCorr(Action):
    def __init__(self, llm=None, stream=False):
        super().__init__(llm=llm, stream=stream)

    def run(self, user_request, content_list, messages=None, prompt_lan='CN'):
        if prompt_lan == 'CN':
            prompt = PROMPT_TEMPLATE_CN.format(
                user_request=user_request,
                content_list=content_list
            )
        elif prompt_lan == 'EN':
            prompt = PROMPT_TEMPLATE_EN.format(
                user_request=user_request,
                content_list=content_list
            )
        return self._run(prompt, messages=messages)

qwen_agent/actions/eval_correlation.py
QwenLM/Qwen-Agent
from qwen_agent.actions.base import Action

PROMPT_TEMPLATE_CN = """
你是一个写作助手，任务是依据参考资料，完成写作任务。
#参考资料：
{ref_doc}

写作标题是：{user_request}
大纲是：
{outline}

此时你的任务是扩写第{index}个一级标题对应的章节：{capture}。注意每个章节负责撰写不同的内容，所以你不需要为了全面而涵盖之后的内容。请不要在这里生成大纲。只依据给定的参考资料来写，不要引入其余知识。
"""

PROMPT_TEMPLATE_EN = """
You are a writing assistant. Your task is to complete writing article based on reference materials.

# References:
{ref_doc}

The title is: {user_request}

The outline is:
{outline}

At this point, your task is to expand the chapter corresponding to the {index} first level title: {capture}.
Note that each chapter is responsible for writing different content, so you don't need to cover the following content. Please do not generate an outline here. Write only based on the given reference materials and do not introduce other knowledge.
"""


class Expand(Action):

    def __init__(self, llm=None, stream=False):
        super().__init__(llm=llm, stream=stream)

    def run(self, ref_doc, user_request, outline='', index='1', capture='', capture_later='', messages=None, prompt_lan='CN'):
        if prompt_lan == 'CN':
            prompt = PROMPT_TEMPLATE_CN.format(
                ref_doc=ref_doc,
                user_request=user_request,
                index=index,
                outline=outline,
                capture=capture
            )
            if capture_later:
                prompt = prompt + '请在涉及 '+capture_later+' 时停止。'
        elif prompt_lan == 'EN':
            prompt = PROMPT_TEMPLATE_EN.format(
                ref_doc=ref_doc,
                user_request=user_request,
                index=index,
                outline=outline,
                capture=capture
            )
            if capture_later:
                prompt = prompt + ' Please stop when writing '+capture_later

        return self._run(prompt, messages=messages)

qwen_agent/actions/expand.py
QwenLM/Qwen-Agent
from qwen_agent.tools.tools import call_plugin  # NOQA


def func_call(query, functions, llm):
    messages = [{
        'role': 'user', 'content': query
    }]
    while True:
        rsp = llm.qwen_chat_func(messages, functions)
        if rsp['function_call']:
            yield rsp['content'].strip() + '\n'
            yield 'Action: '+rsp['function_call']['name'].strip() + '\n'
            yield 'Action Input:\n'+rsp['function_call']['arguments'] + '\n'
            bot_msg = {
                'role': 'assistant',
                'content': rsp['content'],
                'function_call': {
                    'name': rsp['function_call']['name'],
                    'arguments': rsp['function_call']['arguments'],
                }
            }
            messages.append(bot_msg)

            obs = call_plugin(rsp['function_call']['name'], rsp['function_call']['arguments'])
            func_msg = {
                'role': 'function',
                'name': rsp['function_call']['name'],
                'content': obs,
            }
            yield 'Observation: ' + obs + '\n'
            messages.append(func_msg)
        else:
            bot_msg = {
                'role': 'assistant',
                'content': rsp['content'],
            }
            yield 'Thought: I now know the final answer.\n'
            yield 'Final Answer: ' + rsp['content']
            messages.append(bot_msg)
            break

qwen_agent/actions/func_call.py
QwenLM/Qwen-Agent
from qwen_agent.actions.base import Action

PROMPT_TEMPLATE_CN = """
你是一个写作助手，任务是充分理解参考资料，从而完成写作。
#参考资料：
{ref_doc}

写作标题是：{user_request}

为了完成以上写作任务，请先列出大纲。回复只需包含大纲。大纲的一级标题全部以罗马数字计数。只依据给定的参考资料来写，不要引入其余知识。
"""

PROMPT_TEMPLATE_EN = """
You are a writing assistant. Your task is to complete writing article based on reference materials.

# References:
{ref_doc}

The title is: {user_request}

In order to complete the above writing tasks, please provide an outline first. The reply only needs to include an outline. The first level titles of the outline are all counted in Roman numerals. Write only based on the given reference materials and do not introduce other knowledge.
"""


class Outline(Action):

    def __init__(self, llm=None, stream=False):
        super().__init__(llm=llm, stream=stream)

    def run(self, ref_doc, user_request, messages=None, prompt_lan='CN'):
        if prompt_lan == 'CN':
            prompt = PROMPT_TEMPLATE_CN.format(
                ref_doc=ref_doc,
                user_request=user_request,
            )
        elif prompt_lan == 'EN':
            prompt = PROMPT_TEMPLATE_EN.format(
                ref_doc=ref_doc,
                user_request=user_request,
            )
        return self._run(prompt, messages=messages)

qwen_agent/actions/outline.py
QwenLM/Qwen-Agent
import json

from qwen_agent.actions.base import Action

from qwen_agent.tools.tools import call_plugin  # NOQA

TOOL_DESC = """{name_for_model}: Call this tool to interact with the {name_for_human} API. What is the {name_for_human} API useful for? {description_for_model} Parameters: {parameters}"""

PROMPT_REACT = """Answer the following questions as best you can. You have access to the following tools:

{tool_descs}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can be repeated zero or more times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {query}"""


class ReAct(Action):
    def __init__(self, llm=None, stream=False, list_of_plugin_info=[], prompt='default', source='dashscope'):
        super().__init__(llm=llm, stream=stream)

        self.list_of_plugin_info = list_of_plugin_info
        self.history = []
        self.prompt = prompt
        self.source = source

    def run(self, user_request, history=[], ref_doc=None, messages=None):
        prompt = self.build_input_text(user_request)
        rsp = self.run_with_tools(prompt, messages)

        return rsp

    def run_with_tools(self, prompt, messages):
        react_stop_words = ['Observation:', 'Observation:\n']

        new_messages = []
        new_messages.extend(messages)
        new_messages.append({
            'role': 'user', 'content': prompt
        })

        text = ''
        max_turn = 5
        while True and max_turn > 0:
            max_turn -= 1
            if self.source == 'dashscope':
                output = self.llm.chat('', messages=new_messages, stream=False, stop_words=react_stop_words)
            # print(new_messages)
            # print(output)
            action, action_input, output = self.parse_latest_plugin_call(output)
            if action:
                observation = call_plugin(action, action_input)
                print(observation)
                output += f'\nObservation: {observation}\nThought:'
                text += output
                new_messages[-1]['content'] += output
            else:
                text += output
                break
        return text

    def parse_latest_plugin_call(self, text):
        plugin_name, plugin_args = '', ''
        i = text.rfind('\nAction:')
        j = text.rfind('\nAction Input:')
        k = text.rfind('\nObservation:')
        if 0 <= i < j:  # If the text has `Action` and `Action input`,
            if k < j:  # but does not contain `Observation`,
                # then it is likely that `Observation` is ommited by the LLM,
                # because the output text may have discarded the stop word.
                text = text.rstrip() + '\nObservation:'  # Add it back.
            k = text.rfind('\nObservation:')
            plugin_name = text[i + len('\nAction:'):j].strip()
            plugin_args = text[j + len('\nAction Input:'):k].strip()
            text = text[:k]
        return plugin_name, plugin_args, text

    def build_input_text(self, query):
        tool_descs = []
        tool_names = []
        for info in self.list_of_plugin_info:
            tool_descs.append(
                TOOL_DESC.format(
                    name_for_model=info['name_for_model'],
                    name_for_human=info['name_for_human'],
                    description_for_model=info['description_for_model'],
                    parameters=json.dumps(
                        info['parameters'], ensure_ascii=False),
                )
            )
            tool_names.append(info['name_for_model'])
        tool_descs = '\n\n'.join(tool_descs)
        tool_names = ','.join(tool_names)

        prompt = PROMPT_REACT.format(tool_descs=tool_descs, tool_names=tool_names, query=query)
        return prompt

qwen_agent/actions/react.py
QwenLM/Qwen-Agent
from qwen_agent.actions.base import Action

PROMPT_TEMPLATE_CN = """
你是一个写作助手，请依据参考资料，充分理解参考资料内容，组织出满足用户需求的条理清晰的回复。如果用户需求中没有指定回复语言，则回复语言必须与用户需求的语言保持一致。
#参考资料：
{ref_doc}

#用户需求：
{user_request}

#助手回复：
"""

PROMPT_TEMPLATE_EN = """
You are a writing assistant. Please fully understand the content of the reference materials, and organize a clear response that meets the user requirements.
If the response language is not specified in the user requirements, the response language must be consistent with the language of the user requirements.

# References:
{ref_doc}

# User Requirements:
{user_request}

# Assistant Response:
"""

PROMPT_TEMPLATE_WITH_HISTORY_CN = """
你是一个写作助手，请依据参考资料，充分理解参考资料内容，组织出满足用户需求的条理清晰的回复。
如果用户需求中没有指定回复语言，则回复语言必须与用户需求的语言保持一致。历史对话以```开始，以```结束，如果用户需求需要理解历史对话，你可以从中提取需要的信息，请注意你要解决的始终是用户需求，而不是历史对话中的问题。
#参考资料：
{ref_doc}

#历史对话
```
{history}
```

#用户需求：
{user_request}

#助手回复：
"""

PROMPT_TEMPLATE_WITH_HISTORY_EN = """
You are a writing assistant. Please fully understand the content of the reference materials, and organize a clear response that meets the user requirements.
If the response language is not specified in the user requirements, the response language must be consistent with the language of the user requirements.
The historical dialogue starts with ``` and ends with ```. If the user needs to understand the historical dialogue, you can extract the necessary information from it. Please note that you always need to solve the user requirements, not the problems in the historical dialogue.

# References:
{ref_doc}

# Historical Dialogue
```
{history}
```

# User Requirements:
{user_request}

# Assistant Response:
"""


class Simple(Action):
    def __init__(self, llm=None, stream=False):
        super().__init__(llm=llm, stream=stream)

    def run(self, ref_doc, user_request, messages=None, prompt_lan='CN', input_max_token=6000):
        history = ''
        query = user_request
        if isinstance(user_request, list):  # history
            # Todo: consider history
            # history = self._get_history(user_request[:-1], ref_doc + user_request[-1][0], input_max_token)
            query = user_request[-1][0]
        prompt = query
        if history:
            if prompt_lan == 'CN':
                prompt = PROMPT_TEMPLATE_WITH_HISTORY_CN.format(
                    ref_doc=ref_doc,
                    history=history,
                    user_request=query
                )
            elif prompt_lan == 'EN':
                prompt = PROMPT_TEMPLATE_WITH_HISTORY_EN.format(
                    ref_doc=ref_doc,
                    history=history,
                    user_request=query
                )
        else:
            if prompt_lan == 'CN':
                prompt = PROMPT_TEMPLATE_CN.format(
                    ref_doc=ref_doc,
                    user_request=query,
                )
            elif prompt_lan == 'EN':
                prompt = PROMPT_TEMPLATE_EN.format(
                    ref_doc=ref_doc,
                    user_request=query,
                )
        # print(prompt)
        return self._run(prompt, messages=messages)

qwen_agent/actions/simple.py
QwenLM/Qwen-Agent
"""
Vote:    From the initial state to the final goal, there are n rounds of generation in between, with each round recorded as
        one thought, and n thoughts form the final path
Step1:  Generate k thoughts
Step2:  Evaluate whether the thoughts are beneficial for achieving goals, only choose positive thoughts
"""
import json

from qwen_agent.actions import ContinueWriting, EvalCorr
from qwen_agent.utils.util import get_last_one_line_context


class Vote:
    def __init__(self, llm=None, steps=1, choices=3):
        self.llm = llm
        self.steps = steps
        self.choices = choices

    def run(self, ref_doc, user_request):
        for step in range(self.steps):
            # Generate
            chois = []
            choices = []
            yield '\n========================= \n'
            yield '> 生成选项：\n'
            for i in range(self.choices):
                tmp = ''
                id_tmp = str(i+1)+': '
                yield str(i+1)+': '
                agent = ContinueWriting(self.llm, stream=True)
                res = agent.run(ref_doc, user_request)

                for trunk in res:
                    tmp += trunk
                    id_tmp += trunk
                    yield trunk
                id_tmp += '\n'
                yield '\n'
                chois.append(tmp)
                choices.append(id_tmp)
            # print(choices)

            # Eval
            yield '\n========================= \n'
            yield '> 评估：\n'
            up_context = get_last_one_line_context(user_request)
            agent = EvalCorr(self.llm, stream=True)
            res = agent.run(up_context, choices)
            tmp = ''
            for trunk in res:
                tmp += trunk
                yield trunk
            res = tmp
            # print(res)

        try:
            res = json.loads(res)
            choi = chois[int(res['best_id'])-1]
            yield '\n========================= \n'
            yield '> Final Answer: '+choi
            # return choi
        except Exception as ex:
            print('Error: ', ex)
            yield '\n========================= \n'
            yield '> Final Answer: '+chois[0]
            # return chois[0]

qwen_agent/actions/vote.py
QwenLM/Qwen-Agent
import json
import re

from qwen_agent.actions import Expand, Outline, Simple
from qwen_agent.actions.actions import get_action_list
from qwen_agent.planning.plan import Plan, default_plan
from qwen_agent.utils.util import print_traceback

PROMPT_REACT_CUSTOM_EN = """Answer the following questions as best you can. You have access to the following tools:

{tools_text}

Rules for using tools:
If the question includes code, you can consider selecting relevant tools to run the code.
If the question is to describe a series of data, consider analyzing the data and drawing a graph.
If the question is to introduce a well-known entity, such as Spider Man, the Forbidden City, etc., you can consider drawing a related diagram.
...

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tools_name_text}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can be repeated zero or more times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question. If you do not use tools, please end the generation directly without answering the Final Answer...


Begin!

Question: {query}"""

PROMPT_REACT_CUSTOM_CN = """给你一段文本，你需要分析文本，并按照选择工具的规则选择合适的工具来生成想要的结果。

可用的工具:

{tools_text}

选择工具的规则:

如果文本中包含Python代码，可以考虑选择 代码解释器 工具运行输入文本中的代码
如果文本是描述一系列数据，可以考虑选择 代码解释器 工具分析数据并生成代码画图
如果文本是介绍一个有名的实体，例如蜘蛛侠、故宫等，可以考虑画一幅相关的图
......

使用如下模板:

Text: the input text
Thought: you should always think about what to do
Action: the action to take, should be one of [{tools_name_text}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can be repeated zero or more times)
Thought: I now know the final answer
Final Answer: the final answer to the original input text. 如果不使用工具，请直接结束生成，不需要生成Final Answer。


Begin!

Text: {query}"""


def is_roman_numeral(s):
    pattern = r'^(I|V|X|L|C|D|M)+'
    match = re.match(pattern, s)
    return match is not None


class WriteFromZero:
    def __init__(self, llm=None, task='browserqwen', stream=False, auto_agent=False):
        self.llm = llm
        self.stream = stream
        self.task = task
        self.auto_agent = auto_agent

    def run(self, ref_doc, user_request, open_write_plan=False, messages=None, prompt_lan='CN'):
        action_list = get_action_list(self.task)
        # plan
        if open_write_plan:
            agent_plan = Plan(llm=self.llm, stream=self.stream)
            plans = agent_plan.run(ref_doc, user_request, action_list)
            yield '\n========================= \n'
            yield '> Plans: \n'
            res_plans = ''
            for trunk in plans:
                res_plans += trunk
                yield trunk
            try:
                res_plans = json.loads(res_plans.split('Plan:')[-1].strip())
            except Exception:
                print_traceback()
                yield '\n========================= \n'
                yield '> Use Default plans: \n'
                yield default_plan
                res_plans = json.loads(default_plan)
        else:
            yield '\n========================= \n'
            yield '> Use Default plans: \n'
            yield default_plan
            res_plans = json.loads(default_plan)

        summ = ''
        outline = ''
        for plan_id, plan in res_plans.items():
            if plan == 'summarize':
                yield '\n========================= \n'
                yield '> Summarize Browse Content: \n'
                summ = ''
                sum_agent = Simple(llm=self.llm, stream=self.stream)
                if prompt_lan == 'CN':
                    res_sum = sum_agent.run(ref_doc, '总结参考资料的主要内容')
                elif prompt_lan == 'EN':
                    res_sum = sum_agent.run(ref_doc, 'Summarize the main content of reference materials.')
                for trunk in res_sum:
                    summ += trunk
                    yield trunk
            elif plan == 'outline':
                yield '\n========================= \n'
                yield '> Generate Outline: \n'
                outline = ''
                otl_agent = Outline(llm=self.llm, stream=self.stream)
                res_otl = otl_agent.run(summ, user_request)
                for trunk in res_otl:
                    outline += trunk
                    yield trunk
            elif plan == 'expand':
                yield '\n========================= \n'
                yield '> Writing Text: \n'
                outline_list_all = outline.split('\n')
                outline_list = []
                for x in outline_list_all:
                    if is_roman_numeral(x):
                        outline_list.append(x)

                otl_num = len(outline_list)
                for i, v in enumerate(outline_list):
                    yield '\n# '
                    text = ''
                    index = i+1
                    capture = v.strip()
                    capture_later = ''
                    if i < otl_num-1:
                        capture_later = outline_list[i+1].strip()
                    exp_agent = Expand(llm=self.llm, stream=self.stream)
                    res_exp = exp_agent.run(ref_doc, user_request, outline=outline, index=str(index), capture=capture, capture_later=capture_later)
                    for trunk in res_exp:
                        text += trunk
                        yield trunk
            else:
                pass

qwen_agent/actions/writefromzero.py
QwenLM/Qwen-Agent
from abc import abstractmethod


class LLMBase:
    def __init__(self, model='qwen', api_key='',):

        self.model = model
        self.memory = None
        self.api_key = api_key.strip()

    def chat(self, query, stream=False, messages=None):
        if stream:
            return self.chat_stream(query, messages)
        else:
            return self.chat_no_stream(query, messages)

    @abstractmethod
    def chat_stream(self, query, messages=None):
        """
        :param query: str
        :param messages: List[dict]
        :return: str
        """
        raise NotImplementedError

    @abstractmethod
    def chat_no_stream(self, query, messages=None):
        """
        :param query: str
        :param messages: List[dict]
        :return: Iterator[str]
        """
        raise NotImplementedError

qwen_agent/llm/base.py
QwenLM/Qwen-Agent
import os
from http import HTTPStatus

import dashscope
import openai
from dashscope import Generation

from qwen_agent.llm.base import LLMBase


class Qwen(LLMBase):
    def __init__(self, model='qwen', api_key='', model_server=''):
        super().__init__(model=model, api_key=api_key)

        self.model_server = model_server
        self.source = ''
        self.gen = None

        if self.model_server.startswith('http'):
            openai.api_base = self.model_server
            openai.api_key = 'none'
            self.source = 'local'
        elif self.model_server.startswith('dashscope'):
            dashscope.api_key = self.api_key or os.getenv('DASHSCOPE_API_KEY', default='')
            if not dashscope.api_key:
                print('There is no DASHSCOPE_API_KEY!')
            self.source = 'dashscope'

    def chat(self, query, stream=False, messages=None, stop_words=None):

        if self.source == 'dashscope':
            if dashscope.api_key:  # use dashscope interface
                print('Now using dashscope api')
                self.gen = Generation()
                return self.chat_dashscope(query, stream=stream, messages=messages, stop_words=stop_words)
            else:
                print('There is no DASHSCOPE_API_KEY!')
                return 'Failed! There is no DASHSCOPE_API_KEY!'
        else:  # use locally deployed qwen
            print('Now using locally deployed qwen')
            return super().chat(query, stream=stream, messages=messages)

    def chat_stream(self, query, messages=None):
        print('begin: stream in qwen')
        if messages:
            response = openai.ChatCompletion.create(model=self.model, messages=messages, stream=True)
        else:
            response = openai.ChatCompletion.create(model=self.model,
                                                    messages=[{
                                                        'role': 'user',
                                                        'content': query
                                                    }],
                                                    stream=True)
        for chunk in response:
            if hasattr(chunk.choices[0].delta, 'content'):
                print(chunk.choices[0].delta.content, end='', flush=True)
                yield chunk.choices[0].delta.content

    def chat_no_stream(self, query, messages=None):
        print('begin: no stream in qwen')
        if messages:
            response = openai.ChatCompletion.create(model=self.model,
                                                    messages=messages,
                                                    stream=False)
        else:
            response = openai.ChatCompletion.create(model=self.model,
                                                    messages=[{
                                                        'role': 'user',
                                                        'content': query
                                                    }],
                                                    stream=False)
        print(response.choices[0].message.content)
        return response.choices[0].message.content

    def chat_dashscope(self, query, stream=False, messages=None, stop_words=None):
        if stream:
            return self.chat_dashscope_stream(query, messages)
        else:
            return self.chat_dashscope_no_stream(query, messages, stop_words=stop_words)

    def chat_dashscope_stream(self, query, messages=None):
        # print(query)
        if messages:
            response = self.gen.call(
                self.model,
                messages=messages,
                result_format='message',  # set the result to be "message" format.
                stream=True,
            )
        else:
            response = self.gen.call(
                self.model,
                messages=[{
                            'role': 'user',
                            'content': query
                        }],
                result_format='message',  # set the result to be "message" format.
                stream=True,
            )
        last_len = 0
        delay_len = 5
        in_delay = 0
        text = ''
        for trunk in response:
            if trunk.status_code == HTTPStatus.OK:
                # print(trunk)
                text = trunk.output.choices[0].message.content
                if (len(text)-last_len) <= delay_len:  # less than delay_len
                    in_delay = 1
                    continue
                else:
                    in_delay = 0  # print
                    real_text = text[:-delay_len]  # delay print
                    now_rsp = real_text[last_len:]
                    yield now_rsp
                    last_len = len(real_text)
            else:
                err = 'Error code: %s, error message: %s' % (
                    trunk.code, trunk.message
                )
                if trunk.code == 'DataInspectionFailed':
                    err += '\n'
                    err += '错误码: 数据检查失败，错误信息: 输入数据可能包含不适当的内容。'
                print(err)
                text = ''
                yield f'{err}'
        if text and (in_delay == 1 or last_len != len(text)):
            yield text[last_len:]

    def chat_dashscope_no_stream(self, query, messages=None, stop_words=None):
        print('begin: no stream in dashscope')
        if not stop_words:
            stop_words = []
        if messages:
            response = self.gen.call(
                self.model,
                messages=messages,
                result_format='message',  # set the result to be "message" format.
                stream=False,
                stop_words=[{'stop_str': word, 'mode': 'include'} for word in stop_words],
                top_p=0.5
            )
        else:
            response = self.gen.call(
                self.model,
                messages=[{
                            'role': 'user',
                            'content': query
                        }],
                result_format='message',  # set the result to be "message" format.
                stream=False,
                stop_words=[{'stop_str': word, 'mode': 'include'} for word in stop_words],
            )
        return response.output.choices[0].message.content

    def qwen_chat_func(self, messages, functions=None):
        print('begin: no stream in qwen_chat_func')
        if not functions:
            functions = []
        if functions:
            response = openai.ChatCompletion.create(
                model='Qwen', messages=messages, functions=functions
            )
        else:
            response = openai.ChatCompletion.create(model='Qwen', messages=messages)
        return response.choices[0].message

qwen_agent/llm/qwen.py
QwenLM/Qwen-Agent
from .memory import Memory

__all__ = ['Memory']

qwen_agent/memory/__init__.py
QwenLM/Qwen-Agent
from typing import List

from qwen_agent.schema import RefMaterial
from qwen_agent.tools.similarity_search import SimilaritySearch
from qwen_agent.utils.util import count_tokens


class Memory:
    def __init__(self, open_ss, ss_type):
        self.open_ss = open_ss
        self.ss_type = ss_type

    def get(self, query: str, records: list, llm=None, stream=False, max_token=4000) -> List[RefMaterial]:

        if not self.open_ss:
            _ref_list = self.get_top(records)
        else:
            search_agent = SimilaritySearch(type=self.ss_type, llm=llm, stream=stream)
            _ref_list = []
            for record in records:
                now_ref_list = search_agent.run(record, query)
                if now_ref_list['text']:
                    _ref_list.append(now_ref_list)

        if not _ref_list:
            _ref_list = self.get_top(records)
        # token number
        new_ref_list = []
        single_max_token = int(max_token/len(_ref_list))
        for _ref in _ref_list:
            tmp = {
                'url': _ref['url'],
                'text': []
            }
            now_token = 0
            print(len(_ref['text']))
            for x in _ref['text']:
                # lenx = len(x)
                lenx = count_tokens(x)
                if (now_token + lenx) <= single_max_token:
                    tmp['text'].append(x)
                    now_token += lenx
                else:
                    use_rate = (single_max_token-now_token)/lenx
                    tmp['text'].append(x[:int(len(x)*use_rate)])
                    break
            new_ref_list.append(tmp)

        return new_ref_list

    def get_top(self, records: list, k=6):
        _ref_list = []
        for record in records:
            raw = record['raw']
            k = min(len(raw), k)
            _ref_list.append(RefMaterial(url=record['url'], text=[x['page_content'] for x in raw[:k]]).to_dict())
        return _ref_list

qwen_agent/memory/memory.py
QwenLM/Qwen-Agent

PROMPT_TEMPLATE = """
你是一个写作助手，擅长阅读参考资料，来完成用户需求。
#参考资料：
{ref_doc}

下面是动作选项列表，可以选择有助于完成用户需求的一个或多个动作：
#动作列表
{action_list}

请依据参考资料，制定计划完成用户需求，按照如下格式返回：
Question: 用户需求。
Thought: 生成计划的原因。
Plan: 生成的计划。以JSON格式返回，例如{"action1": action_name of the first action, "action2": action_name of the second action}

举例：
#Example1：
Question: 主题在讲什么？
Thought: 该需求可以直接回答
Plan: {"action1": summarize}

#Example2:
Question: 写一篇预言TFBOYS解散的稿子
Thought: 为了实现该需求，需要先总结参考资料，列出稿子大纲，扩写稿子
Plan: {"action1": summarize, "action3": outline, "action4": expand}

Question：{user_request}
"""

default_plan = """{"action1": "summarize", "action2": "outline", "action3": "expand"}"""


class Plan:
    def __init__(self, llm=None, stream=False):
        self.llm = llm
        self.stream = stream

    def run(self, ref_doc, user_request, action_list, messages=None):
        prompt = PROMPT_TEMPLATE.format(
            ref_doc=ref_doc,
            user_request=user_request,
            action_list=action_list,
        )
        if self.llm:
            return self.llm.chat(prompt, messages=messages, stream=self.stream)
        else:
            raise NotImplementedError

qwen_agent/planning/plan.py
QwenLM/Qwen-Agent

from dataclasses import dataclass


@dataclass
class Message:
    role: str  # system / user / assistant
    content: str

    def to_str(self) -> str:
        return f'{self.role}: {self.content}'

    def to_dict(self) -> dict:
        return {
            'role': self.role,
            'content': self.content
        }


@dataclass
class Record:
    url: str
    time: str
    type: str
    raw: list
    extract: str
    topic: str
    checked: bool
    session: list

    def to_dict(self) -> dict:
        return {
            'url': self.url,
            'time': self.time,
            'type': self.type,
            'raw': self.raw,
            'extract': self.extract,
            'topic': self.topic,
            'checked': self.checked,
            'session': self.session
        }


@dataclass
class RefMaterial:
    url: str
    text: list

    def to_dict(self) -> dict:
        return {
            'url': self.url,
            'text': self.text,
        }

qwen_agent/schema.py
QwenLM/Qwen-Agent
from enum import Enum


class SimilaritySearchType(Enum):
    KeyWord = 'keyword'
    QueryMatch = 'querymatch'
    LLM = 'llm'
    Jaccard = 'jaccard'

qwen_agent/tools/__init__.py
QwenLM/Qwen-Agent
import asyncio
import atexit
import base64
import glob
import io
import json
import os
import queue
import re
import shutil
import signal
import subprocess
import sys
import time
import traceback
import uuid
from pathlib import Path
from typing import Dict, Optional

import matplotlib
import PIL.Image
from jupyter_client import BlockingKernelClient

sys.path.insert(
    0,
    str(Path(__file__).absolute().parent.parent.parent))  # NOQA

from qwen_server import config_browserqwen  # NOQA
from qwen_agent.utils.util import extract_code, print_traceback  # NOQA

WORK_DIR = os.getenv('CODE_INTERPRETER_WORK_DIR', config_browserqwen.code_interpreter_ws)
LAUNCH_KERNEL_PY = """
from ipykernel import kernelapp as app
app.launch_new_instance()
"""
INIT_CODE_FILE = str(Path(__file__).absolute().parent / 'code_interpreter_init_kernel.py')
ALIB_FONT_FILE = str(Path(__file__).absolute().parent / 'AlibabaPuHuiTi-3-45-Light.ttf')

_KERNEL_CLIENTS: Dict[int, BlockingKernelClient] = {}


def _start_kernel(pid) -> BlockingKernelClient:
    connection_file = os.path.join(WORK_DIR,
                                   f'kernel_connection_file_{pid}.json')
    launch_kernel_script = os.path.join(WORK_DIR, f'launch_kernel_{pid}.py')
    for f in [connection_file, launch_kernel_script]:
        if os.path.exists(f):
            print(f'WARNING: {f} already exists')
            os.remove(f)

    os.makedirs(WORK_DIR, exist_ok=True)

    with open(launch_kernel_script, 'w') as fout:
        fout.write(LAUNCH_KERNEL_PY)

    kernel_process = subprocess.Popen([
        sys.executable,
        launch_kernel_script,
        '--IPKernelApp.connection_file',
        connection_file,
        '--matplotlib=inline',
        '--quiet',
    ],
        cwd=WORK_DIR)
    print(f"INFO: kernel process's PID = {kernel_process.pid}")

    # Wait for kernel connection file to be written
    while True:
        if not os.path.isfile(connection_file):
            time.sleep(0.1)
        else:
            # Keep looping if JSON parsing fails, file may be partially written
            try:
                with open(connection_file, 'r') as fp:
                    json.load(fp)
                break
            except json.JSONDecodeError:
                pass

    # Client
    kc = BlockingKernelClient(connection_file=connection_file)
    asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())
    kc.load_connection_file()
    kc.start_channels()
    kc.wait_for_ready()
    return kc


def _kill_kernels():
    for v in _KERNEL_CLIENTS.values():
        v.shutdown()
    for k in list(_KERNEL_CLIENTS.keys()):
        del _KERNEL_CLIENTS[k]


atexit.register(_kill_kernels)
signal.signal(signal.SIGTERM, _kill_kernels)
signal.signal(signal.SIGINT, _kill_kernels)


def _serve_image(image_base64: str) -> str:
    image_file = f'{uuid.uuid4()}.png'
    local_image_file = os.path.join(WORK_DIR, image_file)

    png_bytes = base64.b64decode(image_base64)
    assert isinstance(png_bytes, bytes)
    bytes_io = io.BytesIO(png_bytes)
    PIL.Image.open(bytes_io).save(local_image_file, 'png')

    if os.path.exists(config_browserqwen.address_file):
        with open(config_browserqwen.address_file) as f:
            data = json.load(f)
        figure_url = 'http://' + data['address'] + ':' + str(config_browserqwen.fast_api_port) + '/static'
    else:
        figure_url = config_browserqwen.fast_api_figure_url
    print(figure_url)
    image_url = f'{figure_url}/{image_file}'
    return image_url


def _escape_ansi(line: str) -> str:
    ansi_escape = re.compile(r'(?:\x1B[@-_]|[\x80-\x9F])[0-?]*[ -/]*[@-~]')
    return ansi_escape.sub('', line)


def _execute_code(kc: BlockingKernelClient, code: str) -> str:
    kc.wait_for_ready()
    kc.execute(code)
    result = ''
    image_idx = 0
    while True:
        text = ''
        image = ''
        finished = False
        msg_type = 'error'
        try:
            msg = kc.get_iopub_msg()
            msg_type = msg['msg_type']
            if msg_type == 'status':
                if msg['content'].get('execution_state') == 'idle':
                    finished = True
            elif msg_type == 'execute_result':
                text = msg['content']['data'].get('text/plain', '')
                if 'image/png' in msg['content']['data']:
                    image_b64 = msg['content']['data']['image/png']
                    image_url = _serve_image(image_b64)
                    image_idx += 1
                    image = '![fig-%03d](%s)' % (image_idx, image_url)
            elif msg_type == 'display_data':
                if 'image/png' in msg['content']['data']:
                    image_b64 = msg['content']['data']['image/png']
                    image_url = _serve_image(image_b64)
                    image_idx += 1
                    image = '![fig-%03d](%s)' % (image_idx, image_url)
                else:
                    text = msg['content']['data'].get('text/plain', '')
            elif msg_type == 'stream':
                msg_type = msg['content']['name']  # stdout, stderr
                text = msg['content']['text']
            elif msg_type == 'error':
                text = _escape_ansi('\n'.join(msg['content']['traceback']))
                if 'M6_CODE_INTERPRETER_TIMEOUT' in text:
                    text = 'Timeout: Code execution exceeded the time limit.'
        except queue.Empty:
            text = 'Timeout: Code execution exceeded the time limit.'
            finished = True
        except Exception:
            text = 'The code interpreter encountered an unexpected error.'
            print(''.join(traceback.format_exception(*sys.exc_info())))
            finished = True
        if text:
            result += f'\n\n{msg_type}:\n\n```\n{text}\n```'
        if image:
            result += f'\n\n{image}'
        if finished:
            break
    result = result.lstrip('\n')
    return result


def fix_matplotlib_cjk_font_issue():
    ttf_name = os.path.basename(ALIB_FONT_FILE)
    local_ttf = os.path.join(
        os.path.abspath(
            os.path.join(matplotlib.matplotlib_fname(), os.path.pardir)),
        'fonts', 'ttf', ttf_name)
    if not os.path.exists(local_ttf):
        try:
            shutil.copy(ALIB_FONT_FILE, local_ttf)
            font_list_cache = os.path.join(matplotlib.get_cachedir(),
                                           'fontlist-*.json')
            for cache_file in glob.glob(font_list_cache):
                with open(cache_file) as fin:
                    cache_content = fin.read()
                if ttf_name not in cache_content:
                    os.remove(cache_file)
        except Exception:
            print_traceback()


def code_interpreter(action_input: str, timeout: Optional[int] = 30) -> str:
    code = extract_code(action_input)

    if not code.strip():
        return ''

    pid: int = os.getpid()
    if pid in _KERNEL_CLIENTS:
        kc = _KERNEL_CLIENTS[pid]
    else:
        fix_matplotlib_cjk_font_issue()
        kc = _start_kernel(pid)
        with open(INIT_CODE_FILE) as fin:
            start_code = fin.read()
            start_code = start_code.replace('{{M6_FONT_PATH}}', repr(ALIB_FONT_FILE)[1:-1])
        print(_execute_code(kc, start_code))
        _KERNEL_CLIENTS[pid] = kc

    if timeout:
        code = f'_M6CountdownTimer.start({timeout})\n{code}'

    fixed_code = []
    for line in code.split('\n'):
        fixed_code.append(line)
        if line.startswith('sns.set_theme('):
            fixed_code.append('plt.rcParams["font.family"] = _m6_font_prop.get_name()')
    fixed_code = '\n'.join(fixed_code)
    result = _execute_code(kc, fixed_code)

    if timeout:
        _execute_code(kc, '_M6CountdownTimer.cancel()')

    return result


def _get_multiline_input(hint: str) -> str:
    print(hint)
    print('// Press ENTER to make a new line. Press CTRL-D to end input.')
    lines = []
    while True:
        try:
            line = input()
        except EOFError:  # CTRL-D
            break
        lines.append(line)
    print('// Input received.')
    if lines:
        return '\n'.join(lines)
    else:
        return ''


if __name__ == '__main__':
    while True:
        print(code_interpreter(_get_multiline_input('Enter python code:')))

qwen_agent/tools/code_interpreter.py
QwenLM/Qwen-Agent
import json  # noqa
import math  # noqa
import os  # noqa
import re  # noqa
import signal

import matplotlib  # noqa
import matplotlib.pyplot as plt
import numpy as np  # noqa
import pandas as pd  # noqa
import seaborn as sns
from matplotlib.font_manager import FontProperties
from sympy import Eq, solve, symbols  # noqa


def input(*args, **kwargs):
    raise NotImplementedError('Python input() function is disabled.')


def _m6_timout_handler(_signum=None, _frame=None):
    raise TimeoutError('M6_CODE_INTERPRETER_TIMEOUT')


try:
    signal.signal(signal.SIGALRM, _m6_timout_handler)
except AttributeError:  # windows
    pass


class _M6CountdownTimer:
    @classmethod
    def start(cls, timeout: int):
        try:
            signal.alarm(timeout)
        except AttributeError:  # windows
            pass  # TODO: I haven't found a solution that works with jupyter yet.

    @classmethod
    def cancel(cls):
        try:
            signal.alarm(0)
        except AttributeError:  # windows
            pass  # TODO


sns.set_theme()

_m6_font_prop = FontProperties(fname='{{M6_FONT_PATH}}')
plt.rcParams['font.family'] = _m6_font_prop.get_name()

qwen_agent/tools/code_interpreter_init_kernel.py
QwenLM/Qwen-Agent
import json
import urllib.parse

import json5


def image_gen(plugin_args):
    prompt = json5.loads(plugin_args)['prompt']
    prompt = urllib.parse.quote(prompt)
    return json.dumps({'image_url': f'https://image.pollinations.ai/prompt/{prompt}'}, ensure_ascii=False)

qwen_agent/tools/image_gen.py
QwenLM/Qwen-Agent
import re

import html2text

from qwen_agent.actions import Simple


def gen_q(text):
    agent = Simple(stream=False)
    query = '根据参考资料提几个最合适的问题，问题答案可以在参考资料中找出'
    res = agent.run(text, query)
    return res


def parse_pdf_pypdf(path, pre_gen_question=False):
    from langchain.document_loaders import PyPDFLoader
    loader = PyPDFLoader(path)
    pages = loader.load_and_split()
    # print(pages)
    if pre_gen_question:
        res = []
        for page in pages:
            print(len(page.page_content.split(' ')))
            res.append({'page_content': page.page_content, 'metadata': page.metadata, 'related_questions': gen_q(page.page_content)})
    else:
        res = [{'page_content': page.page_content, 'metadata': page.metadata} for page in pages]

    return res


def parse_html(htmltext):
    return html2text.html2text(htmltext)


def pre_process_html(s):
    # replace multiple newlines
    s = re.sub('\n+', '\n', s)
    # replace special string
    s = s.replace('Add to Qwen\'s Reading List', '')
    return s


def parse_html_bs(path, pre_gen_question=False):
    from langchain.document_loaders import BSHTMLLoader
    loader = BSHTMLLoader(path, open_encoding='utf-8')
    pages = loader.load_and_split()

    if pre_gen_question:
        res = []
        for page in pages:
            print(len(page.page_content.split(' ')))
            res.append({'page_content': pre_process_html(page.page_content), 'metadata': page.metadata, 'related_questions': gen_q(page.page_content)})
    else:
        res = [{'page_content': pre_process_html(page.page_content), 'metadata': page.metadata} for page in pages]

    return res

qwen_agent/tools/parse_doc.py
QwenLM/Qwen-Agent
import importlib

from qwen_agent.tools import SimilaritySearchType


class SimilaritySearch:
    def __init__(self, type='keyword', llm=None, stream=False):
        self.type = type
        if type == SimilaritySearchType.KeyWord.value:
            module = 'qwen_agent.tools.similarity_search_keyword'
            run_func = importlib.import_module(module).SSKeyWord(llm, stream).run
        elif type == SimilaritySearchType.QueryMatch.value:
            module = 'qwen_agent.tools.similarity_search_querymatch'
            run_func = importlib.import_module(module).SSQueryMatch(llm, stream).run
        elif type == SimilaritySearchType.LLM.value:
            module = 'qwen_agent.tools.similarity_search_llm'
            run_func = importlib.import_module(module).SSLLM(llm).run
        elif type == SimilaritySearchType.Jaccard.value:
            module = 'qwen_agent.tools.similarity_search_jaccard'
            run_func = importlib.import_module(module).SSJaccard(llm, stream).run
        else:
            raise NotImplementedError
        self.run_func = run_func

    def run(self, path, query):
        return self.run_func(path, query)

qwen_agent/tools/similarity_search.py
QwenLM/Qwen-Agent
from qwen_agent.schema import RefMaterial
from qwen_agent.utils.util import get_split_word


class SSJaccard:
    def __init__(self, llm=None, stream=False):
        self.llm = llm
        self.stream = stream

    def run(self, line, query):
        """
        Input: one line
        Output: the relative text
        """
        wordlist = get_split_word(query)

        content = line['raw']
        if isinstance(content, str):
            content = content.split('\n')

        res = []
        sims = []
        for i, page in enumerate(content):
            sim = self.filter_section(page, wordlist)
            sims.append([i, sim])
        sims.sort(key=lambda x: x[1], reverse=True)
        # print('sims: ', sims)
        max_sims = sims[0][1]
        if max_sims != 0:
            for i, x in enumerate(sims):
                if x[1] < max_sims and i > 3:
                    break
                page = content[x[0]]
                text = ''
                if isinstance(page, str):
                    text = content[x[0]]
                elif isinstance(page, dict):
                    text = page['page_content']
                res.append(text)
            # for x in res:
            #     print("=========")
            #     print(x)
        return RefMaterial(url=line['url'], text=res).to_dict()

    def filter_section(self, page, wordlist):
        if isinstance(page, str):
            text = page
        elif isinstance(page, dict):
            text = page['page_content']
        else:
            print(type(page))
            raise TypeError

        pagelist = get_split_word(text)
        sim = self.jaccard_similarity(wordlist, pagelist)

        return sim

    def jaccard_similarity(self, list1, list2):
        s1 = set(list1)
        s2 = set(list2)
        return len(s1.intersection(s2))  # avoid text length impact
        # return len(s1.intersection(s2)) / len(s1.union(s2))  # jaccard similarity

qwen_agent/tools/similarity_search_jaccard.py
QwenLM/Qwen-Agent
from qwen_agent.schema import RefMaterial
from qwen_agent.utils.util import get_key_word


class SSKeyWord:
    def __init__(self, llm=None, stream=False):
        self.llm = llm
        self.stream = stream

    def run(self, line, query):
        """
        Input: one line
        Output: the relative text
        """
        wordlist = get_key_word(query)

        content = line['raw']
        if isinstance(content, str):
            content = content.split('\n')

        res = []
        for page in content:
            rel_text = self.filter_section(page, wordlist)
            if rel_text:
                res.append(rel_text)
        return RefMaterial(url=line['url'], text=res).to_dict()

    def filter_section(self, page, wordlist):
        if isinstance(page, str):
            text = page
        elif isinstance(page, dict):
            text = page['page_content']
        else:
            print(type(page))
            raise TypeError

        for x in wordlist:
            if x in text:
                return text
        return ''

qwen_agent/tools/similarity_search_keyword.py
QwenLM/Qwen-Agent
import json

from qwen_agent.schema import RefMaterial
from qwen_agent.utils.util import print_traceback

PROMPT_TEMPLATE = """
Given reference materials and questions, please determine if the reference materials and questions are relevant
reference materials:
{ref_doc}

Use the following format:

Question: the input question you must answer
Thought: you should always think about the reason.
Final Answer: {{"res": just judge if the reference materials and question are relevant, should be on of [related, unrelated, 相关, 不相关]}}, the answer must be json format, no additional replies allowed.

Begin!

Question: {user_request}"""


class SSLLM:
    def __init__(self, llm=None):
        self.llm = llm

    def run(self, line, query):
        """
        Input: one line
        Output: the relative text
        """
        content = line['raw']
        if isinstance(content, str):
            content = content.split('\n')

        res = []
        for page in content:
            rel_text = self.filter_section(page, query)
            if rel_text:
                res.append(rel_text)
        return RefMaterial(url=line['url'], text=res).to_dict()

    def filter_section(self, page, query):
        if isinstance(page, str):
            text = page
        elif isinstance(page, dict):
            # text = page['related_questions']
            text = page['page_content']
        else:
            print(type(page))
            raise TypeError
        prompt = PROMPT_TEMPLATE.format(
            ref_doc=text,
            user_request=query,
        )

        res = self.llm.chat(prompt, stream=False)

        print(res)
        if 'Final Answer:' in res:
            fa = res.split('Final Answer:')[-1].strip()
            try:
                fa = json.loads(fa)
                answer = fa['res']
            except Exception:
                print_traceback()
                answer = fa
            if answer == 'related' or answer == '相关':
                print(res)
                return text
            else:
                return ''
        return ''

qwen_agent/tools/similarity_search_llm.py
QwenLM/Qwen-Agent
from qwen_agent.schema import RefMaterial
from qwen_agent.utils.util import get_key_word


class SSQueryMatch:
    def __init__(self, llm=None, stream=False):
        self.llm = llm
        self.stream = stream

    def run(self, doc, query):
        """
        Input: lines
        Output: the relative text
        """
        wordlist = get_key_word(query)

        res = []
        for page in doc['raw']:
            # print(page)
            rel_text = self.filter_section(page, wordlist)
            if rel_text:
                res.append(rel_text)
        return RefMaterial(url=doc['url'], text=res).to_dict()

    def filter_section(self, page, wordlist):
        text = page['related_questions']
        res = ''
        for x in wordlist:
            if x in text:
                res = page['page_content']
                break
        return res

qwen_agent/tools/similarity_search_querymatch.py
QwenLM/Qwen-Agent
from qwen_agent.tools.code_interpreter import code_interpreter  # NOQA
from qwen_agent.tools.image_gen import image_gen  # NOQA

tools_list = [
        {
            'name_for_human': '代码解释器',
            'name_for_model': 'code_interpreter',
            'description_for_model': '代码解释器，可用于执行Python代码。'
            + ' Format the arguments as a JSON object.',
            'parameters': [{'name': 'code', 'type': 'string', 'description': '待执行的代码'}]
        },
        {
            'name_for_human': '文生图',
            'name_for_model': 'image_gen',
            'description_for_model': '文生图是一个AI绘画（图像生成）服务，输入文本描述，返回根据文本作画得到的图片的URL。'
            + ' Format the arguments as a JSON object.',
            'parameters': [
                {
                    'name': 'prompt',
                    'description': '英文关键词，描述了希望图像具有什么内容',
                    'required': True,
                    'schema': {'type': 'string'},
                }
            ],
        },
    ]


def call_plugin(plugin_name: str, plugin_args: str) -> str:
    if plugin_name == 'code_interpreter':
        return code_interpreter(plugin_args)
    elif plugin_name == 'image_gen':
        return image_gen(plugin_args)
    else:
        raise NotImplementedError

qwen_agent/tools/tools.py
QwenLM/Qwen-Agent
import json
import re
import sys
import traceback

import jieba
import json5
import requests
from jieba import analyse

import tiktoken


def print_traceback():
    print(''.join(traceback.format_exception(*sys.exc_info())))


def save_text_to_file(path, text):
    try:
        with open(path, 'w', encoding='utf-8') as fp:
            fp.write(text)
        return 'SUCCESS'
    except Exception as ex:
        print_traceback()
        return ex


def count_tokens(text):
    encoding = tiktoken.get_encoding('cl100k_base')
    tokens = encoding.encode(text)
    return len(tokens)


def gen_rec(text):
    return 'text'


def get_html_content(url):
    response = requests.get(url)
    html_content = response.text
    return html_content


def send_msg(url, msg):
    return requests.post(url, params=msg)


def get_split_word(text):
    text = text.lower()
    _wordlist = jieba.lcut(text.strip())
    wordlist = []
    for x in _wordlist:
        if x not in [' ', '  ', '\t', '\n', '\\', 'is', 'are', 'what', 'how', '的', '吗', '是', '了', '怎么', '如何', '什么', '？', '?', '!']:
            wordlist.append(x)
    # print('wordlist: ', wordlist)
    return wordlist


def get_key_word(text):
    text = text.lower()
    _wordlist = analyse.extract_tags(text)
    wordlist = []
    for x in _wordlist:
        if x not in [' ', '  ', '\t', '\n', '\\', 'is', 'are', 'what', 'how', '的', '吗', '是', '了', '怎么', '如何', '什么', '？', '?', '!']:
            wordlist.append(x)
    print('wordlist: ', wordlist)
    return wordlist


def get_last_one_line_context(text):
    lines = text.split('\n')
    n = len(lines)
    for i in range(n-1, -1, -1):
        if lines[i].strip():
            res = lines[i]
            break
    return res


def extract_urls(text):
    pattern = re.compile(r'https?://\S+')
    urls = re.findall(pattern, text)
    return urls


def extract_obs(text):
    k = text.rfind('\nObservation:')
    j = text.rfind('\nThought:')
    obs = text[k+len('\nObservation:'):j]
    return obs.strip()


def extract_code(text):
    # Match triple backtick blocks first
    triple_match = re.search(r'```[^\n]*\n(.+?)```', text, re.DOTALL)
    if triple_match:
        text = triple_match.group(1)
    else:
        try:
            text = json5.loads(text)['code']
        except Exception:
            pass
    # If no code blocks found, return original text
    return text


def parse_latest_plugin_call(text):
    plugin_name, plugin_args = '', ''
    i = text.rfind('\nAction:')
    j = text.rfind('\nAction Input:')
    k = text.rfind('\nObservation:')
    if 0 <= i < j:  # If the text has `Action` and `Action input`,
        if k < j:  # but does not contain `Observation`,
            # then it is likely that `Observation` is ommited by the LLM,
            # because the output text may have discarded the stop word.
            text = text.rstrip() + '\nObservation:'  # Add it back.
        k = text.rfind('\nObservation:')
        plugin_name = text[i + len('\nAction:'):j].strip()
        plugin_args = text[j + len('\nAction Input:'):k].strip()
        text = text[:k]
    return plugin_name, plugin_args, text


def format_answer(text):
    action, action_input, output = parse_latest_plugin_call(text)
    print('==format_answer==')
    print('action: ', action)
    print('action input: ', action_input)
    print('output: ',  output)
    if 'code_interpreter' in text:
        rsp = ''
        code = extract_code(action_input)
        rsp += ('\n```py\n' + code + '\n```\n')
        obs = extract_obs(text)
        if '![fig' in obs:
            rsp += obs
        return rsp
    elif 'image_gen' in text:
        # get url of FA
        # img_urls = URLExtract().find_urls(text.split("Final Answer:")[-1].strip())
        obs = text.split('Observation:')[-1].split('Final Answer')[0].strip()
        img_urls = []
        if obs:
            print(obs)
            try:
                obs = json.loads(obs)
                img_urls.append(obs['image_url'])
            except Exception:
                print_traceback()
                img_urls = []
        if not img_urls:
            img_urls = extract_urls(text.split('Final Answer:')[-1].strip())
        print(img_urls)
        rsp = ''
        for x in img_urls:
            rsp += '\n![picture]('+x.strip()+')'
        return rsp
    else:
        return text.split('Final Answer:')[-1].strip()

qwen_agent/utils/util.py
QwenLM/Qwen-Agent
import datetime
import importlib
import json
import os
import random
import shutil
import sys
from pathlib import Path

import gradio as gr
import jsonlines

import config_browserqwen

sys.path.insert(
    0,
    str(Path(__file__).absolute().parent.parent))  # NOQA

from qwen_agent.tools.tools import tools_list, call_plugin  # NOQA
from qwen_agent.utils.util import get_last_one_line_context, save_text_to_file, count_tokens, get_html_content, format_answer  # NOQA
from qwen_agent.actions import ContinueWriting, Simple, WriteFromZero, ReAct  # NOQA
from qwen_agent.memory import Memory  # NOQA
from qwen_agent.actions.func_call import func_call  # NOQA

prompt_lan = sys.argv[1]
llm_name = sys.argv[2]
max_ref_token = int(sys.argv[3])
workstation_port = int(sys.argv[4])
model_server = sys.argv[5]
api_key = sys.argv[6]
server_host = sys.argv[7]

if model_server.startswith('http'):
    source = 'local'
elif model_server.startswith('dashscope'):
    source = 'dashscope'

if llm_name.startswith('gpt'):
    module = 'qwen_agent.llm.gpt'
    llm = importlib.import_module(module).GPT(llm_name)
elif llm_name.startswith('Qwen') or llm_name.startswith('qwen'):
    module = 'qwen_agent.llm.qwen'
    llm = importlib.import_module(module).Qwen(llm_name, model_server=model_server, api_key=api_key)
else:
    raise NotImplementedError

mem = Memory(config_browserqwen.similarity_search, config_browserqwen.similarity_search_type)

app_global_para = {
    'time': [str(datetime.date.today()), str(datetime.date.today())],
    'cache_file': os.path.join(config_browserqwen.cache_root, config_browserqwen.browser_cache_file),
    'use_ci_flag': False,
    'messages': [],
    'last_turn_msg_id': [],
    'is_first_upload': True
}

with open(Path(__file__).resolve().parent / 'css/main.css', 'r') as f:
    css = f.read()
with open(Path(__file__).resolve().parent / 'js/main.js', 'r') as f:
    js = f.read()


def add_text(history, text):
    history = history + [(text, None)]
    app_global_para['last_turn_msg_id'] = []
    return history, gr.update(value='', interactive=False)


def rm_text(history):
    if not history:
        gr.Warning('No input content!')
    elif not history[-1][1]:
        return history, gr.update(value='', interactive=False)
    else:
        history = history[:-1] + [(history[-1][0], None)]
        return history, gr.update(value='', interactive=False)


def chat_clear():
    app_global_para['messages'] = []
    return None, None, None


def chat_clear_last():
    # print(app_global_para['last_turn_msg_id'][::-1])
    # print(app_global_para['messages'])
    for index in app_global_para['last_turn_msg_id'][::-1]:
        del app_global_para['messages'][index]
    app_global_para['last_turn_msg_id'] = []


def add_file(file):
    output_filepath = config_browserqwen.code_interpreter_ws
    fn = os.path.basename(file.name)
    new_path = os.path.join(output_filepath, fn)
    if os.path.exists(new_path):
        os.remove(new_path)
    shutil.move(file.name, output_filepath)
    path_show = '<p></p><p>File</p>'  # NOQA
    app_global_para['is_first_upload'] = True
    return path_show, new_path


def check_file():
    if not app_global_para['use_ci_flag']:
        gr.Warning('Select the Code Interpreter to analyze the file!')


def read_records(file, times=None):
    lines = []
    if times:
        for line in jsonlines.open(file):
            if times[0] <= line['time'] <= times[1]:
                lines.append(line)
    return lines


def update_rec_list(flag):
    rec_list = []
    if not os.path.exists(app_global_para['cache_file']):
        return 'No browsing records'
    lines = []
    for line in jsonlines.open(app_global_para['cache_file']):
        if (app_global_para['time'][0] <= line['time'] <= app_global_para['time'][1]) and line['checked']:
            if flag == 'load' and line['topic']:
                rec_list.append(line['topic'])
            else:
                agent = Simple(llm=llm, stream=False)
                page_number = len(line['raw'])
                index = random.randint(0, page_number-1)
                if prompt_lan == 'CN':
                    topicprompt = '请提出一个有新意的吸引人的话题'
                elif prompt_lan == 'EN':
                    topicprompt = 'Please propose a new and attractive topic'
                rec = agent.run(line['raw'][index]['page_content'], topicprompt, prompt_lan=prompt_lan)
                assert isinstance(rec, str)
                rec_list.append(rec)
                line['topic'] = rec

        lines.append(line)
    if lines:
        # update cache file
        with jsonlines.open(app_global_para['cache_file'], mode='w') as writer:
            for new_line in lines:
                writer.write(new_line)

    res = '<ol>{rec}</ol>'
    rec = ''
    for x in rec_list:
        rec += '<li>{rec_topic}</li>'.format(rec_topic=x)
    res = res.format(rec=rec)
    return res


def update_app_global_para(date1, date2):
    app_global_para['time'][0] = date1
    app_global_para['time'][1] = date2
    app_global_para['use_ci_flag'] = False


def update_browser_list():
    if not os.path.exists(app_global_para['cache_file']):
        return 'No browsing records'
    lines = read_records(app_global_para['cache_file'], times=app_global_para['time'])

    br_list = [[line['url'], line['extract'], line['checked']] for line in lines]
    print('browser_list: ', len(br_list))
    res = '<ol>{bl}</ol>'
    bl = ''
    for i, x in enumerate(br_list):
        ck = '<input type="checkbox" class="custom-checkbox" id="ck-'+x[0]+'" '
        if x[2]:
            ck += 'checked>'
        else:
            ck += '>'
        bl += '<li>{checkbox}{title}<a href="{url}"> [url]</a></li>'.format(checkbox=ck, url=x[0], title=x[1])
    res = res.format(bl=bl)
    return res


def update_all(date1, date2):
    update_app_global_para(date1, date2)
    return update_browser_list(), update_rec_list('update')


def layout_to_right(text):
    return text, text


def download_text(text):
    now = datetime.datetime.now()
    current_time = now.strftime('%Y-%m-%d_%H-%M-%S')
    filename = f'file_{current_time}.md'
    rsp = save_text_to_file(os.path.join(config_browserqwen.download_root, filename), text)
    if rsp == 'SUCCESS':
        raise gr.Info('Saved')
    else:
        raise gr.Error("Can't Save: ", rsp)


def count_token(text):
    return count_tokens(text)


def change_use_ci_flag():
    if app_global_para['use_ci_flag']:
        app_global_para['use_ci_flag'] = False
    else:
        app_global_para['use_ci_flag'] = True


def add_url_manu(url, date):
    text = get_html_content(url)
    msg = {'content': text, 'query': '', 'url': url, 'task': 'cache', 'type': 'html'}

    from main import cache_data  # NOQA
    cache_data(msg, app_global_para['cache_file'])


def bot(history, upload_file):
    if not history:
        yield history
    else:
        history[-1][1] = ''
        if app_global_para['use_ci_flag']:  # use code interpreter
            prompt_upload_file = ''
            if upload_file and app_global_para['is_first_upload']:
                workspace_dir = config_browserqwen.code_interpreter_ws
                file_relpath = os.path.relpath(path=upload_file, start=workspace_dir)
                if prompt_lan == 'EN':
                    prompt_upload_file = f'[Upload file {file_relpath}]'
                elif prompt_lan == 'CN':
                    prompt_upload_file = f'[上传文件{file_relpath}]'
                app_global_para['is_first_upload'] = False
            history[-1][0] = prompt_upload_file+history[-1][0]
            if source == 'local':  # using func call interface
                message = {
                    'role': 'user', 'content': history[-1][0]
                }
                app_global_para['last_turn_msg_id'].append(len(app_global_para['messages']))
                app_global_para['messages'].append(message)
                while True:
                    print(app_global_para['messages'])
                    functions = [x for x in tools_list if x['name_for_model'] == 'code_interpreter']
                    rsp = llm.qwen_chat_func(app_global_para['messages'], functions)
                    if rsp['function_call']:
                        history[-1][1] += rsp['content'].strip() + '\n'
                        yield history
                        history[-1][1] += 'Action: '+rsp['function_call']['name'].strip() + '\n'
                        yield history
                        history[-1][1] += 'Action Input:\n'+rsp['function_call']['arguments'] + '\n'
                        yield history
                        bot_msg = {
                            'role': 'assistant',
                            'content': rsp['content'],
                            'function_call': {
                                'name': rsp['function_call']['name'],
                                'arguments': rsp['function_call']['arguments'],
                            }
                        }
                        app_global_para['last_turn_msg_id'].append(len(app_global_para['messages']))
                        app_global_para['messages'].append(bot_msg)

                        obs = call_plugin(rsp['function_call']['name'], rsp['function_call']['arguments'])
                        func_msg = {
                            'role': 'function',
                            'name': rsp['function_call']['name'],
                            'content': obs,
                        }
                        history[-1][1] += ('Observation: ' + obs + '\n')
                        yield history
                        app_global_para['last_turn_msg_id'].append(len(app_global_para['messages']))
                        app_global_para['messages'].append(func_msg)
                    else:
                        bot_msg = {
                            'role': 'assistant',
                            'content': rsp['content'],
                        }
                        history[-1][1] += rsp['content']
                        yield history
                        app_global_para['last_turn_msg_id'].append(len(app_global_para['messages']))
                        app_global_para['messages'].append(bot_msg)
                        break
            elif source == 'dashscope':  # using dashscope interface:
                print(app_global_para['messages'])
                functions = [x for x in tools_list if x['name_for_model'] == 'code_interpreter']
                agent = ReAct(llm=llm, list_of_plugin_info=functions, source=source)
                response = agent.run(history[-1][0], messages=app_global_para['messages'])
                if 'Action' not in response:
                    response = response.split('Final Answer:')[-1]
                history[-1][1] = response
                yield history

                message = {
                    'role': 'user', 'content': history[-1][0]
                }
                app_global_para['last_turn_msg_id'].append(len(app_global_para['messages']))
                app_global_para['messages'].append(message)
                rsp_message = {
                    'role': 'assistant', 'content': history[-1][1]
                }
                app_global_para['last_turn_msg_id'].append(len(app_global_para['messages']))
                app_global_para['messages'].append(rsp_message)

        else:
            lines = []
            if not os.path.exists(app_global_para['cache_file']):
                _ref = ''
            else:
                for line in jsonlines.open(app_global_para['cache_file']):
                    if (app_global_para['time'][0] <= line['time'] <= app_global_para['time'][1]) and line['checked']:
                        lines.append(line)
                if lines:
                    _ref_list = mem.get(history[-1][0], lines, llm=llm, stream=True, max_token=max_ref_token)
                    _ref = '\n'.join(json.dumps(x, ensure_ascii=False) for x in _ref_list)
                else:
                    _ref = ''
                    gr.Warning('No reference materials selected, Qwen will answer directly')

            agent = Simple(llm=llm, stream=True)
            response = agent.run(_ref, history)

            for chunk in response:
                history[-1][1] += chunk
                yield history

            # append message
            message = {
                'role': 'user', 'content': history[-1][0]
            }
            app_global_para['last_turn_msg_id'].append(len(app_global_para['messages']))
            app_global_para['messages'].append(message)

            message = {
                'role': 'assistant', 'content': history[-1][1]
            }
            app_global_para['last_turn_msg_id'].append(len(app_global_para['messages']))
            app_global_para['messages'].append(message)


def generate(context):
    sp_query = get_last_one_line_context(context)
    res = ''
    if config_browserqwen.code_flag in sp_query:  # router to code interpreter
        sp_query = sp_query.split(config_browserqwen.code_flag)[-1]
        sp_query += ', 必须使用code_interpreter工具'
        functions = [x for x in tools_list if x['name_for_model'] == 'code_interpreter']
        if source == 'local':  # using func call interface
            response = func_call(sp_query, functions, llm)
            for chunk in response:
                res += chunk
                yield res
        elif source == 'dashscope':  # using dashscope interface
            agent = ReAct(llm=llm, list_of_plugin_info=functions, source=source)
            response = agent.run(sp_query, messages=[])
            yield response
    elif config_browserqwen.plugin_flag in sp_query:  # router to plugin
        sp_query = sp_query.split(config_browserqwen.plugin_flag)[-1]
        functions = tools_list
        if source == 'local':  # using func call interface
            response = func_call(sp_query, functions, llm)
            for chunk in response:
                res += chunk
                yield res
        elif source == 'dashscope':  # using dashscope interface
            agent = ReAct(llm=llm, list_of_plugin_info=functions, source=source)
            response = agent.run(sp_query, messages=[])
            yield response
    else:  # router to continue writing
        lines = []
        for line in jsonlines.open(app_global_para['cache_file']):
            if (app_global_para['time'][0] <= line['time'] <= app_global_para['time'][1]) and line['checked']:
                lines.append(line)
        if lines:
            if config_browserqwen.similarity_search:
                res += '\n========================= \n'
                yield res
                res += '> Search for relevant information: \n'
                yield res
            sp_query_no_title = sp_query
            if config_browserqwen.title_flag in sp_query:  # /title
                sp_query_no_title = sp_query.split(config_browserqwen.title_flag)[-1]

            _ref_list = mem.get(sp_query_no_title, lines, llm=llm, stream=True, max_token=max_ref_token)
            _ref = '\n'.join(json.dumps(x, ensure_ascii=False) for x in _ref_list)
            res += _ref
            yield res
            res += '\n'
        else:
            _ref = ''
            gr.Warning('No reference materials selected, Qwen will answer directly')

        if config_browserqwen.title_flag in sp_query:  # /title
            sp_query = sp_query.split(config_browserqwen.title_flag)[-1]
            agent = WriteFromZero(llm=llm, stream=True, auto_agent=config_browserqwen.auto_agent)
        else:
            res += '\n========================= \n'
            res += '> Writing Text: \n'
            yield res
            agent = ContinueWriting(llm=llm, stream=True)

        response = agent.run(_ref, context, prompt_lan=prompt_lan)
        for chunk in response:
            res += chunk
            yield res
        print('OK!')


def format_generate(edit, context):
    res = edit
    yield res
    if '> Writing Text: ' in context:
        text = context.split('> Writing Text: ')[-1].strip()
        res += '\n'
        res += text
        yield res
    elif 'Final Answer' in context:
        response = format_answer(context)
        res += '\n'
        res += response
        yield res
    else:
        res += context
        yield res


with gr.Blocks(css=css, theme='soft') as demo:
    title = gr.Markdown('Qwen Agent: BrowserQwen', elem_classes='title')
    desc = gr.Markdown(
        'This is the editing workstation of BrowserQwen, where Qwen has collected the browsing history. Qwen can assist you in completing your creative work!',
        elem_classes='desc')

    with gr.Row():
        with gr.Column():
            rec = gr.Markdown('Browsing History', elem_classes='rec')
            with gr.Row():
                with gr.Column(scale=0.3, min_width=0):
                    date1 = gr.Dropdown([str(datetime.date.today()-datetime.timedelta(days=i)) for i in range(
                        config_browserqwen.max_days)], value=str(datetime.date.today()), label='Start Date')  # NOQA
                    date2 = gr.Dropdown([str(datetime.date.today()-datetime.timedelta(days=i)) for i in range(
                        config_browserqwen.max_days)], value=str(datetime.date.today()), label='End Date')  # NOQA
                with gr.Column(scale=0.7, min_width=0):
                    browser_list = gr.HTML(value='', label='browser_list', elem_classes=['div_tmp', 'add_scrollbar'])

    with gr.Tab('Editor', elem_id='default-tab'):
        with gr.Row():
            with gr.Column():
                with gr.Row():
                    edit_area = gr.Textbox(
                        value='',
                        elem_classes=['textbox_default', 'add_scrollbar'],
                        lines=30,
                        label='Input',
                        show_copy_button=True)
                    # token_count = gr.HTML(value='<span>0</span>',
                    #                       elem_classes=[
                    #                           'token-counter',
                    #                           'default-token-counter'
                    #                       ])

                with gr.Row():
                    ctn_bt = gr.Button('Continue', variant='primary')
                    stop_bt = gr.Button('Stop')
                    clr_bt = gr.Button('Clear')
                    dld_bt = gr.Button('Download')

                # with gr.Row():
                #     layout_bt = gr.Button('👉', variant='primary')

            with gr.Column():
                cmd_area = gr.Textbox(lines=10, max_lines=10, label="Qwen's Inner Thought", elem_id='cmd')
                with gr.Tab('Markdown'):
                    # md_out_bt = gr.Button('Render')
                    md_out_area = gr.Markdown(elem_classes=[
                                                   'md_tmp',
                                                   'add_scrollbar'
                                               ])

                with gr.Tab('HTML'):
                    html_out_area = gr.HTML()

                with gr.Tab('Raw'):
                    text_out_area = gr.Textbox(lines=20,
                                               label='',
                                               elem_classes=[
                                                   'textbox_default_output',
                                                   'add_scrollbar'
                                               ],
                                               show_copy_button=True)
        # br_input_bt.click(add_url_manu, br_input, None).then(update_browser_list, None, browser_list).then(lambda: None, None, None, _js=f'() => {{{js}}}')
        # .then(update_rec_list, gr.Textbox('update', visible=False), rec_list)
        # clk_ctn_bt = ctn_bt.click(qwen_ctn, edit_area, edit_area)
        clk_ctn_bt = ctn_bt.click(generate, edit_area, cmd_area)
        clk_ctn_bt.then(format_generate, [edit_area, cmd_area], edit_area)

        edit_area_change = edit_area.change(layout_to_right, edit_area, [text_out_area, md_out_area])
        # edit_area_change.then(count_token, edit_area, token_count)

        stop_bt.click(lambda: None, cancels=[clk_ctn_bt], queue=False)
        clr_bt.click(lambda: [None, None, None], None, [edit_area, cmd_area, md_out_area], queue=False)
        dld_bt.click(download_text, edit_area, None)

        # layout_bt.click(layout_to_right,
        #                 edit_area, [text_out_area, md_out_area],
        #                 queue=False)
        gr.Markdown('''
    ### Usage Tips:
    - Browsing History:
        - Start Date/End Date: Selecting the browsed materials for the desired time period, including the start and end dates
        - The browsed materials list: supporting the selection or removal of specific browsing content
    - Editor: In the editing area, you can directly input content or special instructions, and then click the ```Continue``` button to have Qwen assist in completing the editing work:
        - After inputting the content, directly click the ```Continue``` button: Qwen will begin to continue writing based on the browsing information
        - Using special instructions:
            - /title + content: Qwen enables the built-in planning process and writes a complete manuscript
            - /code + content: Qwen enables the code interpreter plugin, writes and runs Python code, and generates replies
            - /plug + content: Qwen enables plugin and select appropriate plugin to generate reply
    - Chat: Interactive area. Qwen generates replies based on given reference materials. Selecting Code Interpreter will enable the code interpreter plugin

        ''')

    with gr.Tab('Chat', elem_id='chat-tab'):
        with gr.Column():

            chatbot = gr.Chatbot([],
                                 elem_id='chatbot',
                                 height=680,
                                 show_copy_button=True,
                                 avatar_images=(None, (os.path.join(
                                     Path(__file__).resolve().parent, 'img/logo.png'))))
            with gr.Row():
                with gr.Column(scale=0.05, min_width=0):
                    file_btn = gr.UploadButton('Upload', file_types=['file'])  # NOQA
                with gr.Column(scale=0.16, min_width=0):
                    plug_bt = gr.Checkbox(label='Code Interpreter')
                with gr.Column(scale=0.02, min_width=0):
                    show_path_md = gr.HTML('')
                # with gr.Column(scale=0.03, min_width=0):
                #     show_path_bt = gr.Button('✖️', visible=False)  # NOQA
                with gr.Column(scale=0.65):
                    chat_txt = gr.Textbox(show_label=False, placeholder='Chat with Qwen...', container=False)  # NOQA
                # with gr.Column(scale=0.05, min_width=0):
                #     chat_smt_bt = gr.Button('⏎')  # NOQA
                with gr.Column(scale=0.05, min_width=0):
                    chat_clr_bt = gr.Button('Clear')  # NOQA

                with gr.Column(scale=0.05, min_width=0):
                    chat_stop_bt = gr.Button('Stop')  # NOQA
                with gr.Column(scale=0.05, min_width=0):
                    chat_re_bt = gr.Button('Again')  # NOQA

            hidden_file_path = gr.Textbox(visible=False)

            txt_msg = chat_txt.submit(add_text, [chatbot, chat_txt], [chatbot, chat_txt], queue=False).then(bot, [chatbot, hidden_file_path], chatbot)
            txt_msg.then(lambda: gr.update(interactive=True), None, [chat_txt], queue=False)

            # txt_msg_bt = chat_smt_bt.click(add_text, [chatbot, chat_txt], [chatbot, chat_txt], queue=False).then(bot, chatbot, chatbot)
            # txt_msg_bt.then(lambda: gr.update(interactive=True), None, [chat_txt], queue=False)
            # (None, None, None, cancels=[txt_msg], queue=False).then
            re_txt_msg = chat_re_bt.click(rm_text, [chatbot], [chatbot, chat_txt], queue=False).then(chat_clear_last, None, None).then(bot, [chatbot, hidden_file_path], chatbot)
            re_txt_msg.then(lambda: gr.update(interactive=True), None, [chat_txt], queue=False)

            file_msg = file_btn.upload(add_file, [file_btn], [show_path_md, hidden_file_path], queue=False).then(check_file, None, None)

            chat_clr_bt.click(chat_clear, None, [chatbot, hidden_file_path, show_path_md], queue=False)
            # re_bt.click(re_bot, chatbot, chatbot)
            chat_stop_bt.click(chat_clear_last, None, None, cancels=[txt_msg, re_txt_msg], queue=False)

            # show_path_bt.click(lambda x:[None, gr.update(visible=False)], hidden_file_path, [show_path_md,show_path_bt])

            plug_bt.change(change_use_ci_flag)

    # img_cloud = gr.Image(height='50px')
    # update_bt.click(update_rec_list, gr.Textbox('update', visible=False), rec_list, queue=False)

    # date.change(update_all, date,  [img_cloud, browser_list, rec_list], queue=False)
    date1.change(update_app_global_para, [date1, date2], None).then(update_browser_list, None, browser_list).then(lambda: None, None, None, _js=f'() => {{{js}}}').then(chat_clear, None, [chatbot, hidden_file_path, show_path_md])
    date2.change(update_app_global_para, [date1, date2], None).then(update_browser_list, None, browser_list).then(lambda: None, None, None, _js=f'() => {{{js}}}').then(chat_clear, None, [chatbot, hidden_file_path, show_path_md])

    # demo.load(update_all, date,  [img_cloud, browser_list, rec_list])
    demo.load(update_app_global_para, [date1, date2], None).then(update_browser_list, None, browser_list).then(lambda: None, None, None, _js=f'() => {{{js}}}').then(chat_clear, None, [chatbot, hidden_file_path, show_path_md])
    # .then(update_rec_list, gr.Textbox('load', visible=False), rec_list, queue=False)

demo.queue().launch(server_name=server_host, server_port=workstation_port)

qwen_server/app.py
QwenLM/Qwen-Agent
import importlib
import json
import os
import sys
from pathlib import Path

import gradio as gr
import jsonlines

import config_browserqwen

sys.path.insert(
    0,
    str(Path(__file__).absolute().parent.parent))  # NOQA

from qwen_agent.actions import Simple  # NOQA
from qwen_agent.memory import Memory  # NOQA

prompt_lan = sys.argv[1]
llm_name = sys.argv[2]
max_ref_token = int(sys.argv[3])
model_server = sys.argv[4]
api_key = sys.argv[5]
server_host = sys.argv[6]

if llm_name.startswith('gpt'):
    module = 'qwen_agent.llm.gpt'
    llm = importlib.import_module(module).GPT(llm_name)
elif llm_name.startswith('Qwen') or llm_name.startswith('qwen'):
    module = 'qwen_agent.llm.qwen'
    llm = importlib.import_module(module).Qwen(llm_name, model_server=model_server, api_key=api_key)
else:
    raise NotImplementedError

mem = Memory(config_browserqwen.similarity_search, config_browserqwen.similarity_search_type)

cache_file = os.path.join(config_browserqwen.cache_root, config_browserqwen.browser_cache_file)
cache_file_popup_url = os.path.join(config_browserqwen.cache_root, config_browserqwen.url_file)

page_url = []

with open(Path(__file__).resolve().parent / 'css/main.css', 'r') as f:
    css = f.read()
with open(Path(__file__).resolve().parent / 'js/main.js', 'r') as f:
    js = f.read()


def add_text(history, text):
    history = history + [(text, None)]
    return history, gr.update(value='', interactive=False)


def rm_text(history):
    if not history:
        gr.Warning('No input content!')
    elif not history[-1][1]:
        return history, gr.update(value='', interactive=False)
    else:
        history = history[:-1] + [(history[-1][0], None)]
        return history, gr.update(value='', interactive=False)


def add_file(history, file):
    history = history + [((file.name, ), None)]
    return history


def set_page_url():
    lines = []
    assert os.path.exists(cache_file_popup_url)
    for line in jsonlines.open(cache_file_popup_url):
        lines.append(line)
    page_url.append(lines[-1]['url'])
    print('now page url is: ', page_url[-1])


def bot(history):
    if not history:
        yield history
    else:
        now_page = None
        _ref = ''
        if not os.path.exists(cache_file):
            gr.Info("Please add this page to Qwen's Reading List first!")
        else:
            for line in jsonlines.open(cache_file):
                if line['url'] == page_url[-1]:
                    now_page = line

            if not now_page:
                gr.Info("This page has not yet been added to the Qwen's reading list!")
            elif not now_page['raw']:
                gr.Info('Please wait, Qwen is analyzing this page...')
            else:
                _ref_list = mem.get(history[-1][0], [now_page], llm=llm, stream=False, max_token=max_ref_token)
                if _ref_list:
                    _ref = '\n'.join(json.dumps(x, ensure_ascii=False) for x in _ref_list)
                else:
                    _ref = ''
        # print(_ref)
        agent = Simple(stream=True, llm=llm)
        history[-1][1] = ''
        response = agent.run(_ref, history, prompt_lan=prompt_lan)

        for chunk in response:
            history[-1][1] += chunk
            yield history

        # save history
        if now_page:
            now_page['session'] = history
            lines = []
            for line in jsonlines.open(cache_file):
                if line['url'] != page_url[-1]:
                    lines.append(line)

            lines.append(now_page)
            with jsonlines.open(cache_file, mode='w') as writer:
                for new_line in lines:
                    writer.write(new_line)


def load_history_session(history):
    now_page = None
    if not os.path.exists(cache_file):
        gr.Info("Please add this page to Qwen's Reading List first!")
        return []
    for line in jsonlines.open(cache_file):
        if line['url'] == page_url[-1]:
            now_page = line
    if not now_page:
        gr.Info("Please add this page to Qwen's Reading List first!")
        return []
    if not now_page['raw']:
        gr.Info('Please wait, Qwen is analyzing this page...')
        return []
    return now_page['session']


def clear_session():
    if not os.path.exists(cache_file):
        return None
    now_page = None
    lines = []
    for line in jsonlines.open(cache_file):
        if line['url'] == page_url[-1]:
            now_page = line
        else:
            lines.append(line)
    if not now_page:
        return None
    now_page['session'] = []
    lines.append(now_page)
    with jsonlines.open(cache_file, mode='w') as writer:
        for new_line in lines:
            writer.write(new_line)

    return None


with gr.Blocks(css=css, theme='soft') as demo:
    chatbot = gr.Chatbot([],
                         elem_id='chatbot',
                         height=480,
                         avatar_images=(None, (os.path.join(
                             Path(__file__).resolve().parent, 'img/logo.png'))))
    with gr.Row():
        with gr.Column(scale=0.7):
            txt = gr.Textbox(show_label=False,
                             placeholder='Chat with Qwen...',
                             container=False)
        # with gr.Column(scale=0.06, min_width=0):
        #     smt_bt = gr.Button('⏎')
        with gr.Column(scale=0.1, min_width=0):
            clr_bt = gr.Button('🧹', elem_classes='bt_small_font')
        with gr.Column(scale=0.1, min_width=0):
            stop_bt = gr.Button('🚫', elem_classes='bt_small_font')
        with gr.Column(scale=0.1, min_width=0):
            re_bt = gr.Button('🔁', elem_classes='bt_small_font')

    txt_msg = txt.submit(add_text, [chatbot, txt], [chatbot, txt],
                         queue=False).then(bot, chatbot, chatbot)
    txt_msg.then(lambda: gr.update(interactive=True), None, [txt], queue=False)

    # txt_msg_bt = smt_bt.click(add_text, [chatbot, txt], [chatbot, txt],
    #                           queue=False).then(bot, chatbot, chatbot)
    # txt_msg_bt.then(lambda: gr.update(interactive=True),
    #                 None, [txt],
    #                 queue=False)

    clr_bt.click(clear_session, None, chatbot, queue=False)
    re_txt_msg = re_bt.click(rm_text, [chatbot], [chatbot, txt], queue=False).then(bot, chatbot, chatbot)
    re_txt_msg.then(lambda: gr.update(interactive=True), None, [txt], queue=False)

    stop_bt.click(None, None, None, cancels=[txt_msg, re_txt_msg], queue=False)

    demo.load(set_page_url).then(load_history_session, chatbot, chatbot)

demo.queue().launch(server_name=server_host, server_port=config_browserqwen.app_in_browser_port)

qwen_server/app_in_browser.py
QwenLM/Qwen-Agent
import os
from pathlib import Path

""" ===== global setting ===== """
# the path of browser infomation
work_space_root = os.path.join(Path(__file__).parent.parent, 'workspace/')
cache_root = os.path.join(work_space_root, 'browser_cache/')
download_root = os.path.join(work_space_root, 'download/')
browser_cache_file = 'browse.jsonl'
url_file = 'popup_url.jsonl'
# the path of workspace of code interpreter
code_interpreter_ws = os.path.join(work_space_root, 'ci_workspace/')

MAX_TOKEN = 4000  # the max token number of reference material
llm = 'Qwen-7B-Chat'  # ['Qwen-7B-Chat', 'gpt-4'] - the llm for using
prompt_lan = 'EN'  # ['CN', 'EN'] - the language of built-in prompt

# using similarity search on reference material before answer
similarity_search = True  # [True, False]
similarity_search_type = 'jaccard'  # ['keyword', 'querymatch', 'llm', 'jaccard']


""" ===== main.py setting ===== """
# the port of main.py
fast_api_port = 7866

figure_host = '127.0.0.1'
fast_api_figure_url = 'http://'+figure_host+':'+str(fast_api_port)+'/static'
address_file = os.path.join(work_space_root, 'address_file.json')

pre_gen_question = False  # [True, False] - pre gen qustion for each block

""" ===== app.py setting (editing workstation) ===== """

max_days = 7  # the number of days for displaying
auto_agent = False  # [True, False] - automatic using plug-in after wrinting every section

# special instructions in editing
plugin_flag = '/plug'  # using plug-in
code_flag = '/code'  # using code interpreter
title_flag = '/title'  # writing a full article with planning


""" ===== app_in_browser.py setting (browser interactive interfaces) ===== """
app_in_browser_port = 7863

qwen_server/config_browserqwen.py
QwenLM/Qwen-Agent
import datetime
import importlib
import multiprocessing
import os
import re
import sys
from pathlib import Path
from urllib.parse import unquote, urlparse

import jsonlines
import requests
import uvicorn
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles

import config_browserqwen

sys.path.insert(
    0,
    str(Path(__file__).absolute().parent.parent))  # NOQA

from qwen_agent.actions import Simple  # NOQA
from qwen_agent.tools.parse_doc import parse_pdf_pypdf, parse_html_bs  # NOQA
from qwen_agent.utils.util import print_traceback, save_text_to_file  # NOQA
from qwen_agent.schema import Record  # NOQA

prompt_lan = sys.argv[1]
llm_name = sys.argv[2]
max_ref_token = int(sys.argv[3])
workstation_port = int(sys.argv[4])
model_server = sys.argv[5]
api_key = sys.argv[6]
server_host = sys.argv[7]

app = FastAPI()

origins = [
    'http://127.0.0.1:'+str(workstation_port),
    'http://localhost:'+str(workstation_port),
    'http://0.0.0.0:'+str(workstation_port),
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=['*'],
    allow_headers=['*'],
)

app.mount('/static', StaticFiles(directory=config_browserqwen.code_interpreter_ws), name='static')

if llm_name.startswith('gpt'):
    module = 'qwen_agent.llm.gpt'
    llm = importlib.import_module(module).GPT(llm_name)
elif llm_name.startswith('Qwen') or llm_name.startswith('qwen'):
    module = 'qwen_agent.llm.qwen'
    llm = importlib.import_module(module).Qwen(llm_name, model_server=model_server, api_key=api_key)
else:
    raise NotImplementedError


def update_pop_url(data, cache_file_popup_url):
    new_line = {'url': data['url']}

    with jsonlines.open(cache_file_popup_url, mode='w') as writer:
        writer.write(new_line)

    response = 'Update URL'
    return response


def is_local_path(path):
    if path.startswith('file://'):
        return True
    else:
        return False


def get_title(text, cacheprompt=''):
    agent = Simple(llm=llm, stream=False)
    extract = agent.run(text, cacheprompt)

    return extract


def download_pdf(url, save_path):
    response = requests.get(url)
    with open(save_path, 'wb') as file:
        file.write(response.content)


def sanitize_chrome_file_path(file_path: str) -> str:
    # For Linux and macOS.
    if os.path.exists(file_path):
        return file_path

    # For native Windows, drop the leading '/' in '/C:/'
    win_path = file_path
    if win_path.startswith('/'):
        win_path = win_path[1:]
    if os.path.exists(win_path):
        return win_path

    # For Windows + WSL.
    if re.match(r'^[A-Za-z]:/', win_path):
        wsl_path = f'/mnt/{win_path[0].lower()}/{win_path[3:]}'
        if os.path.exists(wsl_path):
            return wsl_path

    # For native Windows, replace / with \.
    win_path = win_path.replace('/', '\\')
    if os.path.exists(win_path):
        return win_path

    return file_path


def cache_data(data, cache_file):
    extract = ''  # extract a title for display
    print('Begin cache...')
    if data['url'][-4:] in ['.pdf', '.PDF']:
        date1 = datetime.datetime.now()

        # generate one processing record
        new_record = Record(url=data['url'], time='', type=data['type'], raw=[],
                            extract='', topic='', checked=False, session=[]).to_dict()
        with jsonlines.open(cache_file, mode='a') as writer:
            writer.write(new_record)

        # deal pdf path
        if is_local_path(data['url']):
            parsed_url = urlparse(data['url'])
            print('parsed_url: ', parsed_url)
            pdf_path = unquote(parsed_url.path)
            pdf_path = sanitize_chrome_file_path(pdf_path)
        else:
            pdf_path = data['url']

        try:
            pdf_content = parse_pdf_pypdf(pdf_path, pre_gen_question=config_browserqwen.pre_gen_question)
        except Exception:
            print_traceback()
            # del the processing record
            lines = []
            if os.path.exists(cache_file):
                for line in jsonlines.open(cache_file):
                    if line['url'] != data['url']:
                        lines.append(line)
            with jsonlines.open(cache_file, mode='w') as writer:
                for new_line in lines:
                    writer.write(new_line)
            return 'failed'

        date2 = datetime.datetime.now()
        print('parse pdf time: ', date2 - date1)
        data['content'] = pdf_content
        data['type'] = 'pdf'
        if prompt_lan == 'CN':
            cacheprompt = '参考资料是一篇论文的首页，请提取出一句话作为标题。'
        elif prompt_lan == 'EN':
            cacheprompt = 'The reference material is the first page of a paper. Please extract one sentence as the title'
        extract = get_title(pdf_content[0]['page_content'], cacheprompt=cacheprompt)
    else:
        if data['content'] and data['type'] == 'html':
            new_record = Record(url=data['url'], time='', type=data['type'], raw=[], extract='', topic='', checked=False, session=[]).to_dict()
            with jsonlines.open(cache_file, mode='a') as writer:
                writer.write(new_record)

            try:
                tmp_html_file = os.path.join(config_browserqwen.cache_root, 'tmp.html')
                save_text_to_file(tmp_html_file, data['content'])
                data['content'] = parse_html_bs(tmp_html_file, pre_gen_question=config_browserqwen.pre_gen_question)
            except Exception:
                print_traceback()
            extract = data['content'][0]['metadata']['title']

    today = datetime.date.today()
    new_record = Record(url=data['url'], time=str(today), type=data['type'], raw=data['content'],
                        extract=extract, topic='', checked=True, session=[])
    lines = []
    if os.path.exists(cache_file):
        for line in jsonlines.open(cache_file):
            if line['url'] != data['url']:
                lines.append(line)
    lines.append(new_record.to_dict())  # cache
    with jsonlines.open(cache_file, mode='w') as writer:
        for new_line in lines:
            writer.write(new_line)

    response = 'Cached'
    return response


def change_checkbox_state(text, cache_file):
    if not os.path.exists(cache_file):
        return {'result': 'no file'}
    lines = []
    for line in jsonlines.open(cache_file):
        if line['url'] == text[3:]:
            if line['checked']:
                line['checked'] = False
            else:
                line['checked'] = True
        lines.append(line)

    with jsonlines.open(cache_file, mode='w') as writer:
        for new_line in lines:
            writer.write(new_line)

    return {'result': 'changed'}


def update_addr_for_figure(address):
    new_line = {'address': address}

    with jsonlines.open(config_browserqwen.address_file, mode='w') as writer:
        writer.write(new_line)

    response = 'Update Address'
    print('Update Address')
    return response


@app.post('/endpoint')
async def web_listening(request: Request):
    data = await request.json()
    msg_type = data['task']

    cache_file_popup_url = os.path.join(config_browserqwen.cache_root, config_browserqwen.url_file)
    cache_file = os.path.join(config_browserqwen.cache_root, config_browserqwen.browser_cache_file)

    if msg_type == 'change_checkbox':
        rsp = change_checkbox_state(data['ckid'], cache_file)
    elif msg_type == 'cache':
        cache_obj = multiprocessing.Process(target=cache_data, args=(data, cache_file))
        cache_obj.start()
        # rsp = cache_data(data, cache_file)
        rsp = 'caching'
    elif msg_type == 'pop_url':
        # What a misleading name! pop_url actually means add_url. pop is referring to the pop_up ui.
        rsp = update_pop_url(data, cache_file_popup_url)
    elif msg_type == 'set_addr':
        rsp = update_addr_for_figure(data['addr'])

    return JSONResponse(content=rsp)


if __name__ == '__main__':
    uvicorn.run(app='main:app', host=server_host, port=config_browserqwen.fast_api_port, reload=True)

qwen_server/main.py
QwenLM/Qwen-Agent
import argparse
import os
import signal
import stat
import subprocess
import sys

from qwen_server import config_browserqwen


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument('-m', '--model_server', type=str, default='dashscope')
    parser.add_argument('-k', '--api_key', type=str, default='')
    parser.add_argument('-l', '--llm', type=str, default='qwen-turbo',
                        choices=['qwen-plus', 'qwen-turbo', 'qwen-14b-chat', 'qwen-7b-chat'])
    parser.add_argument('-s', '--server_host', type=str, default='127.0.0.1')
    parser.add_argument('-lan', '--prompt_language', type=str, default='CN', choices=['EN', 'CN'],
                        help='the language of built-in prompt')  # TODO: auto detect based on query and ref
    parser.add_argument('-t', '--max_ref_token', type=int, default=4000,
                        help='the max token number of reference material')
    parser.add_argument('-w', '--workstation_port', type=int, default=7864,
                        help='the port of editing workstation')
    args = parser.parse_args()
    args.model_server = args.model_server.replace('0.0.0.0', '127.0.0.1')
    return args


def _fix_secure_write_for_code_interpreter():
    if 'linux' in sys.platform.lower():
        fname = os.path.join(config_browserqwen.code_interpreter_ws, 'test_file_permission.txt')
        if os.path.exists(fname):
            os.remove(fname)
        with os.fdopen(os.open(fname, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o0600), 'w') as f:
            f.write('test')
        file_mode = stat.S_IMODE(os.stat(fname).st_mode) & 0o6677
        if file_mode != 0o0600:
            os.environ['JUPYTER_ALLOW_INSECURE_WRITES'] = '1'


if __name__ == '__main__':
    args = parse_args()

    if not os.path.exists(config_browserqwen.work_space_root):
        os.makedirs(config_browserqwen.work_space_root)
    if not os.path.exists(config_browserqwen.cache_root):
        os.makedirs(config_browserqwen.cache_root)
    if not os.path.exists(config_browserqwen.download_root):
        os.makedirs(config_browserqwen.download_root)
    if not os.path.exists(config_browserqwen.code_interpreter_ws):
        os.makedirs(config_browserqwen.code_interpreter_ws)
    _fix_secure_write_for_code_interpreter()

    servers = {
        'database': subprocess.Popen(
            [sys.executable, os.path.join(os.getcwd(), 'qwen_server/main.py'), args.prompt_language, args.llm,
             str(args.max_ref_token), str(args.workstation_port), args.model_server, args.api_key, args.server_host]),
        'workstation': subprocess.Popen(
            [sys.executable, os.path.join(os.getcwd(), 'qwen_server/app.py'), args.prompt_language, args.llm,
             str(args.max_ref_token), str(args.workstation_port), args.model_server, args.api_key, args.server_host]),
        'browser': subprocess.Popen(
            [sys.executable, os.path.join(os.getcwd(), 'qwen_server/app_in_browser.py'), args.prompt_language,
             args.llm, str(args.max_ref_token), args.model_server, args.api_key, args.server_host])
    }

    def signal_handler(_sig, _frame):
        for v in servers.values():
            v.terminate()
        for k in list(servers.keys()):
            del servers[k]

    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)

    for p in list(servers.values()):
        p.wait()

run_server.py
QwenLM/Qwen-Agent