|
|
import random |
|
|
from typing import Literal |
|
|
|
|
|
from langchain_core.messages import AIMessage |
|
|
from langgraph.graph import START, MessagesState, StateGraph |
|
|
from langgraph.types import Command |
|
|
|
|
|
|
|
|
class AgentState(MessagesState, total=False): |
|
|
"""`total=False` is PEP589 specs. |
|
|
|
|
|
documentation: https://typing.readthedocs.io/en/latest/spec/typeddict.html#totality |
|
|
""" |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def node_a(state: AgentState) -> Command[Literal["node_b", "node_c"]]: |
|
|
print("Called A") |
|
|
value = random.choice(["a", "b"]) |
|
|
goto: Literal["node_b", "node_c"] |
|
|
|
|
|
if value == "a": |
|
|
goto = "node_b" |
|
|
else: |
|
|
goto = "node_c" |
|
|
|
|
|
|
|
|
return Command( |
|
|
|
|
|
update={"messages": [AIMessage(content=f"Hello {value}")]}, |
|
|
|
|
|
goto=goto, |
|
|
) |
|
|
|
|
|
|
|
|
def node_b(state: AgentState): |
|
|
print("Called B") |
|
|
return {"messages": [AIMessage(content="Hello B")]} |
|
|
|
|
|
|
|
|
def node_c(state: AgentState): |
|
|
print("Called C") |
|
|
return {"messages": [AIMessage(content="Hello C")]} |
|
|
|
|
|
|
|
|
builder = StateGraph(AgentState) |
|
|
builder.add_edge(START, "node_a") |
|
|
builder.add_node(node_a) |
|
|
builder.add_node(node_b) |
|
|
builder.add_node(node_c) |
|
|
|
|
|
|
|
|
command_agent = builder.compile() |
|
|
|