せっかく作ったAIエージェントは一般ユーザにも届けたいですよね？？
皆さん、AIエージェント触ってますか？
私は最近LangGraphを触るのが楽しいです。
せっかく作ったエージェントは一般ユーザにも届けたいですよね？？
で、あればフロントエンドも作りたいんですが、これが私の様な素人には意外と難しいです。
APIとしてデプロイして蹴って使うのも手間ですし、Streamlitで実装するのも、
チャット履歴は？ツールを使った場合の表示はどうする？など意外と考える事が多くて面倒です。
もっとエージェント開発に注力してフロントエンドはサクッとモダンなものを実装したい...
そんな風にして調べていたらChainlitと出会ってしまいました。
※この記事の続編はこちら
[Chainlit✖︎AWS]超簡単!?LangGraphマルチエージェントのチャット履歴をAWSクラウド上に保存しよう
Chainlitとは...
Chainlitは、PythonでChatGPTのようなユーザーインターフェース（UI）を簡単に構築できるオープンソースのパッケージです。これにより、複雑なUIの実装に時間をかけることなく、対話型AIアプリケーションの開発に集中できます。
参考: Chainlit概要
認証の導入やUIのカスタマイズ、LangGraphとの統合もドキュメントで示されており、これは良いと出会った瞬間にビビッときました。
Reactとの統合が容易なのも魅力的ですよね。
参考: Reactとの統合
初めてのLangGraph✖︎Chainlit
LangGraphユーザが初めて試すなら以下の例がわかりやすいです。
参考:LangChain/LangGraphとの統合
今まで、LangGraph関係の開発環境にchainlitを追加しましょう。
例：
pip install chainlit
以下は天気情報を取得する関数をツールとしたAIエージェントの例です。
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()
以下のコマンドでアプリの起動ができます。
chainlit run app.py
ローカルにアクセスすると以下のようなそれっぽいアプリが起動できます。
サンプルではニューヨークの天気を曇りと返す関数をtoolとして定義しているので、ニューヨークの天気を曇りと回答しています。
chainlitのコード部分は以下の部分のみです。
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
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のパラメータ設定機能なども追加で設定してみています。気になる方は以下を参考にしてください。
参考：
こんなアプリがサクッとできます。
今月って言ってるのに3月4月のイベントを出してくるあたりまだまだですが、フロントエンドの実装が圧倒的に楽なのがお分かりいただけるでしょうか。
圧倒的にエージェント開発に注力できる
フロントエンドへの労力を抑えて圧倒的にエージェント側の開発に注力できるのがとても魅力的だと思います。皆さんもサクッと触ってみませんか？