Spaces:
Sleeping
Sleeping
Download src/agents/base/flow.py from ABAO77/Run_code_api: direct link, hf CLI and curl.
- Browser
- Download file 859 Bytes
-
https://huggingface.co/spaces/ABAO77/Run_code_api/resolve/1ea1cfafb0da2586063e44c7e3ce001edf1022cd/src/agents/base/flow.py
- Command line
-
hf download hf://spaces/ABAO77/Run_code_api@1ea1cfafb0da2586063e44c7e3ce001edf1022cd/src/agents/base/flow.py
-
curl -L -o flow.py https://huggingface.co/spaces/ABAO77/Run_code_api/resolve/1ea1cfafb0da2586063e44c7e3ce001edf1022cd/src/agents/base/flow.py
859 Bytes
| from langgraph.graph import StateGraph, START, END | |
| from .func import State | |
| from langgraph.graph.state import CompiledStateGraph | |
| from langgraph.store.memory import InMemoryStore | |
| class PrimaryChatBot: | |
| def __init__(self): | |
| pass | |
| def should_continue(state: State): | |
| messages = state["messages"] | |
| last_message = messages[-1] | |
| if not last_message.tool_calls: | |
| return "end" | |
| else: | |
| return "continue" | |
| def node(self, graph: StateGraph): | |
| return graph | |
| def edge(self, graph: StateGraph): | |
| return graph | |
| def __call__(self, checkpointer=InMemoryStore()) -> CompiledStateGraph: | |
| graph = StateGraph(State) | |
| graph: StateGraph = self.node(graph) | |
| graph: StateGraph = self.edge(graph) | |
| return graph.compile(checkpointer=checkpointer) | |