Spaces:
Sleeping
Sleeping
File size: 4,831 Bytes
cc42847 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 |
from langgraph.graph import StateGraph, END, START
from langchain_core.rate_limiters import InMemoryRateLimiter
from langgraph.prebuilt import ToolNode
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from tools import get_rendered_html, download_file, post_request, run_code, add_dependencies
from typing import TypedDict, Annotated, List, Any
from langchain.chat_models import init_chat_model
from langgraph.graph.message import add_messages
import os
from dotenv import load_dotenv
load_dotenv()
EMAIL = os.getenv("EMAIL")
SECRET = os.getenv("SECRET")
RECURSION_LIMIT = 5000
# -------------------------------------------------
# STATE
# -------------------------------------------------
class AgentState(TypedDict):
messages: Annotated[List, add_messages]
TOOLS = [run_code, get_rendered_html, download_file, post_request, add_dependencies]
# -------------------------------------------------
# GEMINI LLM
# -------------------------------------------------
rate_limiter = InMemoryRateLimiter(
requests_per_second=9/60,
check_every_n_seconds=1,
max_bucket_size=9
)
llm = init_chat_model(
model_provider="google_genai",
model="gemini-2.5-flash",
rate_limiter=rate_limiter
).bind_tools(TOOLS)
# -------------------------------------------------
# SYSTEM PROMPT
# -------------------------------------------------
SYSTEM_PROMPT = f"""
You are an autonomous quiz-solving agent.
Your job is to:
1. Load the quiz page from the given URL.
2. Extract ALL instructions, required parameters, submission rules, and the submit endpoint.
3. Solve the task exactly as required.
4. Submit the answer ONLY to the endpoint specified on the current page (never make up URLs).
5. Read the server response and:
- If it contains a new quiz URL → fetch it immediately and continue.
- If no new URL is present → return "END".
STRICT RULES — FOLLOW EXACTLY:
GENERAL RULES:
- NEVER stop early. Continue solving tasks until no new URL is provided.
- NEVER hallucinate URLs, endpoints, fields, values, or JSON structure.
- NEVER shorten or modify URLs. Always submit the full URL.
- NEVER re-submit unless the server explicitly allows or it's within the 3-minute limit.
- ALWAYS inspect the server response before deciding what to do next.
- ALWAYS use the tools provided to fetch, scrape, download, render HTML, or send requests.
TIME LIMIT RULES:
- Each task has a hard 3-minute limit.
- The server response includes a "delay" field indicating elapsed time.
- If your answer is wrong retry again.
STOPPING CONDITION:
- Only return "END" when a server response explicitly contains NO new URL.
- DO NOT return END under any other condition.
ADDITIONAL INFORMATION YOU MUST INCLUDE WHEN REQUIRED:
- Email: {EMAIL}
- Secret: {SECRET}
YOUR JOB:
- Follow pages exactly.
- Extract data reliably.
- Never guess.
- Submit correct answers.
- Continue until no new URL.
- Then respond with: END
"""
prompt = ChatPromptTemplate.from_messages([
("system", SYSTEM_PROMPT),
MessagesPlaceholder(variable_name="messages")
])
llm_with_prompt = prompt | llm
# -------------------------------------------------
# AGENT NODE
# -------------------------------------------------
def agent_node(state: AgentState):
result = llm_with_prompt.invoke({"messages": state["messages"]})
return {"messages": state["messages"] + [result]}
# -------------------------------------------------
# GRAPH
# -------------------------------------------------
def route(state):
last = state["messages"][-1]
# support both objects (with attributes) and plain dicts
tool_calls = None
if hasattr(last, "tool_calls"):
tool_calls = getattr(last, "tool_calls", None)
elif isinstance(last, dict):
tool_calls = last.get("tool_calls")
if tool_calls:
return "tools"
# get content robustly
content = None
if hasattr(last, "content"):
content = getattr(last, "content", None)
elif isinstance(last, dict):
content = last.get("content")
if isinstance(content, str) and content.strip() == "END":
return END
if isinstance(content, list) and content[0].get("text").strip() == "END":
return END
return "agent"
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(TOOLS))
graph.add_edge(START, "agent")
graph.add_edge("tools", "agent")
graph.add_conditional_edges(
"agent",
route
)
app = graph.compile()
# -------------------------------------------------
# TEST
# -------------------------------------------------
def run_agent(url: str) -> str:
app.invoke({
"messages": [{"role": "user", "content": url}]},
config={"recursion_limit": RECURSION_LIMIT},
)
print("Tasks completed succesfully")
|