# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
# 版权声明：Apache License 2.0 许可证
# 除非符合许可证要求，否则不得使用此文件
# 可以在 http://www.apache.org/licenses/LICENSE-2.0 获取许可证副本
# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
# 导入必要的Python库
import os # 用于操作系统相关的功能，如环境变量
import logging # 用于日志记录
import json # 用于JSON数据的处理
from dotenv import load_dotenv # 用于加载.env环境变量文件
from camel.models import ModelFactory # 导入模型工厂类
from camel.types import ModelPlatformType # 导入模型平台类型
# 导入工具包
from camel.toolkits import (
SearchToolkit, # 搜索工具包
BrowserToolkit, # 浏览器工具包
)
from camel.societies import RolePlaying # 导入角色扮演类
from camel.logger import set_log_level, get_logger # 导入日志相关功能
import pathlib # 用于处理文件路径
# 获取基础目录路径
base_dir = pathlib.Path(__file__).parent.parent
# 设置环境变量文件路径
env_path = base_dir / "owl" / ".env"
# 加载环境变量
load_dotenv(dotenv_path=str(env_path))
# 设置日志级别为DEBUG
set_log_level(level="DEBUG")
# 获取当前模块的日志记录器
logger = get_logger(__name__)
# 创建文件处理器，将日志写入文件
file_handler = logging.FileHandler("learning_journey.log")
# 设置文件处理器的日志级别
file_handler.setLevel(logging.DEBUG)
# 设置日志格式
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
file_handler.setFormatter(formatter)
# 将文件处理器添加到日志记录器
logger.addHandler(file_handler)
# 获取根日志记录器并添加文件处理器
root_logger = logging.getLogger()
root_logger.addHandler(file_handler)
# 设置模型类型
modelT="qwen-max-latest"
def construct_learning_society(task: str) -> RolePlaying:
"""构建学习旅程伴侣的智能体社会。
Args:
task (str): 学习任务描述，包括用户想要学习的内容和已有的知识。
Returns:
RolePlaying: 配置好的学习伴侣智能体社会。
"""
# 创建不同类型的模型实例
models = {
"user": ModelFactory.create( # 用户模型
model_platform=ModelPlatformType.OPENAI_COMPATIBLE_MODEL,
model_type=modelT,
api_key=os.getenv("OPENAI_API_KEY"),
model_config_dict={"temperature": 0.4},
),
"assistant": ModelFactory.create( # 助手模型
model_platform=ModelPlatformType.OPENAI_COMPATIBLE_MODEL,
model_type=modelT,
api_key=os.getenv("OPENAI_API_KEY"),
model_config_dict={"temperature": 0.4},
),
"content_researcher": ModelFactory.create( # 内容研究模型
model_platform=ModelPlatformType.OPENAI_COMPATIBLE_MODEL,
model_type=modelT,
api_key=os.getenv("OPENAI_API_KEY"),
model_config_dict={"temperature": 0.2},
),
"planning": ModelFactory.create( # 规划模型
model_platform=ModelPlatformType.OPENAI_COMPATIBLE_MODEL,
model_type=modelT,
api_key=os.getenv("OPENAI_API_KEY"),
model_config_dict={"temperature": 0.3},
),
}
# 创建浏览器工具包实例
browser_toolkit = BrowserToolkit(
headless=False, # 非无头模式运行
web_agent_model=models["content_researcher"], # 使用内容研究模型
planning_agent_model=models["planning"], # 使用规划模型
)
# 组合所有可用工具
tools = [
*browser_toolkit.get_tools(), # 浏览器工具
SearchToolkit().search_duckduckgo, # DuckDuckGo搜索工具
]
# 配置用户智能体参数
user_agent_kwargs = {
"model": models["user"],
}
# 配置助手智能体参数
assistant_agent_kwargs = {
"model": models["assistant"],
"tools": tools,
}
# 配置任务参数
task_kwargs = {
"task_prompt": task,
"with_task_specify": False,
}
# 创建并返回角色扮演社会
society = RolePlaying(
**task_kwargs,
user_role_name="learner", # 用户角色名
user_agent_kwargs=user_agent_kwargs,
assistant_role_name="learning_companion", # 助手角色名
assistant_agent_kwargs=assistant_agent_kwargs,
)
return society
def analyze_chat_history(chat_history):
"""分析聊天历史并提取工具调用信息。"""
print("\n============ Tool Call Analysis ============")
logger.info("========== Starting tool call analysis ==========")
# 存储工具调用信息
tool_calls = []
# 遍历聊天历史
for i, message in enumerate(chat_history):
# 检查助手消息中的工具调用
if message.get("role") == "assistant" and "tool_calls" in message:
for tool_call in message.get("tool_calls", []):
if tool_call.get("type") == "function":
function = tool_call.get("function", {})
# 记录工具调用信息
tool_info = {
"call_id": tool_call.get("id"),
"name": function.get("name"),
"arguments": function.get("arguments"),
"message_index": i,
}
tool_calls.append(tool_info)
# 打印工具调用信息
print(
f"Tool Call: {function.get('name')} Args: {function.get('arguments')}"
)
logger.info(
f"Tool Call: {function.get('name')} Args: {function.get('arguments')}"
)
# 检查工具返回结果
elif message.get("role") == "tool" and "tool_call_id" in message:
for tool_call in tool_calls:
if tool_call.get("call_id") == message.get("tool_call_id"):
result = message.get("content", "")
# 截取结果摘要
result_summary = (
result[:100] + "..." if len(result) > 100 else result
)
print(
f"Tool Result: {tool_call.get('name')} Return: {result_summary}"
)
logger.info(
f"Tool Result: {tool_call.get('name')} Return: {result_summary}"
)
# 打印工具调用总数
print(f"Total tool calls found: {len(tool_calls)}")
logger.info(f"Total tool calls found: {len(tool_calls)}")
logger.info("========== Finished tool call analysis ==========")
# 将聊天历史保存到JSON文件
with open("learning_journey_history.json", "w", encoding="utf-8") as f:
json.dump(chat_history, f, ensure_ascii=False, indent=2)
print("Records saved to learning_journey_history.json")
print("============ Analysis Complete ============\n")
def run_learning_companion(task: str = None):
"""运行学习伴侣。
Args:
task (str, optional): 学习任务描述。默认为示例任务。
"""
# 设置默认学习任务
task = """
I want to learn about the transformers architecture in an llm.
I've also taken a basic statistics course.
I have about 10 hours per week to dedicate to learning. Devise a roadmap for me .
"""
# 构建学习社会
society = construct_learning_society(task)
from owl.utils import run_society
# 运行社会并获取结果
answer, chat_history, token_count = run_society(society, round_limit=5)
# 记录工具使用历史
analyze_chat_history(chat_history)
# 打印答案
print(f"\033[94mAnswer: {answer}\033[0m")
# 主程序入口
if __name__ == "__main__":
run_learning_companion()
这个文件主要实现了一个学习助手系统，主要功能：
环境设置和初始化：
主要功能模块：
construct_learning_society: 构建学习助手系统，包括：
- 创建不同类型的AI模型（用户、助手、内容研究、规划）
- 配置浏览器和搜索工具
- 设置角色扮演参数
工具分析功能：
analyze_chat_history: 分析聊天历史，记录和展示：
- 工具调用情况
- 工具返回结果
- 将历史记录保存到JSON文件
主运行函数：
run_learning_companion: 运行学习助手系统，包括：
- 设置默认学习任务
- 构建学习社会
- 运行对话
- 分析工具使用情况
- 输出结果
这个系统的主要目的是帮助用户学习特定主题（如示例中的transformer架构），通过多个AI模型的协作来提供个性化的学习体验。系统使用了多种工具（如浏览器和搜索引擎）来获取和提供学习资源。
