Hệ thống trợ lý thông minh sử dụng LlamaIndex và mô hình GLM

Với sự phát triển nhanh chóng của trí tuệ nhân tạo, việc quản lý và khai thác hiệu quả tài sản kiến thức nội bộ của doanh nghiệp trở thành thách thức lớn trong quá trình chuyển đổi số. Bài viết này giới thiệu một hệ thống trợ lý kiến thức doanh nghiệp dựa trên khung LlamaIndex và mô hình AI lớn GLM, tích hợp cơ sở dữ liệu vectơ cục bộ và nhiều công cụ tiện ích.

1. Tải mô hình

Mô hình được chọn là glm-4-flash, một phiên bản nhẹ của dòng mô hình GLM. Mặc dù khả năng sinh không bằng các phiên bản cao cấp hơn như glm-pro, nhưng nó đủ đáp ứng nhu cầu cho các dự án demo với ưu điểm miễn phí.

# Thiết lập khóa API
ZHIPU_API_KEY = "Your_Key"
os.environ["ZHIPU_API_KEY"] = ZHIPU_API_KEY

# Khởi tạo mô hình GLM
llm = ZhipuAI(model="glm-4-flash", temperature=0.5, api_key=ZHIPU_API_KEY)
Settings.llm = llm

2. Tải cơ sở dữ liệu vectơ

Hệ thống sử dụng FAISS để lưu trữ thư viện kiến thức tài liệu của doanh nghiệp:

# Hàm tải cơ sở dữ liệu vectơ
def load_vector_db(db_path, model_path):
    embedding_model = SentenceTransformerEmbeddings(
        model_name=model_path,
        model_kwargs={'device': 'cpu'},
        encode_kwargs={'normalize_embeddings': False}
    )
    
    vector_db = FAISS.load_local(
        folder_path=db_path,
        embeddings=embedding_model,
        allow_dangerous_deserialization=True
    )
    return vector_db

# Tải cơ sở dữ liệu FAISS
faiss_vector_db = load_vector_db("./faiss_index", "./local_models/all-mpnet-base-v2")

3. Xây dựng các phương pháp

3.1. Bộ điều chỉnh truy vấn FAISS (Kiểm tra kho kiến thức)

# Lớp truy vấn FAISS
class FAISSQueryEngine:
    def __init__(self, vector_db, top_k=3):
        self.vector_db = vector_db
        self.top_k = top_k
    
    def query(self, query_str: str) -> str:
        try:
            results = self.vector_db.similarity_search_with_score(query_str, k=self.top_k)
            
            response = []
            for i, (doc, score) in enumerate(results):
                response.append(
                    f"Tài liệu {i+1} -  (Độ tương đồng: {1-score:.2f})\n"
                    f"Nội dung: {doc.page_content[:200]}..."
                )
            
            return "\n\n".join(response)
        except Exception as e:
            return f"Lỗi truy vấn: {str(e)}"

3.2. Công cụ tính toán toán học

# Hàm thực hiện phép tính toán học
def calculator_tool(expression: str) -> str:
    sanitized_expr = re.sub(r'[()$$$${};"\'\`<>]', '', expression)
    try:
        sanitized_expr = sanitized_expr.replace('×', '*').replace('÷', '/')
        result = eval(sanitized_expr)
        return f"Kết quả: {expression} = {result}"
    except Exception as e:
        return f"Lỗi tính toán: {str(e)}"

3.3. Công cụ tìm kiếm web

# Hàm tìm kiếm web sử dụng DuckDuckGo
def web_search(query: str) -> str:
    try:
        response = requests.get(
            "https://api.duckduckgo.com/",
            params={
                "q": query,
                "format": "json",
                "no_html": "1",
                "no_redirect": "1",
                "t": "MyApp"
            },
            timeout=10
        )
        data = response.json()
        
        results = []
        if data.get("AbstractText"):
            results.append(f"Tóm tắt: {data['AbstractText']}")
        
        for topic in data.get("RelatedTopics", []):
            if "FirstURL" in topic and "Text" in topic:
                results.append(f"{topic['Text']} ({topic['FirstURL']})")
        
        return "\n".join(results[:3]) if results else "Không tìm thấy thông tin liên quan"
    except Exception as e:
        return f"Lỗi tìm kiếm: {str(e)}"

3.4. Thông tin địa chỉ số điện thoại

# Hàm lấy thông tin địa chỉ số điện thoại
def get_phone_location(phone_number: str) -> str:
    api_url = "http://apis.juhe.cn/mobile/get"
    api_key = ""  # Thay thế bằng khóa API thực tế
    
    params = {
        "phone": phone_number,
        "key": api_key
    }
    
    try:
        response = requests.get(api_url, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        
        if data.get("error_code") != 0:
            error_msg = data.get("reason", "Lỗi không xác định")
            return f"Lỗi API: {error_msg} (Mã lỗi: {data.get('error_code')})"
        
        result = data.get("result", {})
        province = result.get("province", "Tỉnh không xác định")
        city = result.get("city", "Thành phố không xác định")
        operator = result.get("company", "Nhà mạng không xác định")
        
        return f"Số điện thoại: {phone_number}\nĐịa chỉ: {province}{city}\nNhà mạng: {operator}"
    except Exception as e:
        return f"Lỗi xử lý: {str(e)}"

4. Quản lý công cụ FunctionTool

FunctionTool đóng vai trò quan trọng trong việc bao gói các hàm tùy chỉnh để mô hình ngôn ngữ lớn có thể gọi đến.

# Tạo tập hợp công cụ
document_tool = FunctionTool.from_defaults(
    fn=lambda query: document_query_engine.query(query),
    name="search_document",
    description="Truy vấn tài liệu nội bộ và thư viện kiến thức"
)

calculator_tool = FunctionTool.from_defaults(
    fn=calculator_tool,
    name="math_calculator",
    description="Thực hiện phép tính toán học"
)

web_search_tool = FunctionTool.from_defaults(
    fn=web_search,
    name="search_web",
    description="Sử dụng DuckDuckGo để tìm kiếm thông tin mới nhất trên web"
)

get_phone_location_tool = FunctionTool.from_defaults(
    fn=get_phone_location,
    name="get_phone_info",
    description="Lấy thông tin địa chỉ số điện thoại"
)

tools = [
    document_tool,
    web_search_tool,
    get_phone_location_tool,
    calculator_tool
]

5. Xây dựng trợ lý thông minh

Dưới đây là cách tạo ra một trợ lý đơn giản với khả năng mở rộng mạnh mẽ nhờ vào FunctionTool:

# Tạo trợ lý
def create_agent():
    return ReActAgent.from_tools(
        tools=tools,
        llm=llm,
        verbose=True,
        system_prompt=f"""
        Ngày hiện tại: {datetime.now().strftime('%Y-%m-%d')}
        Bạn là trợ lý chuyên nghiệp có thể truy cập vào thư viện tài liệu, tìm kiếm web và công cụ tính toán.
        Tuân theo quy tắc:
        1. Sử dụng công cụ search_document cho câu hỏi tài liệu
        2. Sử dụng công cụ search_web cho câu hỏi thời gian thực
        3. Sử dụng math_calculator cho phép tính toán
        4. Sử dụng get_phone_info để kiểm tra số điện thoại
        """
    )

class ChatAgent:
    def __init__(self):
        self.session_history = {}
    
    def get_agent(self, session_id):
        if session_id not in self.session_history:
            self.session_history[session_id] = create_agent()
        return self.session_history[session_id]
    
    def chat(self, message, session_id="default"):
        agent = self.get_agent(session_id)
        return agent.chat(message)

chat_agent = ChatAgent()

6. Dịch vụ FastAPI

# Tạo dịch vụ API
app = FastAPI(title="Hỗ trợ kiến thức doanh nghiệp")

class ChatRequest(BaseModel):
    message: str
    session_id: str = "default"

def sanitize_input(text: str) -> str:
    return re.sub(r'[<>\"\';%&\\]', '', text)[:500]

@app.post("/chat")
async def chat_endpoint(request: ChatRequest):
    try:
        clean_message = sanitize_input(request.message)
        session_id = sanitize_input(request.session_id)[:64]
        agent = chat_agent.get_agent(session_id)
        response = agent.chat(clean_message)
        return {"response": str(response)}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    print("=" * 60)
    print("Hệ thống trợ lý kiến thức - LlamaIndex + ZhipuAI")
    print(f"Mô hình: {llm.model} | Khóa API: {ZHIPU_API_KEY[:12]}...")
    print(f"Số lượng tài liệu trong FAISS: {faiss_vector_db.index.ntotal}")
    print("=" * 60)
    
    print("\nKhởi động dịch vụ API: http://localhost:8000")
    print("Sử dụng CTRL+C để dừng dịch vụ")
    uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")