python+openai兼容api+html实现类似Clawdbot(MoltBot)的ai自主分析任务编写代码运行完成任务代码-python代码-BFW代码库

代码语言：python

所属分类：其他

代码描述：python+openai兼容api+html实现类似Clawdbot(MoltBot)的ai自主分析任务编写代码运行完成任务代码，ai的编程能力大幅提升后，任何的任务都可以通过编程来完成，那么ai可以操作本地来完成用户的任务，包括电脑设置，文档处理，视频剪辑等等。可以接入ollama本地模型，也可以接入第三方兼容openai的其他大模型api。

代码标签： python openai 兼容 html api 类似 Clawdbot MoltBot 对话

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

#!/usr/local/python3/bin/python3
# -*- coding: utf-8 -*
import os
import sys
import json
import asyncio
import platform
import re
import datetime
import shutil
import time
import uuid
from typing import List, Dict, Any, Optional

# Web 框架依赖
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, UploadFile, File, Request, Depends, HTTPException
from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse
from fastapi.staticfiles import StaticFiles

# AI 依赖
from openai import OpenAI
# Playwright 依赖
from playwright.async_api import async_playwright
# ================= 配置区域 =================
PLANNER_CONFIG = {
    "api_key": "sk-",  # 请替换你的 Key
    "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
    "model": "qwen-max"
}

CODER_CONFIG = {
    "api_key": "sk-",  # 请替换你的 Key
    "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
    "model": "qwen3-coder-flash" # 建议用 coder 模型
}
# 登录认证配置
AUTH_CONFIG = {
    "username": "admin",
    "password": "bfwbot2025", 
    "cookie_name": "bfwbot_session",
    "cookie_value": "authorized_token_xyz"
}

# 目录配置
WORKSPACE_DIR = os.path.join(os.getcwd(), "workspace")
UPLOAD_DIR = os.path.join(os.getcwd(), "uploads")

os.makedirs(WORKSPACE_DIR, exist_ok=True)
os.makedirs(UPLOAD_DIR, exist_ok=True)

# ================= 数据存储 (内存数据库) =================
CHAT_SESSIONS: Dict[str, Dict[str, Any]] = {}

def save_log_to_session(session_id: str, log_entry: dict):
    if session_id not in CHAT_SESSIONS:
        CHAT_SESSIONS[session_id] = {
            "title": "新会话", 
            "timestamp": time.time(), 
            "logs": []
        }
    
    # 如果是消息类型，且当前标题是默认值，尝试更新标题
    if log_entry.get("type") == "message" and log_entry.get("content"):
        current_title = CHAT_SESSIONS[session_id]["title"]
        if current_title == "新会话" or current_title == "无标题会话":
            # 截取前20个字符作为标题
            title = log_entry.get("content", "")[:20]
            if not title and log_entry.get("attachments"):
                title = "文件上传会话"
            elif not title:
                title = "无标题会话"
            CHAT_SESSIONS[session_id]["title"] = title

    CHAT_SESSIONS[session_id]["logs"].append(log_entry)
    CHAT_SESSIONS[session_id]["timestamp"] = time.time()

# ================= 1. 环境感知模块 =================
class SystemEnvironment:
    @staticmethod
    def get_info() -> str:
        base_url = "http://localhost:8000"
        return (
            f"操作系统: {platform.system()} {platform.release()} ({platform.machine()})\n"
            f"Python版本: {sys.version.split()[0]}\n"
            f"文件系统与URL映射规则 (非常重要):\n"
            f"1. 本地目录 '{WORKSPACE_DIR}' -> Web路径 '{base_url}/download/'\n"
            f"2. 本地目录 '{UPLOAD_DIR}'    -> Web路径 '{base_url}/uploads/'\n"
        )

# ================= 2. 搜索智能体 (截图版) =================
class SearchAgent:
    def __init__(self, send_log_func):
        self.send_log = send_log_func

    async def run(self, query: str) -> str:
        await self.send_log("browser", f"正在启动 Playwright 浏览器以搜索: {query}...")
        results = []
        try:
            async with async_playwright() as p:
                browser = await p.chromium.launch(headless=True)
                context = await browser.new_context(
                    viewport={'width': 1280, 'height': 800},
                    user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
                )
                page = await context.new_page()
                
                search_url = f"https://www.bing.com/search?q={query}"
                await page.goto(search_url, timeout=20000)
                await page.wait_for_load_state("networkidle")
                
                timestamp = int(time.time())
                screenshot_filename = f"screenshot_{timestamp}.png"
                screenshot_path = os.path.join(UPLOAD_DIR, screenshot_filename)
                await page.screenshot(path=screenshot_path)
                
                screenshot_url = f"/uploads/{screenshot_filename}"
                await self.send_log("browser_snapshot", screenshot_url)

                elements = await page.locator("li.b_algo").all()
                await self.send_log("browser", f"页面加载完成，提取到 {len(elements)} 条结果，正在解析...")

                for i, el in enumerate(elements[:4]):
                    try:
                        title = await el.locator("h2").inner_text()
                        link_el = el.locator("a").first
                        url = await link_el.get_attribute("href")
                        snippet_el = el.locator("p")
                        if await snippet_el.count() == 0:
                            snippet_el = el.locator(".b_caption")
                        snippet = await snippet_el.inner_text() if await snippet_el.count() > 0 else "无摘要"
                        results.append({"title": title, "url": url, "snippet": snippet})
                    except: continue

                await browser.close()
        except Exception as e:
            await self.send_log("error", f"Playwright 浏览器出错: {e}")
            return f"浏览器搜索失败: {e}"

        if not results: return "未找到搜索结果。"
        formatted_res = f"【搜索结果 - {query}】\n"
        for i, item in enumerate(results):
 .........完整代码请登录后点击上方下载按钮下载查看