|
|
from typing import TypedDict, Annotated |
|
|
from langgraph.graph.message import add_messages |
|
|
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage |
|
|
from langgraph.prebuilt import ToolNode |
|
|
from langgraph.graph import START, StateGraph |
|
|
from langgraph.prebuilt import tools_condition |
|
|
from langchain_huggingface import HuggingFaceEndpoint, ChatHuggingFace |
|
|
|
|
|
|
|
|
import os |
|
|
from retriever import guest_info_tool |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
HF_TOKEN = os.environ['HF_TOKEN'] |
|
|
llm = HuggingFaceEndpoint( |
|
|
repo_id="Qwen/Qwen2.5-Coder-32B-Instruct", |
|
|
huggingfacehub_api_token=HF_TOKEN, |
|
|
) |
|
|
|
|
|
|
|
|
|
|
|
chat = ChatHuggingFace(llm=llm, verbose=True) |
|
|
tools = [guest_info_tool] |
|
|
chat_with_tools = chat.bind_tools(tools) |
|
|
|
|
|
|
|
|
class AgentState(TypedDict): |
|
|
messages: Annotated[list[AnyMessage], add_messages] |
|
|
|
|
|
def assistant(state: AgentState): |
|
|
return { |
|
|
"messages": [chat_with_tools.invoke(state["messages"])], |
|
|
} |
|
|
|
|
|
|
|
|
builder = StateGraph(AgentState) |
|
|
|
|
|
|
|
|
builder.add_node("assistant", assistant) |
|
|
builder.add_node("tools", ToolNode(tools)) |
|
|
|
|
|
|
|
|
builder.add_edge(START, "assistant") |
|
|
builder.add_conditional_edges( |
|
|
"assistant", |
|
|
|
|
|
|
|
|
tools_condition, |
|
|
) |
|
|
builder.add_edge("tools", "assistant") |
|
|
alfred = builder.compile() |
|
|
|
|
|
messages = [HumanMessage(content="Tell me about our guest named 'Lady Ada Lovelace'.")] |
|
|
response = alfred.invoke({"messages": messages}) |
|
|
|
|
|
print("π© Alfred's Response:") |
|
|
print(response['messages'][-1].content) |
|
|
|
|
|
|
|
|
if __name__ == "__main__": |
|
|
GradioUI(alfred).launch() |