Tài liệu này hướng dẫn quy trình xây dựng ứng dụng trí tuệ nhân tạo hoàn chỉnh sử dụng ngôn ngữ Python cùng thư viện LangChain kết hợp với mô hình lớn Tongyi Qianwen của Alibaba. Chúng ta sẽ lần lượt đi qua các bước từ giao diện hội thoại đơn giản đến việc tích hợp cơ sở dữ liệu doanh nghiệp (RAG), đảm bảo tính thực tiễn và khả năng mở rộng.

Chuẩn Bị Môi Trường

Trước khi bắt đầu mã hóa, cần thiết lập môi trường làm việc ổn định bằng cách cài đặt các gói phụ thuộc yêu cầu và cấu hình quyền truy cập vào dịch vụ đám mây.

1. Cài đặt thư viện:

pip install langchain langchain-community langchain-core python-dotenv faiss-cpu pypdf

2. Lấy khóa API: Truy cập nền tảng阿里云百炼 (Alibaba Cloud Bailian), tạo khóa mới cho API và lưu trữ an toàn.
3. Bộ nhớ môi trường: Tạo tập tin `.env` ở thư mục gốc dự án để chứa thông tin xác thực.

DASHSCOPE_API_KEY=khóa_của_bạn_ở_đây

Giai Đoạn 1: Khởi Tạo Hội Thoại Đơn

Đây là mức độ cơ bản nhất, nơi mô hình nhận một đầu vào duy nhất và trả về kết quả tương ứng mà không lưu giữ trạng thái trước đó.

Mã Nguồn Thực Hiện

import os
from langchain_community.chat_models import ChatTongyi
from dotenv import load_dotenv
from langchain_core.messages import HumanMessage

# Tải biến môi trường
load_dotenv()

# Khởi tạo khách hàng AI
ai_client = ChatTongyi()

# Gửi lệnh yêu cầu
prompt_msg = [HumanMessage(content="Bạn là ai?")]
result_output = ai_client.invoke(prompt_msg)

print(result_output.content)

Ghi Chú Kỹ Thuật

• ChatTongyi: Lớp Wrapper cho phép kết nối trực tiếp tới Tongyi Qianwen qua LangChain.
• invoke(): Phương thức đồng bộ để gửi yêu cầu và chờ phản hồi.
• Hạn chế: Phương pháp này chỉ phù hợp cho các tác vụ xử lý văn bản độc lập.

Giai Đoạn 2: Tương Tác Vòng Lặp Tại Terminal

Mở rộng chức năng trên để người dùng có thể trao đổi nhiều vòng liên tục tại dòng lệnh mà không cần khởi động lại chương trình.

Mã Nguồn

import sys
from langchain_community.chat_models import ChatTongyi
from dotenv import load_dotenv
from langchain_core.messages import HumanMessage

load_dotenv()
engine = ChatTongyi()

print("Chế độ chat console: Gõ 'quit' để thoát")

while True:
    try:
        user_text = input("Người dùng > ")
        if not user_text.strip():
            continue
        
        if user_text.lower() == 'quit':
            break
            
        msg_list = [HumanMessage(content=user_text)]
        reply = engine.invoke(msg_list)
        print(f"Bot > {reply.content}\n")
        
    except KeyboardInterrupt:
        break
except Exception as e:
    print(f"Lỗi hệ thống: {e}")

Ở giai đoạn này, thuật toán chưa có khả năng ghi nhớ các thông tin đã nói trước đó. Mỗi câu hỏi đều được xem xét như một vấn đề mới hoàn toàn.

Giai Đoạn 3: Duy Trì Bộ Nhớ Hội Thoại

Vấn đề cốt lõi của chatbot thông minh là khả năng liên kết ngữ cảnh. Giải pháp là lưu trữ danh sách các tin nhắn cũ và chuyển toàn bộ lịch sử vào tham số đầu tiên mỗi lần gọi API.

Cấu Trúc Lưu Trữ

from langchain_core.messages import AIMessage, HumanMessage

load_dotenv()
model = ChatTongyi()

history_log = []

print("Nhập 'exit' để kết thúc phiên chat")

while True:
    inp = input("Bạn: ")
    if inp == 'exit': break
    
    # Thêm câu hỏi của người dùng vào lịch sử
    history_log.append(HumanMessage(content=inp))
    
    # Gửi cả lịch sử lên mô hình
    response_obj = model.invoke(history_log)
    ai_answer = response_obj.content
    
    # Lưu câu trả lời của AI vào lịch sử
    history_log.append(AIMessage(content=ai_answer))
    
    print(f"AI: {ai_answer}\n")

Kiến trúc này đảm bảo mô hình luôn có đầy đủ ngữ cảnh để đưa ra phản hồi phù hợp với diễn biến cuộc trò chuyện gần đây.

Giai Đoạn 4: Tích Hợp Chức Năng Công Cụ (Function Calling)

Để vượt qua giới hạn tĩnh của mô hình, ta sẽ gắn thêm các công cụ ngoại vi giúp AI tra cứu thông tin thời gian thực hoặc dữ liệu bên ngoài.

Định Nghĩa Công Cụ

import datetime
import requests
from langchain_core.tools import tool

@tool
def fetch_timestamp() -> str:
    """Trả về thời gian hiện tại theo định dạng YYYY-MM-DD HH:MM"""
    now = datetime.datetime.now()
    return now.strftime("%Y-%m-%d %H:%M:%S")

@tool
def check_weather(city_name: str) -> dict:
    """Lấy thông tin thời tiết của thành phố nhập vào"""
    api_url = f"https://uapis.cn/api/v1/misc/weather?city={city_name}"
    resp = requests.get(api_url)
    return resp.json()

active_tools = [fetch_timestamp, check_weather]

Khối Xử Lý Vòng Gọi

# Nạp config
load_dotenv()
agent_llm = ChatTongyi(model="tongyi-xiaomi-analysis-pro", temperature=0.7)
agent_llm_tool = agent_llm.bind_tools(active_tools)

chat_db = []

while True:
    u_msg = input("User: ")
    if u_msg == 'exit': break
    
    chat_db.append(HumanMessage(content=u_msg))
    raw_resp = agent_llm_tool.invoke(chat_db)
    
    # Kiểm tra xem AI có muốn dùng công cụ không
    if hasattr(raw_resp, 'tool_calls') and len(raw_resp.tool_calls) > 0:
        # Giữ lại tin nhắn gốc
        chat_db.append(raw_resp)
        
        for call_item in raw_resp.tool_calls:
            t_name = call_item['name']
            t_args = call_item['args']
            
            # Tìm hàm tương ứng
            func_ref = next((t for t in active_tools if t.name == t_name), None)
            
            if func_ref:
                exec_result = func_ref.invoke(t_args)
            else:
                exec_result = "Không tìm thấy công cụ"
            
            # Đưa kết quả công cụ vào luồng hội thoại
            chat_db.append({
                "content": str(exec_result),
                "type": "tool_call_id"
            }) 
        
        # Gọi lại mô hình để tổng hợp câu trả lời cuối cùng
        final_msg = agent_llm.invoke(chat_db)
        
    else:
        final_msg = raw_resp
    
    chat_db.append(final_msg)
    print(f"AI: {final_msg.content}\n")

Quy trình hoạt động bao gồm: AI phân tích nhu cầu -> Chọn công cụ phù hợp -> Thực thi -> Báo cáo kết quả -> Tổng hợp câu trả lời tự nhiên.

Giai Đoạn 5: Hệ Thống Tri Thức Riêng (RAG)

Trong môi trường doanh nghiệp, dữ liệu nội bộ thường nhạy cảm. Sử dụng kỹ thuật RAG (Retrieval-Augmented Generation) cho phép mô hình dựa trên tài liệu nội bộ thay vì kiến thức chung bên ngoài.

Lớp Quản Trị Kho Tài Liệu

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document

class DocIntelligenceLayer:
    def __init__(self, emb_model_id="multimodal-embedding-v1"):
        self.emb = DashScopeEmbeddings(model=emb_model_id)
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50
        )
        self.store = None
        self.data_pool = []
        self._init_data_mock()
        self._build_vector_index()

    def _init_data_mock(self):
        self.data_pool = [
            Document(page_content="Công ty TMDT A chuyên về giải pháp AI tại Hà Nội.", metadata={"dept": "info"}),
            Document(page_content="Phòng nhân sự làm việc 8h-17h thứ 2 đến thứ 6.", metadata={"dept": "hr"}),
            Document(page_content="Phần mềm quản lý v2.0 hỗ trợ xuất hóa đơn điện tử.", metadata={"dept": "product"})
        ]

    def _build_vector_index(self):
        try:
            splits = self.splitter.split_documents(self.data_pool)
            self.store = FAISS.from_documents(splits, self.emb)
        except Exception as err:
            print(f"Không thể tạo vector: {err}")

    def retrieve_info(self, query_str: str, top_k=2):
        if self.store:
            return self.store.similarity_search(query_str, k=top_k)
        return []

Tích Hợp Tìm Kiếm Vào Agent

@tool
def search_internal_docs(question: str):
    """Tìm kiếm thông tin trong kho tài liệu nội bộ"""
    kb = DocIntelligenceLayer() # Hoặc pass instance vào global
    hits = kb.retrieve_info(question)
    
    output_text = ""
    for idx, doc in enumerate(hits, 1):
        output_text += f"[{idx}] {doc.page_content}\n\n"
    return output_text

# Cập nhật danh sách công cụ
full_tools = [check_weather, search_internal_docs]
llm_final = agent_llm.bind_tools(full_tools)

session_log = []
print("Hệ thống RAG online")

while True:
    q = input("? ")
    if q == 'stop': break
    session_log.append(HumanMessage(content=q))
    
    res = llm_final.invoke(session_log)
    
    if hasattr(res, 'tool_calls'):
        session_log.append(res)
        for item in res.tool_calls:
            tool_name = item['name']
            tool_args = item['args']
            
            if tool_name == 'search_internal_docs':
                result_val = search_internal_docs.invoke(tool_args)
            elif tool_name == 'check_weather':
                result_val = check_weather.invoke(tool_args)
            else:
                result_val = "Error tool"
                
            session_log.append({"role": "tool", "content": str(result_val)})
        
        final_res = llm_final.invoke(session_log)
        session_log.append(final_res)
        print(f"Result: {final_res.content}")
    else:
        print(f"Result: {res.content}")

Mô hình này tận dụng sức mạnh của việc nhúng vector (Embedding) để tìm kiếm ngữ nghĩa chính xác, kết hợp khả năng hiểu ngôn ngữ tự nhiên của LLM nhằm cung cấp câu trả lời chi tiết và có nguồn gốc rõ ràng cho người dùng doanh nghiệp.