this logic is wrong cuz aint no way

This commit is contained in:
lda committed 2026-04-16 03:20:26 +07:00 Verified
1 parent 08f5bd2d47
commit e33ce762b3
1 file changed
+93 -36
+93 -36
View File
@@ -2,9 +2,10 @@
import operator import operator
import random import random
from collections import Counter
from pathlib import Path from pathlib import Path
from pprint import pprint from pprint import pprint
from typing import Annotated, Literal, TypedDict, cast from typing import TYPE_CHECKING, Annotated, Any, Literal, TypedDict, cast
from langgraph.checkpoint.memory import InMemorySaver from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, StateGraph from langgraph.graph import END, StateGraph
@@ -41,9 +42,9 @@ class State(TypedDict):
simple_counter: Annotated[int, operator.add] # only up! simple_counter: Annotated[int, operator.add] # only up!
counter: Annotated[SophisticatedCounter, operator.or_] counter: Annotated[SophisticatedCounter, operator.or_]
rates: Annotated[SophisticatedRates, operator.or_] rates: Annotated[SophisticatedRates, operator.or_]
current_pools: list[PoolByCategory] current_pools: "list[PoolByCategory]"
this: Entity this: "Entity"
storage: list[Entity] storage: "Annotated[list[Entity], operator.add]"
class Entity(TypedDict): class Entity(TypedDict):
@@ -164,6 +165,7 @@ class CounterUp:
return { return {
"counter": { "counter": {
"c_80": 0, "c_80": 0,
"c_10": 0,
}, },
} }
@@ -209,7 +211,7 @@ def roll(state: State):
this = Entity(category=t["category"], name=random.choice(t["pool"])) this = Entity(category=t["category"], name=random.choice(t["pool"]))
return { return {
"this": this, "this": this,
"storage": [*state.get("storage", []), this], "storage": [this],
} }
@@ -312,10 +314,77 @@ context: Context = {
}, },
"type": "banner", "type": "banner",
} }
if __name__ == "__main__":
from dotenv import load_dotenv
load_dotenv()
def build_input(
rolling: int, rolled_previously: int = 0, until_5: int = 10, until_6: int = 80
) -> Input:
if not 0 < until_5 <= 10:
print(f"{until_5 = } invalid, idc")
if not 0 < until_6 <= 80:
print(f"{until_6 = } invalid, running anyways")
return {
"countdown": rolling,
"simple_counter": rolled_previously,
"counter": {
"c_10": 10 - until_5,
"c_80": 80 - until_6,
},
}
if TYPE_CHECKING:
from langgraph.graph.state import CompiledStateGraph
def running_lite(
app: "CompiledStateGraph[State, Context, Input, State]",
input: Input,
config: RunnableConfig,
context: Context | None = context,
*,
display: bool = False,
) -> dict[str, Any] | Any:
r = app.invoke(
input,
config,
context=context,
stream_mode="updates",
)
s = cast(State, app.get_state(config=config).values)["storage"]
if display:
print(
"\n".join(
f"{ {'1': ' ', '10': '. ', '80': '! ', '240': '!!'}[i['category']] }: {i['name']}"
for i in s
)
)
return r, s
def how_good_is_my_luck(
repeat: int,
app: "CompiledStateGraph[State, Context, Input, State]",
input: Input,
context: Context | None = context,
):
r: list[list[Entity]] = [
app.invoke(
input,
RunnableConfig(configurable={"thread_id": f"lda {i + 1}"}),
context=context,
)["storage"]
for i in range(repeat)
]
pprint([dict(sorted(Counter(i["category"] for i in x).items())) for x in r])
return r
if __name__ == "__main__":
# from dotenv import load_dotenv
# load_dotenv()
mm = app.get_graph().draw_mermaid() mm = app.get_graph().draw_mermaid()
folder = Path(__file__).parent folder = Path(__file__).parent
(folder / "graph.md").write_text(f"""# graph (folder / "graph.md").write_text(f"""# graph
@@ -327,33 +396,21 @@ if __name__ == "__main__":
(folder / "graph.mermaid").write_text(mm) (folder / "graph.mermaid").write_text(mm)
config: RunnableConfig = {"configurable": {"thread_id": "random ahh"}} config: RunnableConfig = {"configurable": {"thread_id": "random ahh"}}
print("running:")
# random.seed("6741") # tung tung tung # random.seed("6741") # tung tung tung
pprint(
app.invoke( r, _ = running_lite(
{ app, build_input(39, until_5=1, until_6=51), config, display=True
"countdown": 39, # hopefully after all of the stuff and gifts
"simple_counter": 0,
"counter": {
"c_10": 10 - 1,
"c_80": 80 - 51,
},
},
config,
context=context,
stream_mode="updates",
),
) )
print() how_good_is_my_luck(10, app, build_input(120)) # BULLSHIT
print("app state after run") # [
v = app.get_state(config) # {"1": 97, "10": 21, "240": 1, "80": 1},
pprint(v._asdict()) # {"1": 102, "10": 15, "240": 2, "80": 1},
# print() # {"1": 96, "10": 20, "240": 2, "80": 2},
# print("value") # {"1": 101, "10": 15, "240": 4},
# pprint(v.values) # {"1": 98, "10": 19, "240": 3},
# h = app.get_state_history(config) # {"1": 101, "10": 15, "240": 3, "80": 1},
# print() # {"1": 99, "10": 19, "240": 2},
# print("hist") # {"1": 103, "10": 13, "240": 3, "80": 1},
# for i, hi in enumerate(h): # {"1": 98, "10": 19, "240": 1, "80": 2},
# print(f"hist step {i}") # {"1": 101, "10": 14, "240": 3, "80": 2},
# pprint(hi._asdict()) # ]