最近在学习browser_use 看了很多的代码 尝试运行,均没有成功。非ui调用方式
有段代码看着比较靠谱,但是报错 "ChatOpenAI" object has no field "ainvoke"
附上代码,请会的朋友解答,感激不尽
import asyncio
import os
import logging
import warnings
from enum import Enum
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from browser_use import Agent
os.environ["ANONYMIZED_TELEMETRY"] = "false"
# 加载环境变量
load_dotenv()
# 配置日志级别和格式
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger("browser_use")
logger.setLevel(logging.INFO)
# 忽略LangChainBetaWarning警告
warnings.filterwarnings("ignore", category=UserWarning, message=".*LangChainBetaWarning.*")
class AIMode(Enum):
"""AI调用方式枚举"""
DEEPSEEK = "deepseek"
SILICONFLOW = "siliconflow"
ALIYUN = "aliyun"
OLLAMA = "ollama"
class AIClient:
"""多AI调用方式的客户端"""
def __init__(self, ai_mode: AIMode):
self.ai_mode = ai_mode
self._validate_api_keys()
def _validate_api_keys(self):
"""验证必要的API密钥是否存在"""
required_keys = {
AIMode.DEEPSEEK: 'DEEPSEEK_API_KEY',
AIMode.SILICONFLOW: 'SILICONFLOW_API_KEY'
}
if self.ai_mode in required_keys:
key = required_keys[self.ai_mode]
if not os.environ.get(key):
raise ValueError(f"缺少必要的环境变量: {key}")
def get_llm(self):
"""根据配置获取对应的LLM实例"""
if self.ai_mode == AIMode.DEEPSEEK:
return ChatOpenAI(
base_url='https://api.deepseek.com/v1',
model='deepseek-chat',
api_key=os.environ.get('DEEPSEEK_API_KEY')
)
elif self.ai_mode == AIMode.SILICONFLOW:
return ChatOpenAI(
base_url='https://api.siliconflow.cn',
model='deepseek-ai/DeepSeek-V2.5',
api_key=os.environ.get('SILICONFLOW_API_KEY')
)
else:
raise ValueError(f"不支持的AI调用方式: {self.ai_mode}")
async def main(ai_mode: AIMode = AIMode.SILICONFLOW):
"""主函数"""
try:
# 初始化AI客户端
ai_client = AIClient(ai_mode)
llm = ai_client.get_llm()
# 创建Agent并定义UI测试任务
agent = Agent(
task="1. 访问新浪官网首页(https://www.sina.com.cn/)",
llm=llm,
use_vision=False # 禁用视觉模式,依赖DOM解析
)
# 执行任务并输出分步推理和执行步骤
logger.info("🚀 开始执行任务")
result = await agent.run()
logger.info("✅ 任务执行完成")
# 输出最终结果
print("\n======== 测试结果 ========")
print(result.final_result())
except Exception as e:
logger.error(f"❌ 任务执行失败: {str(e)}")
if __name__ == "__main__":
# 通过参数选择AI调用方式
selected_mode = AIMode.DEEPSEEK # 使用硅基流动模式
asyncio.run(main(selected_mode))