python+openai+tavily实现ai分步骤深度研究deepresearch前后端代码-python代码-BFW代码库

python+openai+tavily实现ai分步骤深度研究deepresearch前后端代码

代码语言：python

代码描述：python+openai+tavily实现ai分步骤深度研究deepresearch前后端代码，这里使用了兼容openai的qwen系统api与模型，改成兼容openai的其他模型即可。

代码标签： python openai tavily ai 分步骤深度研究 deep research 前

下面为部分代码预览，完整代码请点击下载或在bfwstudio webide中打开

#!/usr/local/python3/bin/python3
# -*- coding: utf-8 -*
import asyncio
import os
import json
# 移除 playwright 导入: from playwright.async_api import async_playwright, Page
import requests # 添加 requests 导入
from openai import AsyncOpenAI
from dotenv import load_dotenv
import re # 用于清理文本
import uuid # 用于生成唯一目录名
import datetime # 用于时间戳

# 加载环境变量
load_dotenv()


# 从环境变量获取API Key
API_KEY = "sk-"#os.getenv("OPENAI_API_KEY")
BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"#os.getenv("OPENAI_BASE_URL") # 新增：获取 Base URL
TAVILY_API_KEY ="tvly-"#os.getenv("TAVILY_API_KEY") # 新增：获取 Tavily Key



# 初始化 OpenAI 客户端
client = AsyncOpenAI(base_url=BASE_URL,api_key=API_KEY)
TEXT_MODEL = "qwen-max" # 或根据需要选用其他模型，如 gpt-3.5-turbo

# --- 1. 任务分解函数 (保持不变) ---
async def decompose_research_topic(topic: str) -> list[str] | None:
    # ... existing code ...
    print(f"--- 正在分解研究主题: {topic} ---")
    prompt_messages = [
        {
            "role": "system",
            "content": """
            你是一个研究规划助手。请将用户提供的复杂研究主题或问题分解成一系列逻辑清晰、可按顺序执行的子任务或子问题。
            这些子任务应有助于逐步深入理解主题。请以 JSON 格式返回结果，包含一个名为 "sub_tasks" 的数组，数组中的每个元素是一个描述子任务的字符串。
            例如：{"sub_tasks": ["定义核心概念", "搜索相关背景信息", "分析主要观点", "查找案例研究", "总结发现"]}
            确保步骤逻辑连贯且有助于深入研究。
            """,
        },
        {
            "role": "user",
            "content": f"请分解以下研究主题：'{topic}'",
        },
    ]
    try:
        response = await client.chat.completions.create(
            model=TEXT_MODEL,
            messages=prompt_messages,
            response_format={"type": "json_object"},
            max_tokens=1000,
            temperature=0.1,
        )
        if response.choices and response.choices[0].message and response.choices[0].message.content:
            content = response.choices[0].message.content.strip()
            print(f"--- AI 任务分解原始响应 ---\n{content}\n--------------------------")
            try:
                task_data = json.loads(content)
                if isinstance(task_data.get("sub_tasks"), list):
                    print("--- 分解得到的子任务 ---")
                    for i, task in enumerate(task_data["sub_tasks"]):
                        print(f"{i+1}. {task}")
                    print("------------------------")
                    return task_data["sub_tasks"]
                else:
                    print("错误: AI 响应中未包含有效的 'sub_tasks' 列表。")
                    return None
            ex.........完整代码请登录后点击上方下载按钮下载查看