File size: 5,254 Bytes
8740e76 |
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 |
"""
LangGraph-based agent fallback for complex quiz solving.
Used when structured strategies fail or for novel quiz types.
"""
import os
import logging
import time
from typing import Dict, Any, Optional, List, Annotated
from typing_extensions import TypedDict
logger = logging.getLogger(__name__)
# Try to import LangGraph components (optional dependency)
try:
from langgraph.graph import StateGraph, END, START
from langgraph.prebuilt import ToolNode
from langchain_core.messages import trim_messages, HumanMessage
from langchain.chat_models import init_chat_model
from langgraph.graph.message import add_messages
from langchain_core.rate_limiters import InMemoryRateLimiter
LANGGRAPH_AVAILABLE = True
except ImportError:
LANGGRAPH_AVAILABLE = False
logger.warning("LangGraph not available - agent fallback disabled")
class AgentState(TypedDict):
"""State for LangGraph agent."""
messages: Annotated[List, add_messages]
class AgentFallback:
"""Agent-based fallback solver using LangGraph."""
def __init__(self, email: str, secret: str):
self.email = email
self.secret = secret
self.agent = None
self.tools = []
if LANGGRAPH_AVAILABLE:
self._initialize_agent()
def _initialize_agent(self):
"""Initialize the LangGraph agent."""
try:
# Define tools (simplified - you'd import from your tools module)
# For now, we'll create a minimal agent
# Initialize LLM with rate limiting
rate_limiter = InMemoryRateLimiter(
requests_per_second=4 / 60,
check_every_n_seconds=1,
max_bucket_size=4
)
llm = init_chat_model(
model_provider="google_genai",
model="gemini-2.5-flash",
rate_limiter=rate_limiter
)
# Create simple graph (you'd add your tools here)
graph = StateGraph(AgentState)
graph.add_node("agent", self._agent_node)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", self._route, {END: END})
self.agent = graph.compile()
logger.info("Agent fallback initialized")
except Exception as e:
logger.error(f"Error initializing agent: {e}")
self.agent = None
def _agent_node(self, state: AgentState):
"""Agent node that processes messages."""
# Simplified - would use actual LLM here
return {"messages": state["messages"]}
def _route(self, state: AgentState):
"""Route logic for agent."""
return END
async def solve(self, question: str, page_content: Dict[str, Any],
remaining_time: float) -> Optional[Any]:
"""
Attempt to solve using agent-based approach.
Args:
question: Question text
page_content: Page content
remaining_time: Time remaining in seconds
Returns:
Answer if solved, None otherwise
"""
if not LANGGRAPH_AVAILABLE or not self.agent:
return None
# Only use agent if we have enough time
if remaining_time < 30.0:
logger.debug("Skipping agent fallback - insufficient time")
return None
try:
logger.info("Attempting agent-based solution...")
system_prompt = f"""
You are a quiz-solving agent. Solve this question:
Question: {question}
Page Content: {page_content.get('text', '')[:2000]}
Email: {self.email}
Secret: {self.secret}
Provide a clear, concise answer.
"""
initial_messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": question}
]
# Run agent with timeout
result = self.agent.invoke(
{"messages": initial_messages},
config={"recursion_limit": 100}
)
# Extract answer from result
if result and "messages" in result:
last_message = result["messages"][-1]
if hasattr(last_message, "content"):
return last_message.content
elif isinstance(last_message, dict) and "content" in last_message:
return last_message["content"]
return None
except Exception as e:
logger.error(f"Error in agent fallback: {e}")
return None
def get_agent_fallback(email: str, secret: str) -> Optional[AgentFallback]:
"""Get or create agent fallback instance."""
if not LANGGRAPH_AVAILABLE:
return None
return AgentFallback(email, secret)
|