autogen_code_excutor
main.py
telegram_bot: rag_exo_bot (rag_bot)
Mở Đầu:
Trí tuệ nhân tạo (AI) đang len lỏi vào mọi ngóc ngách của cuộc sống, và chatbot AI là một minh chứng rõ ràng cho sự phát triển vượt bậc này. Bài viết này sẽ giới thiệu một chương trình Python đột phá, sử dụng sức mạnh của Ollama và mô hình ngôn ngữ lớn Qwen để tạo ra một bot Telegram thông minh, mang đến trải nghiệm trò chuyện tự nhiên và hiệu quả chưa từng có.
Giới Thiệu Chương Trình Dựa Trên Hình Ảnh Thực Tế:
Hình ảnh chụp màn hình cho thấy một dự án được phát triển bài bản trong môi trường Visual Studio Code (VS Code), với cấu trúc thư mục rõ ràng và sử dụng các công nghệ tiên tiến. Chương trình sử dụng các thư viện Python như asyncio, logging, os, dotenv và các module tự phát triển trong các thư mục core, agents, và utils. Mục tiêu chính là triển khai một bot Telegram có khả năng tương tác với người dùng thông qua mô hình ngôn ngữ lớn Ollama, cụ thể là qwen2.5:14b – một model mạnh mẽ từ Qwen, hứa hẹn khả năng xử lý ngôn ngữ vượt trội.
Các Thành Phần Chính Qua Phân Tích Cấu Trúc Dự Án:
asyncio: Lập trình bất đồng bộ, cho phép bot xử lý nhiều yêu cầu cùng lúc, đảm bảo trải nghiệm mượt mà.logging: Ghi nhật ký hoạt động, hỗ trợ theo dõi lỗi và tối ưu hóa hiệu suất.os: Tương tác với hệ điều hành, truy cập biến môi trường.dotenv: Quản lý biến môi trường từ tệp.env, bảo mật token bot Telegram.core.telegram_bot(Quan Sát Từ Cấu Trúc Thư Mục): Module trung tâm, chứa lớpTelegramBot, xử lý các tương tác với Telegram API.config.py: Chứa các thiết lập mặc định, bao gồm model Ollama (DEFAULT_OLLAMA_MODEL) và tên tệp cấu hình (DEFAULT_CONFIG_FILE).utils.config_loader: Quản lý việc tạo và tải cấu hình.agents(Thư Mục): Cho thấy tiềm năng mở rộng với nhiều “agent” AI chuyên biệt.
Quy Trình Hoạt Động Chi Tiết:
- Khởi Động và Xác Thực:
- Tải các biến môi trường từ tệp
.env, đặc biệt làTELEGRAM_BOT_TOKEN. - Kiểm tra sự hiện diện của token; nếu thiếu, chương trình sẽ báo lỗi và dừng.
- Khởi tạo cấu hình mặc định với model
DEFAULT_OLLAMA_MODEL. - Tự động tạo tệp cấu hình (
DEFAULT_CONFIG_FILE) nếu cần và lưu cấu hình mặc định. - Đọc và nạp cấu hình từ tệp.
- Tải các biến môi trường từ tệp
- Khởi Tạo Bot Mạnh Mẽ:
- Tạo đối tượng
TelegramBotvới token, danh sách cấu hình và model Ollamaqwen2.5:14b.
- Tạo đối tượng
- Vận Hành và Tương Tác:
- Kích hoạt phương thức
start()củaTelegramBotđể bắt đầu lắng nghe và xử lý tin nhắn Telegram.
- Kích hoạt phương thức
Điểm Nhấn Công Nghệ Dựa Trên Hình Ảnh và Code:
- Sức Mạnh Ollama: Chạy LLM cục bộ, tăng cường bảo mật và giảm thiểu độ trễ.
- Model Qwen Tiên Tiến: Sử dụng
qwen2.5:14bđảm bảo chất lượng phản hồi và khả năng hiểu ngôn ngữ tự nhiên. - Tùy Chỉnh Linh Hoạt: Dễ dàng thay đổi cấu hình qua tệp, cho phép người dùng điều chỉnh bot theo nhu cầu.
- Hiệu Suất Cao Nhờ Asyncio: Xử lý bất đồng bộ, cho phép bot phản hồi nhanh chóng, ngay cả với nhiều người dùng đồng thời.
- Mã Nguồn Mở và Minh Bạch: Khuyến khích cộng đồng tham gia phát triển và hoàn thiện.
- Phát Triển Chuyên Nghiệp: Sử dụng VS Code và cấu trúc dự án khoa học, thể hiện sự chuyên nghiệp trong phát triển.
- Giao Diện Debug Trực Quan: Cung cấp thông tin chi tiết về hoạt động của bot, hỗ trợ gỡ lỗi và tối ưu hóa.
Ứng Dụng Thực Tế Đa Dạng:
- Hỗ Trợ Khách Hàng 24/7: Trả lời câu hỏi thường gặp, cung cấp thông tin sản phẩm và dịch vụ.
- Trợ Lý Học Tập: Giải đáp thắc mắc, cung cấp tài liệu và hỗ trợ ôn tập hiệu quả.
- Giải Trí Thông Minh: Trò chuyện, kể chuyện, chơi game, mang lại trải nghiệm giải trí tương tác.
- Tự Động Hóa Nhiệm Vụ: Tích hợp vào các hệ thống khác để tự động hóa quy trình như đặt lịch, gửi email và quản lý công việc.
Kết Luận:
Dự án Python này, với việc sử dụng Ollama và Qwen, đã tạo ra một bot Telegram thông minh, mở ra tiềm năng to lớn trong giao tiếp tự động. Sự kết hợp giữa công nghệ tiên tiến, khả năng tùy chỉnh linh hoạt, hiệu suất vượt trội và môi trường phát triển chuyên nghiệp (thể hiện qua hình ảnh) hứa hẹn một tương lai nơi bot AI trở thành trợ thủ đắc lực trong cuộc sống hàng ngày. Chương trình này không chỉ là một bước tiến trong công nghệ AI mà còn là minh chứng cho sức mạnh của cộng đồng mã nguồn mở trong việc tạo ra những giải pháp đột phá.
#main.py
import asyncio
import logging
import os
import dotenv
from core.telegram_bot import TelegramBot
from config import create_default_config, DEFAULT_OLLAMA_MODEL, DEFAULT_CONFIG_FILE
from utils.config_loader import create_config_file, load_config
dotenv.load_dotenv()
# Cấu hình logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
async def main():
bot_token = os.getenv('TELEGRAM_BOT_TOKEN')
if not bot_token:
logging.error("Lỗi: Không tìm thấy TELEGRAM_BOT_TOKEN. Vui lòng đặt token.")
logging.error("Sử dụng: export TELEGRAM_BOT_TOKEN='your_telegram_bot_token'")
return
default_config = create_default_config(DEFAULT_OLLAMA_MODEL)
config_path = create_config_file(DEFAULT_CONFIG_FILE, default_config)
config_list = load_config(config_path)
ai_bot = TelegramBot(
bot_token=bot_token,
default_config = config_list,
ollama_model='qwen2.5:14b'
)
await ai_bot.start()
if __name__ == "__main__":
asyncio.run(main())
# utils/config_loader.py
import os
import json
import logging
from typing import List, Dict, Any
from config import get_config_path, create_default_config
def load_config(config_file: str) -> List[Dict[str, Any]]:
"""Tải cấu hình từ file JSON"""
config_path = get_config_path(config_file)
try:
with open(config_path, 'r') as f:
config_data = json.load(f)
return config_data.get('config_list', [])
except FileNotFoundError:
logging.error(f"Không tìm thấy file cấu hình: {config_file}. Sử dụng cấu hình mặc định.")
return []
except json.JSONDecodeError:
logging.error(f"Lỗi giải mã JSON trong file cấu hình: {config_file}. Sử dụng cấu hình mặc định.")
return []
def create_config_file(filename: str, default_config: dict) -> str:
"""Tạo file cấu hình Autogen nếu chưa tồn tại"""
config_path = get_config_path(filename)
if not os.path.exists(config_path):
with open(config_path, 'w') as f:
json.dump(default_config, f, indent=4)
logging.info(f"Đã tạo file cấu hình tại {config_path}")
return config_path
def load_agent_config(agent_name: str, default_config: dict) -> List[Dict[str, Any]]:
config_file = f"{agent_name.lower()}_agent_config.json"
if os.path.exists(config_file):
return load_config(config_file)
return default_config# agents/base_agent.py
from abc import ABC, abstractmethod
from typing import Dict, List
from autogen import AssistantAgent
class BaseAgent(ABC):
def __init__(self, name: str, system_message: str, llm_config: Dict, default_config: List[Dict]):
self.name = name
self.system_message = system_message
self.llm_config = llm_config
self.default_config = default_config
self.agent = self._create_agent() # Create agent upon instantiation
@abstractmethod
def _create_agent(self) -> AssistantAgent:
"""Abstract method to create and return an AssistantAgent"""
pass
def get_agent(self) -> AssistantAgent:
"""Return the created agent"""
return self.agent# agents/math_agent.py
from typing import Dict, List
from autogen import AssistantAgent
from core.agents.base_agent import BaseAgent
from utils.config_loader import load_agent_config
class MathAgent(BaseAgent):
def __init__(self, default_config: List[Dict]):
llm_config = {
"config_list": load_agent_config("math", default_config),
"temperature": 0.7,
"functions": [
{
"name": "solve_equation",
"description": "Giải một phương trình toán học",
"parameters": {
"type": "object",
"properties": {
"equation": {
"type": "string",
"description": "Phương trình cần giải"
}
},
"required": ["equation"]
}
}
]
}
super().__init__(
name="MathAgent",
system_message="Bạn là một chuyên gia toán học. Bạn có thể giải các bài toán, thực hiện các phép tính.",
llm_config = llm_config,
default_config = default_config
)
def _create_agent(self) -> AssistantAgent:
return AssistantAgent(
name=self.name,
llm_config=self.llm_config,
system_message=self.system_message
)# agents/researcher_agent.py
from typing import Dict, List
from autogen import AssistantAgent
from core.agents.base_agent import BaseAgent
from utils.config_loader import load_agent_config
class ResearcherAgent(BaseAgent):
def __init__(self, default_config: List[Dict]):
llm_config = {
"config_list": load_agent_config("researcher", default_config),
"temperature": 0.7,
"functions": [
{
"name": "search_internet",
"description": "Tìm kiếm thông tin trên internet.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Từ khóa tìm kiếm"
}
},
"required": ["query"]
}
},
{
"name": "search_webpage",
"description": "Tìm kiếm thông tin trên một trang web.",
"parameters": {
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "URL của trang web"
}
},
"required": ["url"]
}
}
]
}
super().__init__(
name="ResearcherAgent",
system_message="Bạn là một chuyên gia nghiên cứu. Bạn có thể tìm kiếm thông tin, tổng hợp dữ liệu và cung cấp các phân tích.",
llm_config= llm_config,
default_config = default_config
)
def _create_agent(self) -> AssistantAgent:
return AssistantAgent(
name=self.name,
llm_config=self.llm_config,
system_message=self.system_message
)# agents/stock_agent.py
from io import BytesIO
from typing import Dict, List
from autogen import AssistantAgent
from core.agents.base_agent import BaseAgent
from utils.config_loader import load_agent_config
from core.stock_tools import StockTools
from core.knowledge_retriever import KnowledgeRetriever
class StockAgent(BaseAgent):
def __init__(self, default_config: List[Dict], knowledge_retriever: KnowledgeRetriever):
llm_config = {
"config_list": load_agent_config("stock", default_config),
"temperature": 0.7,
"functions": [
{
"name": "get_stock_info",
"description": "Lấy thông tin cổ phiếu từ cơ sở dữ liệu local.",
"parameters": {
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Mã cổ phiếu."
}
},
"required": ["symbol"]
}
},
{
"name": "get_stock_price",
"description": "Lấy giá cổ phiếu mới nhất từ cơ sở dữ liệu local.",
"parameters": {
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Mã cổ phiếu."
}
},
"required": ["symbol"]
}
},
{
"name": "get_stock_chart",
"description": "Lấy biểu đồ cổ phiếu từ cơ sở dữ liệu local.",
"parameters": {
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Mã cổ phiếu."
}
},
"required": ["symbol"]
}
}
]
}
super().__init__(
name="StockAgent",
system_message="Bạn là một chuyên gia về cổ phiếu. Bạn có thể cung cấp thông tin, giá và biểu đồ về cổ phiếu.",
llm_config=llm_config,
default_config=default_config
)
self.stock_tools = StockTools(knowledge_retriever) # Initialize tools
def _create_agent(self) -> AssistantAgent:
return AssistantAgent(
name=self.name,
llm_config=self.llm_config,
system_message=self.system_message
)
async def get_stock_info(self, symbol: str) -> str:
"""Call the tool function for getting stock info"""
return await self.stock_tools.get_stock_info(symbol)
async def get_stock_price(self, symbol: str) -> str:
"""Call the tool function for getting stock price"""
return await self.stock_tools.get_stock_price(symbol)
async def get_stock_chart(self, symbol: str) -> BytesIO:
"""Call the tool function for getting stock chart"""
return await self.stock_tools.get_stock_chart(symbol)
# # agents/stock_agent.py
# from io import BytesIO
# from typing import Dict, List
# from autogen import AssistantAgent
# from core.agents.base_agent import BaseAgent
# from utils.config_loader import load_agent_config
# from core.stock_tools import StockTools
# from core.knowledge_retriever import KnowledgeRetriever
# class StockAgent(BaseAgent):
# def __init__(self, default_config: List[Dict], knowledge_retriever: KnowledgeRetriever):
# llm_config = {
# "config_list": load_agent_config("stock", default_config),
# "temperature": 0.7,
# "functions":[
# {
# "name": "get_stock_info",
# "description": "Lấy thông tin cổ phiếu từ cơ sở dữ liệu local.",
# "parameters": {
# "type": "object",
# "properties": {
# "symbol": {
# "type": "string",
# "description": "Mã cổ phiếu."
# }
# },
# "required": ["symbol"]
# }
# },
# {
# "name": "get_stock_chart",
# "description": "Lấy biểu đồ cổ phiếu từ cơ sở dữ liệu local.",
# "parameters": {
# "type": "object",
# "properties": {
# "symbol": {
# "type": "string",
# "description": "Mã cổ phiếu."
# }
# },
# "required": ["symbol"]
# }
# }
# ]
# }
# super().__init__(
# name="StockAgent",
# system_message="Bạn là một chuyên gia về cổ phiếu. Bạn có thể cung cấp thông tin và biểu đồ về cổ phiếu.",
# llm_config=llm_config,
# default_config=default_config
# )
# self.stock_tools = StockTools(knowledge_retriever) # Initialize tools
# def _create_agent(self) -> AssistantAgent:
# return AssistantAgent(
# name=self.name,
# llm_config=self.llm_config,
# system_message=self.system_message
# )
# async def get_stock_info(self, symbol: str) -> str:
# """Call the tool function for getting stock info"""
# return await self.stock_tools.get_stock_info(symbol)
# async def get_stock_chart(self, symbol: str) -> BytesIO:
# """Call the tool function for getting stock chart"""
# return await self.stock_tools.get_stock_chart(symbol)
# # # agents/stock_agent.py
# # from typing import Dict, List
# # from autogen import AssistantAgent
# # from core.agents.base_agent import BaseAgent
# # from utils.config_loader import load_agent_config
# # class StockAgent(BaseAgent):
# # def __init__(self, default_config: List[Dict]):
# # llm_config = {
# # "config_list": load_agent_config("stock", default_config),
# # "temperature": 0.7,
# # "functions":[
# # {
# # "name": "get_stock_info",
# # "description": "Lấy thông tin cổ phiếu từ cơ sở dữ liệu local.",
# # "parameters": {
# # "type": "object",
# # "properties": {
# # "symbol": {
# # "type": "string",
# # "description": "Mã cổ phiếu."
# # }
# # },
# # "required": ["symbol"]
# # }
# # },
# # {
# # "name": "get_stock_chart",
# # "description": "Lấy biểu đồ cổ phiếu từ cơ sở dữ liệu local.",
# # "parameters": {
# # "type": "object",
# # "properties": {
# # "symbol": {
# # "type": "string",
# # "description": "Mã cổ phiếu."
# # }
# # },
# # "required": ["symbol"]
# # }
# # }
# # ]
# # }
# # super().__init__(
# # name="StockAgent",
# # system_message="Bạn là một chuyên gia về cổ phiếu. Bạn có thể cung cấp thông tin và biểu đồ về cổ phiếu.",
# # llm_config=llm_config,
# # default_config=default_config
# # )
# # def _create_agent(self) -> AssistantAgent:
# # return AssistantAgent(
# # name=self.name,
# # llm_config=self.llm_config,
# # system_message=self.system_message
# # )