Bỏ túi LCEL: Ngôn ngữ lập trình GenAI độc đáo

LCEL là gì? LCEL có gì hot?

Trong thời đại số hóa hiện nay, sự phổ biến của các hệ thống ứng dụng GenAI đã mở ra một chân trời mới về công nghệ, đánh dấu bước tiến quan trọng trong việc giải quyết các vấn đề phức tạp trong nhiều lĩnh vực. Tuy nhiên, bất chấp sự phát triển mạnh mẽ và tiềm năng to lớn mà GenAI mang lại, các hệ thống và ứng dụng này cũng đối mặt với những thách thức đáng kể. Hệ thống ứng dụng GenAI thường gặp những khó khăn như:

• Khó thiết kế luồng xử lý: Trong các hệ thống phức tạp, việc mỗi component sử dụng phương thức khác nhau tạo ra rào cản lớn trong việc kết nối và tương tác giữa các thành phần.
• Chờ hồi đáp lâu: Trong các ứng dụng chat sử dụng GenAI, việc chờ đợi kết quả trả về thường tạo ra trải nghiệm không "mượt", không "thật" cho người dùng.
• Hiệu suất kém: Trong một số hệ thống cần truy vấn tài liệu tham khảo (như RAG) hoặc lấy dữ liệu từ 3rd party API, yêu cầu xử lý song song để tăng hiệu suất là vô cùng quan trọng nhưng thường khó triển khai.
• Khó dùng custom function: Việc tích hợp các hàm xử lý logic vào framework có sẵn thường gặp khó khăn do thiếu khả năng tương thích hoặc phức tạp trong cách tích hợp.
• Debugging và Logging: Trong quá trình phát triển, việc theo dõi thông tin xử lý trung gian và handle các sự kiện đặc thù thường tạo ra thách thức lớn.
• Hạn chế tùy biến: Độ tùy biến là một trong những giá trị quan trọng cho những hệ thống hướng platform, nhu cầu tùy chỉnh và thích ứng với yêu cầu mới là rất cần thiết. Tuy nhiên, việc này thường gặp hạn chế do thiếu sự linh hoạt trong cấu hình và mở rộng.

Trong bối cảnh trên, LangChain Expression Language (LCEL) - một framework dạng khai báo (declarative) - giúp chuyển các đoạn code phức tạp thành cú pháp đơn giản, tạo điều kiện phát triển nhanh chóng và linh hoạt các luồng xử lý trong ứng dụng Generative AI. Nhìn chung, LCEL hỗ trợ các khả năng:

• Interface đồng nhất: LCEL giới thiệu Runnable interface, một giải pháp đồng nhất cho phương thức gọi cho các component, mở ra khả năng thiết kế luồng xử lý dễ dàng, linh hoạt. Xem thêm Sử dụng cơ bản.
• Streaming: Khả năng streaming kết quả của LCEL giúp giảm thiểu đáng kể thời gian chờ đợi, mang lại trải nghiệm tốt. Đối với những ứng dụng cần kết quả trả về dưới dạng JSON, LCEL cũng cung cấp tính năng auto-complete, tối ưu thời gian xử lý của Generative AI. Xem thêm Khả năng Streaming.
• Xử lý song song: LCEL cho phép xây dựng chuỗi thực thi các tác vụ song song một cách đơn giản và trực quan. Điều này giúp tăng hiệu suất xử lý đáng kể, đặc biệt trong các luồng xử lý nhiều bước. Xem thêm RAG - Retrieval-Augmented Generation.
• Tích hợp custom function: Với khả năng hỗ trợ tích hợp hàm tự viết qua hàm wrapper đơn giản, LCEL mở ra cơ hội tùy chỉnh cao, cho phép người lập trình dễ dàng thêm các xử lý chuyên biệt vào trong luồng xử lý tổng thể. Xem thêm Branching và Merging.
• Debugging và Logging: LCEL cung cấp các công cụ mạnh mẽ để theo dõi log và debug thông qua việc ghi chép sự kiện, giúp nhà phát triển dễ dàng theo dõi và khắc phục lỗi. Xem thêm Debug chuỗi xử lý.
• Tùy chỉnh độ tùy biến: LCEL nổi bật với khả năng tùy chỉnh độ tùy biến cao, từ việc tùy chỉnh model, prompt cho đến việc thêm các tùy chỉnh mới vào chuỗi xử lý. Điều này không chỉ giúp phục vụ tốt cho các mục đích logic và testing mà còn mở rộng khả năng ứng dụng của hệ thống trong các tình huống cụ thể. Xem thêm Cấu hình tùy chỉnh.

Bài viết này sẽ đưa ra góc nhìn tổng quan về LCEL thông qua các use-case, trong đó chú trọng vào việc xây dựng các Chain - chuỗi xử lý.

Đọc thêm: LangChain: Công cụ không thể bỏ qua trong thời đại AI tạo sinh

Demo

Sử dụng cơ bản

Ta sẽ bắt đầu với việc tạo một chain cơ bản:

Chain sử dụng cơ bản

Với chain này, ta tạo các component:

import os
from langchain.prompts.prompt import PromptTemplate
from langchain_openai import AzureChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# =========== Prompt
_template = """Given the feedback, analyze and detect the sentiment:

<feedback>
{feedback}
</feedback>

Only output sentiment: Negative, Neutral, Positive
Sentiment:"""
prompt = PromptTemplate.from_template(_template)

# =========== Model
os.environ["AZURE_OPENAI_API_KEY"] = ""
os.environ["AZURE_OPENAI_ENDPOINT"] = ""
os.environ["OPENAI_API_VERSION"] = ""
model_deployment_name = ""
model = AzureChatOpenAI(deployment_name=model_deployment_name)

# =========== Output Parser: We will use StrOutputParser to parse the output to String

Sau đó, ta tạo chain và thực thi:

# =========== Chain
sentiment_chain = prompt | model | StrOutputParser()

print(sentiment_chain.invoke({"feedback": "I found the service quite impressive, though the ambiance was a bit lacking."}))

Qua đây, ta thấy được cách tạo các thành phần và thực thi chain. Điểm đặc biệt là sử dụng toán tử | giúp tạo ra một chuỗi các thao tác thực thi tuần tự, gọi là RunnableSequence.

Bên cạnh đó, ta cũng thấy được một số component của LCEL như Prompt, ChatModel, OutputParser có thể “kết nối” với nhau trong chain. Đó là nhờ Runnable Interface.

Khả năng Streaming

Trong các ứng dụng GenAI, LLM/ChatModel chính là nút thắt cổ chai trong hệ thống. Do đó, việc streaming trở nên cần thiết trong các ứng dụng sử dụng LLM trong quá trình tương tác với người dùng. Khi sử dụng LCEL, ta có thể dễ dàng sử dụng hàm stream() / astream() để lấy kết quả dưới dạng streaming. Những component không hỗ trợ streaming sẽ không ảnh hưởng đến khả năng streaming của những thành phần khác.

for chunk in sentiment_chain.stream(
    {"feedback": "I found the service quite impressive, though the ambiance was a bit lacking."}
):
    print(chunk, end="|", flush=True)

RAG - Retrieval-Augmented Generation

Đối với chain cho Retrieval-Augmented Generation sẽ bao gồm việc sử dụng retriever để truy vấn tài liệu tham khảo:

Chain cho Retrieval-Augmented Generation

Với chain này, ta tạo các component:

import os
from operator import itemgetter
from langchain_openai import AzureChatOpenAI, AzureOpenAIEmbeddings
from langchain.prompts.prompt import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.vectorstores import FAISS

# =========== Retriever to reference data
vectorstore = FAISS.from_texts(
    ["Customer Support experiences: Sarah",
     "Product Service feedback: Michael"
     "Website Usability: Alex"], 
     embedding=AzureOpenAIEmbeddings(azure_deployment=embed_deployment_name)
)
retriever = vectorstore.as_retriever()

# =========== (New) Prompt
_template = """Given the feedback, analyze and detect the PIC to resolve based on PIC list:

PIC list:
{document}

Feedback:
<feedback>
{feedback}
</feedback>

Only output the PIC
PIC:"""

prompt = PromptTemplate.from_template(_template)

# =========== Model
model = AzureChatOpenAI(deployment_name=model_deployment_name)

Sau đó, ta tạo chain và thực thi:

# =========== Chains
ref_data_chain = itemgetter("feedback") | retriever
pic_chain = (
    {"document": ref_data_chain, 
     "feedback": itemgetter("feedback")}
    | prompt | model | StrOutputParser()
)

print(pic_chain.invoke({"feedback": "I had trouble navigating the website to find information on product returns, which was quite frustrating. However, once I reached out to customer support, it was incredibly helpful and resolved my issue promptly."}))

Trong minh họa trên, ta thấy được các sử dụng hàm itemgetter() để truy xuất các thành phần có trong input của component liền trước trong chain, và một component không thể thiếu trong các ứng dụng RAG là Retriever.

Branching và Merging

Qua 2 minh họa trên, ta đã có một chain phân tích sentiment và một chain tìm PIC - người phụ trách cho feedback. Tiếp theo, ta sẽ thực hiện chúng song song, giúp tiết kiệm thời gian và tăng hiệu quả công việc:

Chain cho branching và merging

Ta tạo chain mới như sau:

# =========== Custom action
def _send_email(sentiment, pic):
    print(f"Send email about {sentiment} feedback for {pic}")
def send_email(_dict):
    _send_email(_dict["sentiment"], _dict["pic"])
    return "OK"

# =========== Chain
combine_chain = (
    {"feedback": RunnablePassthrough()}
    | RunnableParallel(sentiment=sentiment_chain, pic=pic_chain)
    | {"sentiment": itemgetter("sentiment"), "pic": itemgetter("pic")}
    | RunnableLambda(send_email)
)

combine_chain.invoke("I had trouble navigating the website to find information on product returns, which was quite frustrating. However, once I reached out to customer support, it was incredibly helpful and resolved my issue promptly.")

Trong minh họa trên, ta thấy được:

• RunnableParallel: Dùng để thực thi các chain song song. Ngoài ra RunnableParallel còn dùng để điều chỉnh output của mắt xích trước sao cho phù hợp với input của mắt xích kế tiếp
• RunnablePassthrough: Truyền dữ liệu tới mắt xích kế tiếp. Có thể thêm dữ liệu vào RunnablePassthrough qua hàm assign()
• RunnableLambda: Thực thi custom function như một mắt xích

Routing Logic

Đôi khi ta cần thêm logic để lựa chọn chuỗi xử lý phù hợp. LCEL cung cấp khả năng này thông qua việc sử dụng custom function hoặc RunnableBranch, giúp bạn dễ dàng điều hướng luồng xử lý của mình.

Chain cho Routing Logic

Trong bài này, ta thay đổi chain như sau:

# =========== Customer action for general case
def general_action(_dict):
    print("General action is taken")
    
# =========== Routing logic
def route(_dict):
    if "negative" in _dict['sentiment'].lower():
        return negative_chain
    else:
        return general_chain

# =========== Define sub chains
analyze_sentiment_chain = {"feedback": RunnablePassthrough()} | sentiment_chain | StrOutputParser()
negative_chain = RunnablePassthrough.assign(pic=pic_chain) | RunnableLambda(send_email)
general_chain = RunnableLambda(general_action)

# =========== Main chain
combine_chain = (
    {
        "sentiment": analyze_sentiment_chain, 
        "feedback": RunnablePassthrough()
    }
    | RunnableLambda(route)
)

combine_chain.invoke("I was disappointed with the customer service received; the representative seemed uninterested and my issue remained unresolved. The overall experience was far from what I expected based on previous reviews.")

Qua minh họa trên, ta thấy được cách sử dụng custom function để thêm routing logic vào chain. Một cách khác nữa là sử dụng RunnableBranch, tuy nhiên, custom function được LangChain khuyến khích cho mục đích routing. Bên cạnh đó, ta cũng biết về cách sử dụng RunnablePassthrough.assign() để thêm dữ liệu vào mắt xích kế tiếp.

Debug chuỗi xử lý

Hầu hết các component trong LangChain đều hiện thực hóa Runnable. Interface này không chỉ có các hàm giúp mắt xích tương tác với nhau dễ dàng như invoke(), stream(), batch()và các phiên bản async như ainvoke(), astream(), abatch()mà còn có các hàm hỗ trợ như astream_log() và astream_events().

Ta có thể cập nhật các chain trong demo ở phần trước từ sử dụng invoke() sang sử dụng astream_log()để xem các bước xử lý trung gian:

async def process_log():
    async for chunk in combine_chain.astream_log(
        'I was disappointed with the customer service received; the representative seemed uninterested and my issue remained unresolved. The overall experience was far from what I expected based on previous reviews.',
    ):
        print("-" * 40)
        print(chunk)

asyncio.run(process_log())

Hoặc sử dụng astream_event() để xem các events xảy ra trong quá trình chain thực thi:

async def process_events():
    async for event in combine_chain.astream_events(
        'I was disappointed with the customer service received; the representative seemed uninterested and my issue remained unresolved. The overall experience was far from what I expected based on previous reviews.',
        version="v1",
    ):
        print(event)

asyncio.run(process_events())

Bên cạnh đó, LCEL cũng hỗ trợ hàm get_graph() để xem cấu trúc của chain:

combine_chain.get_graph().print_ascii()

Chain Graph

Cấu hình tùy chỉnh

Một tính năng đa dụng của LCEL là configurable_alternatives(). Hàm này cho phép ta thêm các lựa chọn tùy chỉnh cho chain để tăng độ linh hoạt tùy theo mục đích. Điều này cực kỳ hữu ích cho việc phát triển và kiểm thử ứng dụng.

Trong minh họa đầu tiên, ta có thể cập nhật các component như sau:

# =========== Prompt v1
_sentiment_template = """Given the feedback, analyze and detect the sentiment:

Feedback:
<feedback>
{feedback}
</feedback>

Only output sentiment: Negative, Neutral, Positive
Sentiment:"""

# =========== Prompt v2
_sentiment_analysis_prompt = """Analyze the sentiment of the following user feedback, and categorize it as either Negative, Neutral, or Positive. Do not include any additional information or explanation in your response.

User Feedback:
"{feedback}"

Determine the sentiment of the feedback based solely on its content and context. Your response should be concise, limited to one of the three specified sentiment categories.

Sentiment: """

# =========== Configure alternative prompt
sentiment_prompt = PromptTemplate.from_template(_sentiment_template).configurable_alternatives(
    ConfigurableField(id="prompt"),
    default_key="sentiment_v1",
    # This adds a new option, with name `sentiment_v2`
    sentiment_v2=PromptTemplate.from_template(_sentiment_analysis_prompt),
)

from langchain_google_vertexai import ChatVertexAI
from google.oauth2 import service_account

...

# =========== Configure alternative model
sentiment_model = AzureChatOpenAI(deployment_name=model_deployment_name).configurable_alternatives(
    ConfigurableField(id="llm"),
    default_key="azureopenai",
    # This adds a new option, with name `vertex` that is equal to `AzureChatOpenAI()`
    vertex=ChatVertexAI(project=project_name, credentials=credentials)
)

Khi thực thi chain, ta có thể thêm các tùy chỉnh với hàm with_config():

print(sentiment_chain.with_config(configurable={"prompt": "sentiment_v2", "llm": "vertex"})
      .invoke({"feedback": "I found the service quite impressive, though the ambiance was a bit lacking."}))

Như vậy, trong quá trình vận hành, ta có thể tùy chỉnh các thông số của chain tùy theo mục đích của mình.

Những keyword đáng chú ý

• Runnable: interface “cốt lõi” của hầu hết các component, giúp kết nối các component thành một mắt xích.
• RunnableSequence (toán tử |) /RunnableParallel: Cho phép bạn tạo chuỗi thực thi tuần tự hoặc song song.
• RunnablePassthrough: giúp truyền dữ liệu giữa các mắt xích, có thể thêm dữ liệu mới bằng hàm assign().
• RunnableLambda: giúp biến custom function thành một mắt xích trong chain.
• configurable_alternatives: tính năng cho phép thêm các lựa chọn tùy chỉnh cho chuỗi xử lý, tăng cường độ linh hoạt cho các tình huống cụ thể hoặc thử nghiệm A/B testing.

Tìm hiểu các tính năng khác được hỗ trợ bởi LCEL tại đây.