mirror of
https://github.com/CopilotKit/CopilotKit.git
synced 2026-09-14 16:26:20 +08:00
Markus Ecker
101 lines
3.1 KiB
Python
101 lines
3.1 KiB
Python
import os
|
|
from langchain_anthropic import ChatAnthropic
|
|
from langchain_community.tools.tavily_search import TavilySearchResults
|
|
from langchain_core.messages import AIMessage, ToolMessage
|
|
from langchain_core.runnables import RunnableConfig
|
|
# NOTE: you must use langchain-core >= 0.3 with Pydantic v2
|
|
from pydantic import BaseModel
|
|
|
|
from langgraph.checkpoint.memory import MemorySaver
|
|
from langgraph.graph import StateGraph
|
|
from langgraph.prebuilt import ToolNode, tools_condition
|
|
from copilotkit import CopilotKitState
|
|
from copilotkit.langchain import copilotkit_customize_config
|
|
|
|
|
|
class State(CopilotKitState):
|
|
# This flag is new
|
|
ask_human: bool
|
|
|
|
|
|
class RequestAssistance(BaseModel):
|
|
"""Escalate the conversation to an expert. Use this if you are unable to assist directly or if the user requires support beyond your permissions.
|
|
|
|
To use this function, relay the user's 'request' so the expert can provide the right guidance.
|
|
"""
|
|
|
|
request: str
|
|
|
|
|
|
tool = TavilySearchResults(max_results=2)
|
|
tools = [tool]
|
|
llm = ChatAnthropic(model="claude-3-5-sonnet-20240620", api_key=os.getenv("ANTHROPIC_API_KEY"))
|
|
# We can bind the llm to a tool definition, a pydantic model, or a json schema
|
|
llm_with_tools = llm.bind_tools(tools + [RequestAssistance])
|
|
|
|
|
|
def chatbot(state: State, config: RunnableConfig):
|
|
config = copilotkit_customize_config(config, emit_tool_calls="RequestAssistance")
|
|
response = llm_with_tools.invoke(state["messages"], config=config)
|
|
ask_human = False
|
|
if (
|
|
response.tool_calls
|
|
and response.tool_calls[0]["name"] == RequestAssistance.__name__
|
|
):
|
|
ask_human = True
|
|
return {"messages": [response], "ask_human": ask_human}
|
|
|
|
|
|
graph_builder = StateGraph(State)
|
|
|
|
graph_builder.add_node("chatbot", chatbot)
|
|
graph_builder.add_node("tools", ToolNode(tools=[tool]))
|
|
|
|
|
|
def create_response(response: str, ai_message: AIMessage):
|
|
return ToolMessage(
|
|
content=response,
|
|
tool_call_id=ai_message.tool_calls[0]["id"],
|
|
)
|
|
|
|
|
|
def human_node(state: State):
|
|
new_messages = []
|
|
if not isinstance(state["messages"][-1], ToolMessage):
|
|
# Typically, the user will have updated the state during the interrupt.
|
|
# If they choose not to, we will include a placeholder ToolMessage to
|
|
# let the LLM continue.
|
|
new_messages.append(
|
|
create_response("No response from human.", state["messages"][-1])
|
|
)
|
|
return {
|
|
# Append the new messages
|
|
"messages": new_messages,
|
|
# Unset the flag
|
|
"ask_human": False,
|
|
}
|
|
|
|
|
|
graph_builder.add_node("human", human_node)
|
|
|
|
|
|
def select_next_node(state: State):
|
|
if state["ask_human"]:
|
|
return "human"
|
|
# Otherwise, we can route as before
|
|
return tools_condition(state)
|
|
|
|
|
|
graph_builder.add_conditional_edges(
|
|
"chatbot",
|
|
select_next_node,
|
|
{"human": "human", "tools": "tools", "__end__": "__end__"},
|
|
)
|
|
graph_builder.add_edge("tools", "chatbot")
|
|
graph_builder.add_edge("human", "chatbot")
|
|
graph_builder.set_entry_point("chatbot")
|
|
memory = MemorySaver()
|
|
graph = graph_builder.compile(
|
|
checkpointer=memory,
|
|
interrupt_before=["human"],
|
|
) |