lda/langgraph-demo
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tung tung tung

Browse Source
This commit is contained in:
lda
2026-03-31 01:12:56 +07:00 Unverified
parent 67f52b41c0
commit 2dbd3de074
4 changed files with 175 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+5
View File
@@ -1,3 +1,8 @@
# what the helly
you needa setup langsmith tracing or something and groq/celebras/openrouter if you want idfk
## what the demo
this is what graph is all about. and if just for graph stuff, this is Bullshit. i choose effectjs any day of the week. And effectjs is not a graph focused thing unlike this thing with the name
however this does have some funny stuff you can do with `interrupt` and
graph.md
+20 -13
View File
@@ -9,29 +9,37 @@ config:
graph TD;
__start__([<p>__start__</p>]):::first
init(init)
tick(tick)
counter_up(counter_up)
rate_up(rate_up)
rate_same(rate_same)
rate_guarantee(rate_guarantee)
r_g10(r_g10)
r_g80(r_g80)
r_gs(r_gs)
c_80_p(c_80_p)
c_10_p(c_10_p)
prep(prep)
roll(roll)
c_80(c_80)
c_10(c_10)
counter_up(counter_up)
tick(tick)
end_roll(end_roll)
__end__([<p>__end__</p>]):::last
__start__ --> init;
c_10 --> counter_up;
c_80 --> counter_up;
counter_up -.-> __end__;
counter_up -.-> tick;
init --> tick;
c_10 --> end_roll;
c_10_p --> prep;
c_80 --> end_roll;
c_80_p --> c_10_p;
counter_up -. &nbsp;0&nbsp; .-> rate_same;
counter_up -. &nbsp;65&nbsp; .-> rate_up;
end_roll -.-> __end__;
end_roll -.-> tick;
init -.-> __end__;
init -.-> tick;
prep --> roll;
r_g10 --> prep;
r_g80 --> prep;
r_gs --> prep;
r_g10 --> c_10_p;
r_g80 --> c_80_p;
r_gs --> c_80_p;
rate_guarantee -. &nbsp;1&nbsp; .-> prep;
rate_guarantee -. &nbsp;10&nbsp; .-> r_g10;
rate_guarantee -. &nbsp;80&nbsp; .-> r_g80;
@@ -40,9 +48,8 @@ graph TD;
rate_up --> rate_guarantee;
roll -. &nbsp;10&nbsp; .-> c_10;
roll -. &nbsp;240&nbsp; .-> c_80;
roll -. &nbsp;1&nbsp; .-> counter_up;
tick -. &nbsp;0&nbsp; .-> rate_same;
tick -. &nbsp;65&nbsp; .-> rate_up;
roll -. &nbsp;1&nbsp; .-> end_roll;
tick --> counter_up;
classDef default fill:#f2f0ff,line-height:1.2
classDef first fill-opacity:0
classDef last fill:#bfb6fc
graph.mermaid
+52
View File
@@ -0,0 +1,52 @@
---
config:
flowchart:
curve: linear
---
graph TD;
__start__([<p>__start__</p>]):::first
init(init)
tick(tick)
counter_up(counter_up)
rate_up(rate_up)
rate_same(rate_same)
rate_guarantee(rate_guarantee)
r_g10(r_g10)
r_g80(r_g80)
r_gs(r_gs)
c_80_p(c_80_p)
c_10_p(c_10_p)
prep(prep)
roll(roll)
c_80(c_80)
c_10(c_10)
end_roll(end_roll)
__end__([<p>__end__</p>]):::last
__start__ --> init;
c_10 --> end_roll;
c_10_p --> prep;
c_80 --> end_roll;
c_80_p --> c_10_p;
counter_up -. &nbsp;0&nbsp; .-> rate_same;
counter_up -. &nbsp;65&nbsp; .-> rate_up;
end_roll -.-> __end__;
end_roll -.-> tick;
init -.-> __end__;
init -.-> tick;
prep --> roll;
r_g10 --> c_10_p;
r_g80 --> c_80_p;
r_gs --> c_80_p;
rate_guarantee -. &nbsp;1&nbsp; .-> prep;
rate_guarantee -. &nbsp;10&nbsp; .-> r_g10;
rate_guarantee -. &nbsp;80&nbsp; .-> r_g80;
rate_guarantee -. &nbsp;240&nbsp; .-> r_gs;
rate_same --> rate_guarantee;
rate_up --> rate_guarantee;
roll -. &nbsp;10&nbsp; .-> c_10;
roll -. &nbsp;240&nbsp; .-> c_80;
roll -. &nbsp;1&nbsp; .-> end_roll;
tick --> counter_up;
classDef default fill:#f2f0ff,line-height:1.2
classDef first fill-opacity:0
classDef last fill:#bfb6fc
main.py
+97 -50
View File
@@ -4,12 +4,13 @@ import operator
import random
from pathlib import Path
from pprint import pprint
from typing import Annotated, Literal, Sequence, TypedDict, cast
from typing import Annotated, Literal, TypedDict, cast
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, StateGraph
from langgraph.graph.state import RunnableConfig
from langgraph.graph.state import Command, RunnableConfig
from langgraph.runtime import Runtime
# from langsmith import traceable
# state, which is the state of the entire thing? idfk.
@@ -28,8 +29,10 @@ class SophisticatedRates(TypedDict):
r_240: float # 1 sometimes, 0.5, 0.004 mf
class Input(TypedDict):
class Input(TypedDict, total=False):
countdown: int
simple_counter: int
counter: SophisticatedCounter
class State(TypedDict):
@@ -39,7 +42,7 @@ class State(TypedDict):
rates: Annotated[SophisticatedRates, operator.or_]
current_pools: list[PoolByCategory]
this: Entity
storage: Annotated[list[Entity], operator.add]
storage: list[Entity]
class Entity(TypedDict):
@@ -64,15 +67,22 @@ class UnsophisticatedPool(TypedDict):
class Context(TypedDict):
pool: UnsophisticatedPool
initial_rates: SophisticatedRates
type: Literal["banner", "select"]
"very convoluted logic"
# @traceable(name="initializer")
def init(state: Input, runtime: Runtime[Context]):
if (
runtime.context["type"] == "select"
and runtime.context["initial_rates"]["r_240"] != 0
):
raise ValueError("set r_240 to 0 if select and put them also in the n_80 pool")
c = state.get("counter", {})
return {
"countdown": state["countdown"],
"simple_counter": 0,
"counter": {"c_10": 0, "c_80": 0},
"rates": runtime.context["initial_rates"].copy(),
"countdown": state.get("countdown", 0), # minus 1 logic!
"counter": {"c_10": c.get("c_10", 0), "c_80": c.get("c_80", 0)},
"rates": runtime.context["initial_rates"].copy(), # override!
}
@@ -85,31 +95,34 @@ def rate_booster(state: State) -> Literal["0", "65"]:
# @traceable(name="pre-roll router")
def router(state: State) -> Sequence[Literal["240", "80", "10", "1"]]:
def router(state: State, runtime: Runtime[Context]) -> Literal["240", "80", "10", "1"]:
"super match"
sc, c = state["simple_counter"], state["counter"]
if sc == 120 or (sc > 0 and sc % 240 == 0):
return ["240", "80", "10"]
if (runtime.context["type"] == "banner" and sc == 120) or (
sc > 0 and sc % 240 == 0
):
return "240"
if c.get("c_80", 0) == 80:
return ["80", "10"]
return "80"
if c.get("c_10", 0) == 10:
return ["10"]
return ["1"]
return "10"
return "1"
class RateChange:
# @traceable(name="force 6* rating")
@staticmethod
def r80(state: State, runtime: Runtime[Context]):
def r80(state: State):
return {"rates": {"r_1": 0, "r_10": 0}}
# @traceable(name="force banner rating")
@staticmethod
def r240(state: State, runtime: Runtime[Context]):
return {"rates": {"r_1": 0, "r_10": 0, "r_80": 0}}
def r240(state: State):
return {"rates": {"r_1": 0, "r_10": 0, "r_80": 0, "r_240": 1}}
# @traceable(name="force 5*+ rating")
@staticmethod
def r10(state: State, runtime: Runtime[Context]):
def r10(state: State):
return {"rates": {"r_1": 0}}
# @traceable(name="buff 6* rating")
@@ -121,7 +134,7 @@ class RateChange:
n = c["c_80"] - 64
rpn = n * 0.05
br = runtime.context["initial_rates"]
if br["r_240"]:
if runtime.context["type"] == "banner":
r240 = br["r_240"] * (1 + rpn / 2)
r80 = br["r_80"] * (1 + rpn / 2)
else:
@@ -140,7 +153,7 @@ class RateChange:
# @traceable(name="reset rating")
@staticmethod
def r0(state: State, runtime: Runtime[Context]):
return {"rates": runtime.context["initial_rates"]}
return {"rates": runtime.context["initial_rates"].copy()}
class CounterUp:
@@ -168,7 +181,7 @@ class CounterUp:
c = state["counter"]
return {
"simple_counter": 1,
"counter": {"c_10": c["c_10"] + 1, "c_80": c["c_80"] + 1},
"counter": {"c_10": (c["c_10"] + 1) % 10, "c_80": (c["c_80"] + 1) % 80},
}
@@ -182,21 +195,20 @@ def prep(state: State, runtime: Runtime[Context]):
("80", "r_80", "n_80"),
("240", "r_240", "n_240"),
)
return {
"current_pools": [
PoolByCategory(pool=p[pc], category=ty, rates=r[pr]) for ty, pr, pc in t
]
}
pbc = [PoolByCategory(pool=p[pc], category=ty, rates=r[pr]) for ty, pr, pc in t]
# print(pbc)
return {"current_pools": pbc}
# @traceable(name="pull from pool")
def roll(state: State):
r = state["current_pools"]
# print(r)
(t,) = random.choices(r, weights=[*map(operator.itemgetter("rates"), r)])
this = Entity(category=t["category"], name=random.choice(t["pool"]))
return {
"this": this,
"storage": [this],
"storage": [*state.get("storage", []), this],
}
@@ -206,33 +218,39 @@ def post_roll_router(state: State) -> Literal["240", "80", "10", "1"]:
# @traceable(name="main")
def tick(state: State):
def tick(state: State) -> dict | Command[END]: # type: ignore
return {"countdown": state["countdown"] - 1}
# @traceable(name="loop if countdown")
def keep_rolling(state: State):
return "tick" if state["countdown"] > 0 else END
return "tick" if state.get("countdown", 0) > 0 else END
graph = (
StateGraph(State, context_schema=Context, input_schema=Input)
.add_node(init)
.add_node("tick", tick)
.add_node("counter_up", CounterUp.c1)
.add_node("rate_up", RateChange.r65)
.add_node("rate_same", RateChange.r0)
.add_node("rate_guarantee", lambda state: {})
.add_node("r_g10", RateChange.r10)
.add_node("r_g80", RateChange.r80)
.add_node("r_gs", RateChange.r240)
.add_node("c_80_p", CounterUp.c80)
.add_node("c_10_p", CounterUp.c10)
.add_node(prep)
.add_node(roll)
.add_node("c_80", CounterUp.c80)
.add_node("c_10", CounterUp.c10)
.add_node("counter_up", CounterUp.c1)
.add_node("tick", tick)
.add_node("end_roll", lambda state: {})
.set_entry_point("init")
.add_edge("init", "tick")
.add_conditional_edges("tick", rate_booster, {"0": "rate_same", "65": "rate_up"})
.add_conditional_edges("init", keep_rolling, ["tick", END])
.add_edge("tick", "counter_up")
.add_conditional_edges(
"counter_up", rate_booster, {"0": "rate_same", "65": "rate_up"}
)
.add_edge("rate_up", "rate_guarantee")
.add_edge("rate_same", "rate_guarantee")
.add_conditional_edges(
@@ -241,17 +259,19 @@ graph = (
{"240": "r_gs", "80": "r_g80", "10": "r_g10", "1": "prep"},
)
.add_edge("prep", "roll")
.add_edge("r_gs", "prep")
.add_edge("r_g80", "prep")
.add_edge("r_g10", "prep")
.add_edge("r_gs", "c_80_p")
.add_edge("r_g80", "c_80_p")
.add_edge("r_g10", "c_10_p")
.add_edge("c_80_p", "c_10_p")
.add_edge("c_10_p", "prep")
.add_conditional_edges(
"roll",
post_roll_router,
{"240": "c_80", "80": "c_80", "10": "c_10", "1": "counter_up"},
{"240": "c_80", "80": "c_80", "10": "c_10", "1": "end_roll"},
)
.add_edge("c_80", "counter_up")
.add_edge("c_10", "counter_up")
.add_conditional_edges("counter_up", keep_rolling, ["tick", END])
.add_edge("c_80", "end_roll")
.add_edge("c_10", "end_roll")
.add_conditional_edges("end_roll", keep_rolling, ["tick", END])
)
app = graph.compile(checkpointer=InMemorySaver())
@@ -276,7 +296,7 @@ context: Context = {
"Alesh",
],
"n_80": [
"TangTang",
# "Tangtang",
"Yvonne",
"Gilberta",
"Laevatain",
@@ -286,29 +306,56 @@ context: Context = {
"Last Rite",
"Pogranichnik",
],
"n_240": ["Rossi"],
"n_240": [ # select or banner / logic is hella flawed lowk ong
# "Rossi",
"Tangtang",
],
},
"type": "banner",
}
if __name__ == "__main__":
from dotenv import load_dotenv
load_dotenv()
(Path(__file__).parent / "graph.md").write_text(f"""# graph
mm = app.get_graph().draw_mermaid()
folder = Path(__file__).parent
(folder / "graph.md").write_text(f"""# graph
```mermaid
{app.get_graph().draw_mermaid()}
{mm}
```
""") # good
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
""") # ahh
(folder / "graph.mermaid").write_text(mm)
config: RunnableConfig = {"configurable": {"thread_id": "random ahh"}}
print("running:")
random.seed("6741") # tung tung tung
pprint(
app.invoke(
{"countdown": 100},
{
"countdown": 20,
"simple_counter": 120 - 9,
"counter": {
"c_10": 10 - 10,
"c_80": 80 - 71,
# wow i got 9 whole rolls after the SECOND failed 50/50!
},
},
config,
context=context,
stream_mode="updates",
),
)
v = app.get_state(config).values
pprint(v)
print()
print("app state after run")
v = app.get_state(config)
pprint(v._asdict())
# print()
# print("value")
# pprint(v.values)
# h = app.get_state_history(config)
# print()
# print("hist")
# for i, hi in enumerate(h):
# print(f"hist step {i}")
# pprint(hi._asdict())