マルチAIエージェントのアプリをChainlitで爆速開発しよう

# せっかく作ったAIエージェントは一般ユーザにも届けたいですよね？？

皆さん、AIエージェント触ってますか？
私は最近LangGraphを触るのが楽しいです。

**せっかく作ったエージェントは一般ユーザにも届けたいですよね？？**
で、あればフロントエンドも作りたいんですが、これが私の様な素人には意外と難しいです。

APIとしてデプロイして蹴って使うのも手間ですし、Streamlitで実装するのも、
チャット履歴は？ツールを使った場合の表示はどうする？など意外と考える事が多くて面倒です。

もっとエージェント開発に注力してフロントエンドはサクッとモダンなものを実装したい...

そんな風にして調べていたらChainlitと出会ってしまいました。

※この記事の続編はこちら
[[Chainlit✖︎AWS]超簡単!?LangGraphマルチエージェントのチャット履歴をAWSクラウド上に保存しよう](https://qiita.com/asuka0708japan/items/f2a421bd145971e0ee97)
 # Chainlitとは...
 Chainlitは、PythonでChatGPTのようなユーザーインターフェース（UI）を簡単に構築できるオープンソースのパッケージです。これにより、複雑なUIの実装に時間をかけることなく、対話型AIアプリケーションの開発に集中できます。

 参考: [Chainlit概要](https://docs.chainlit.io/get-started/overview)

 認証の導入やUIのカスタマイズ、LangGraphとの統合もドキュメントで示されており、これは良いと出会った瞬間にビビッときました。
Reactとの統合が容易なのも魅力的ですよね。

 参考: [Reactとの統合](https://docs.chainlit.io/deploy/react/overview)

 # 初めてのLangGraph✖︎Chainlit
LangGraphユーザが初めて試すなら以下の例がわかりやすいです。
参考:[LangChain/LangGraphとの統合](https://docs.chainlit.io/integrations/langchain)

今までのLangGraph関係の開発環境にchainlitを追加しましょう。
例：
```bash
pip install chainlit
```

以下は天気情報を取得する関数をツールとしたAIエージェントの例です。

```python: app.py
from typing import Literal
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolNode
from langchain.schema.runnable.config import RunnableConfig
from langchain_core.messages import HumanMessage

import chainlit as cl

@tool
def get_weather(city: Literal["nyc", "sf"]):
    """Use this to get weather information."""
    if city == "nyc":
        return "It might be cloudy in nyc"
    elif city == "sf":
        return "It's always sunny in sf"
    else:
        raise AssertionError("Unknown city")


tools = [get_weather]
model = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0)
final_model = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0)

model = model.bind_tools(tools)
# NOTE: this is where we're adding a tag that we'll can use later to filter the model stream events to only the model called in the final node.
# This is not necessary if you call a single LLM but might be important in case you call multiple models within the node and want to filter events
# from only one of them.
final_model = final_model.with_config(tags=["final_node"])
tool_node = ToolNode(tools=tools)

from typing import Annotated
from typing_extensions import TypedDict

from langgraph.graph import END, StateGraph, START
from langgraph.graph.message import MessagesState
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage


def should_continue(state: MessagesState) -> Literal["tools", "final"]:
    messages = state["messages"]
    last_message = messages[-1]
    # If the LLM makes a tool call, then we route to the "tools" node
    if last_message.tool_calls:
        return "tools"
    # Otherwise, we stop (reply to the user)
    return "final"


def call_model(state: MessagesState):
    messages = state["messages"]
    response = model.invoke(messages)
    # We return a list, because this will get added to the existing list
    return {"messages": [response]}


def call_final_model(state: MessagesState):
    messages = state["messages"]
    last_ai_message = messages[-1]
    response = final_model.invoke(
        [
            SystemMessage("Rewrite this in the voice of Al Roker"),
            HumanMessage(last_ai_message.content),
        ]
    )
    # overwrite the last AI message from the agent
    response.id = last_ai_message.id
    return {"messages": [response]}


builder = StateGraph(MessagesState)

builder.add_node("agent", call_model)
builder.add_node("tools", tool_node)
# add a separate final node
builder.add_node("final", call_final_model)

builder.add_edge(START, "agent")
builder.add_conditional_edges(
    "agent",
    should_continue,
)

builder.add_edge("tools", "agent")
builder.add_edge("final", END)

graph = builder.compile()

@cl.on_message
async def on_message(msg: cl.Message):
    config = {"configurable": {"thread_id": cl.context.session.id}}
    cb = cl.LangchainCallbackHandler()
    final_answer = cl.Message(content="")
    
    for msg, metadata in graph.stream({"messages": [HumanMessage(content=msg.content)]}, stream_mode="messages", config=RunnableConfig(callbacks=[cb], **config)):
        if (
            msg.content
            and not isinstance(msg, HumanMessage)
            and metadata["langgraph_node"] == "final"
        ):
            await final_answer.stream_token(msg.content)

    await final_answer.send()

```

以下のコマンドでアプリの起動ができます。
```bash
chainlit run app.py
```
ローカルにアクセスすると以下のようなそれっぽいアプリが起動できます。
![スクリーンショット 2025-01-12 15.02.46.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3349905/613d9f45-d71c-b878-542b-0cd141d97642.png)

サンプルではニューヨークの天気を曇りと返す関数をtoolとして定義しているので、ニューヨークの天気を曇りと回答しています。

chainlitによるUI記述は下記の十数行のみです。
```python
import chainlit as cl

(graphの定義省略)

@cl.on_message
async def on_message(msg: cl.Message):
    config = {"configurable": {"thread_id": cl.context.session.id}}
    cb = cl.LangchainCallbackHandler()
    final_answer = cl.Message(content="")
    
    for msg, metadata in graph.stream({"messages": [HumanMessage(content=msg.content)]}, stream_mode="messages", config=RunnableConfig(callbacks=[cb], **config)):
        if (
            msg.content
            and not isinstance(msg, HumanMessage)
            and metadata["langgraph_node"] == "final"
        ):
            await final_answer.stream_token(msg.content)

    await final_answer.send()

```

chainlitはチャットのライフサイクルイベントをフックして制御するようになっています。
例えば、上記のデコーダー **@cl.on_message** はユーザからチャットが送信された際のフックになります。それ以外にも以下のようなフックを利用する事ができます。

-  @cl.on_chat_start
新しいセッションが作られた時
- @cl.on_stop
ユーザがストップボタンを押した時
- @cl.on_chat_end
チャットセッションが終了した時
- @cl.on_chat_resume
チャットセッションが再開した時

# マルチエージェントLangGraphのフロントエンドを作ってみる
チュートリアルを参考にシコシコ作っていたLangGraphのマルチエージェントの構成をくっ付けてみました。
参考:[LangGraph/Multi-agent supervisor](https://langchain-ai.github.io/langgraph/tutorials/multi_agent/agent_supervisor/#construct-graph)

```python
import os
from dotenv import load_dotenv
from typing import Literal, TypedDict, Annotated, Sequence

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain.schema.runnable.config import RunnableConfig
from langchain_core.messages import HumanMessage, BaseMessage
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_openai import ChatOpenAI
from langchain_experimental.utilities import PythonREPL

from langgraph.graph import MessagesStat, StateGraph, START, END
from langgraph.types import Command
from langgraph.prebuilt import create_react_agent, ToolNode

import chainlit as cl
from chainlit.input_widget import Select, Switch, Slider

load_dotenv()



tavily_tool = TavilySearchResults(max_results=2)

# This executes code locally, which can be unsafe
repl = PythonREPL()


@tool
def python_repl_tool(
    code: Annotated[str, "The python code to execute to generate your chart."],
):
    """Use this to execute python code and do math. If you want to see the output of a value,
    you should print it out with `print(...)`. This is visible to the user."""
    try:
        result = repl.run(code)
    except BaseException as e:
        return f"Failed to execute. Error: {repr(e)}"
    result_str = f"Successfully executed:\n\`\`\`python\n{code}\n\`\`\`\nStdout: {result}"
    return result_str


model = ChatOpenAI(temperature=0, streaming=True)
llm=model
from langchain.tools.render import format_tool_to_openai_function

# functions = [format_tool_to_openai_function(t) for t in tools]
# model = model.bind_functions(functions)

members = ["researcher", "coder"]
# Our team supervisor is an LLM node. It just picks the next agent to process
# and decides when the work is completed
options = members + ["FINISH"]

system_prompt = (
    "You are a supervisor tasked with managing a conversation between the"
    f" following workers: {members}. Given the following user request,"
    " respond with the worker to act next. Each worker will perform a"
    " task and respond with their results and status. When finished,"
    " respond with FINISH."
)


class Router(TypedDict):
    """Worker to route to next. If no workers needed, route to FINISH."""

    next: Literal[*options]



def supervisor_node(state: MessagesState) -> Command[Literal[*members, "__end__"]]:
    messages = [
        {"role": "system", "content": system_prompt},
    ] + state["messages"]
    response = llm.with_structured_output(Router).invoke(messages)
    goto = response["next"]
    if goto == "FINISH":
        goto = END

    return Command(goto=goto)

research_agent = create_react_agent(
    llm, tools=[tavily_tool], state_modifier="You are a researcher. DO NOT do any math."
)


def research_node(state: MessagesState) -> Command[Literal["supervisor"]]:
    result = research_agent.invoke(state)
    return Command(
        update={
            "messages": [
                HumanMessage(content=result["messages"][-1].content, name="researcher")
            ]
        },
        goto="supervisor",
    )


# NOTE: THIS PERFORMS ARBITRARY CODE EXECUTION, WHICH CAN BE UNSAFE WHEN NOT SANDBOXED
code_agent = create_react_agent(llm, tools=[python_repl_tool])


def code_node(state: MessagesState) -> Command[Literal["supervisor"]]:
    result = code_agent.invoke(state)
    return Command(
        update={
            "messages": [
                HumanMessage(content=result["messages"][-1].content, name="coder")
            ]
        },
        goto="supervisor",
    )


builder = StateGraph(MessagesState)
builder.add_edge(START, "supervisor")
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", research_node)
builder.add_node("coder", code_node)
app = builder.compile()

@cl.password_auth_callback
def auth_callback(username: str, password: str):
    # Fetch the user matching username from your database
    # and compare the hashed password with the value stored in the database
    if (username, password) == ("admin", "admin"):
        return cl.User(
            identifier="admin", metadata={"role": "admin", "provider": "credentials"}
        )
    else:
        return None

@cl.on_chat_start
async def start():
    settings = await cl.ChatSettings(
        [
            Select(
                id="Model",
                label="OpenAI - Model",
                values=["gpt-3.5-turbo", "gpt-3.5-turbo-16k", "gpt-4", "gpt-4-32k"],
                initial_index=0,
            ),
            Switch(id="Streaming", label="OpenAI - Stream Tokens", initial=True),
            Slider(
                id="Temperature",
                label="OpenAI - Temperature",
                initial=1,
                min=0,
                max=2,
                step=0.1,
            ),
            Slider(
                id="SAI_Steps",
                label="Stability AI - Steps",
                initial=30,
                min=10,
                max=150,
                step=1,
                description="Amount of inference steps performed on image generation.",
            ),
            Slider(
                id="SAI_Cfg_Scale",
                label="Stability AI - Cfg_Scale",
                initial=7,
                min=1,
                max=35,
                step=0.1,
                description="Influences how strongly your generation is guided to match your prompt.",
            ),
            Slider(
                id="SAI_Width",
                label="Stability AI - Image Width",
                initial=512,
                min=256,
                max=2048,
                step=64,
                tooltip="Measured in pixels",
            ),
            Slider(
                id="SAI_Height",
                label="Stability AI - Image Height",
                initial=512,
                min=256,
                max=2048,
                step=64,
                tooltip="Measured in pixels",
            ),
        ]
    ).send()


@cl.on_settings_update
async def setup_agent(settings):
    print("on_settings_update", settings)



@cl.on_message
async def run_convo(message: cl.Message):
    #"what is the weather in sf"
    inputs = {"messages": [HumanMessage(content=message.content)]}

    res = app.invoke(inputs, config=RunnableConfig(callbacks=[
        cl.LangchainCallbackHandler(
            to_ignore=["ChannelRead", "RunnableLambda", "ChannelWrite", "__start__", "_execute"]
            # can add more into the to_ignore: "agent:edges", "call_model"
            # to_keep=

        )]))

    await cl.Message(content=res["messages"][-1].content).send()
```

上記のコードでは認証機能や、LLMのパラメータ設定機能なども追加で設定してみています。気になる方は以下を参考にしてください。
参考：
- [パスワード認証](https://docs.chainlit.io/authentication/password)
- [チャット設定](https://docs.chainlit.io/api-reference/chat-settings)

こんなアプリがサクッとできます。
![スクリーンショット 2025-01-12 15.42.15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3349905/864d687b-0150-23a0-db18-e7fa906df03b.png)
今月って言ってるのに3月4月のイベントを出してくるあたりまだまだですが、フロントエンドの実装が圧倒的に楽なのがお分かりいただけるでしょうか。

# 圧倒的にエージェント開発に注力できる
フロントエンドへの労力を抑えて圧倒的にエージェント側の開発に注力できるのがとても魅力的だと思います。皆さんもサクッと触ってみませんか？