Langchain辅导Agent状态留存方案咨询:确保LLM按预期响应的实现方法
I’ve built similar coding tutor agents with LangChain, and your issue is a classic case of LLM statelessness—without explicit state cues, even the best system prompts can’t stop the model from misinterpreting user inputs. Let’s break down concrete solutions to lock in your workflow, with options both with and without LangGraph.
Core Issue Recap
The problem isn’t your system prompt—it’s that LLMs don’t retain context between interactions unless you explicitly pass it. Your agent needs a clear way to track its current task (e.g., "waiting for user to retry question 3" vs. "selecting a new topic") and inject that state directly into every prompt the LLM sees.
Solution 1: Explicit State Tracking (No LangGraph Dependency)
If you’re wary of adding LangGraph, this approach uses LangChain’s core memory system to enforce state without extra dependencies.
Step 1: Define a Structured State Object
Create a simple state class to track your agent’s current mode and progress:
from langchain.memory import BaseMemory from typing import Dict, List class TutorStateMemory(BaseMemory): def __init__(self): # Initialize core state variables self.state = { "stage": "SELECTING_TOPIC", # Options: SELECTING_TOPIC, GENERATING_QUESTIONS, ASKING_QUESTION, PROVIDING_FEEDBACK, WAITING_RETRY "current_topic_id": None, "current_question_id": None, "completed_questions": 0, "total_questions_per_topic": 5 } @property def memory_variables(self) -> List[str]: return ["agent_state"] def load_memory_variables(self, inputs: Dict) -> Dict: # Inject state into every prompt return {"agent_state": f"CURRENT AGENT STATE:\n{'- ' + '\n- '.join([f'{k}: {v}' for k, v in self.state.items()])}"} def save_context(self, inputs: Dict, outputs: Dict) -> None: # Update state based on LLM output or tool actions # Example: If LLM confirms a correct answer, increment completed_questions if "correct" in outputs["text"].lower() and self.state["stage"] == "PROVIDING_FEEDBACK": self.state["completed_questions"] += 1 self.state["stage"] = "ASKING_QUESTION" if self.state["completed_questions"] < 5 else "SUGGESTING_NEXT_TOPIC" def clear(self) -> None: self.state = { "stage": "SELECTING_TOPIC", "current_topic_id": None, "current_question_id": None, "completed_questions": 0, "total_questions_per_topic": 5 }
Step 2: Integrate State Into Your Chain
Add this custom memory to your LangChain setup so the state is included in every prompt sent to the LLM:
from langchain.chat_models import ChatOpenAI from langchain.chains import LLMChain from langchain.prompts import PromptTemplate # Update your system prompt to prioritize state system_prompt = """ **CRITICAL: ALWAYS FOLLOW THE CURRENT AGENT STATE FIRST.** You are a Python programming tutor focused on testing user understanding of topics with 5 questions per topic. - If the state says ASKING_QUESTION: Only evaluate the user's answer to the current question, give feedback, and either ask for a retry or move to the next question in the same topic. - If the state says WAITING_RETRY: Only prompt the user to retry the current question. - NEVER treat a user's answer to a question as a request for a new topic. {agent_state} {user_input} """ prompt = PromptTemplate( input_variables=["agent_state", "user_input"], template=system_prompt ) llm = ChatOpenAI(temperature=0) memory = TutorStateMemory() chain = LLMChain(llm=llm, prompt=prompt, memory=memory)
Step 3: Sync State With Your SQLite Database
Every time you perform a database action (e.g., inserting a new topic, updating a question’s attempt count), update the state memory:
- After inserting a topic into
TOPICS, setmemory.state["current_topic_id"] = new_topic_id - After marking a question as correct, increment
memory.state["completed_questions"] - When
completed_questionshits 5, setmemory.state["stage"] = "SUGGESTING_NEXT_TOPIC"and resetcompleted_questionsto 0
Solution 2: LangGraph for Enforced Workflow (Recommended)
If you’re open to trying LangGraph, it’s built exactly for this scenario—it turns your workflow into a state machine that the LLM can’t deviate from. Dependency conflicts are minimal (it’s a lightweight LangChain extension), and it eliminates prompt-based drift entirely.
Step 1: Define Your Agent State
Use a typed dict to structure all relevant state data:
from langgraph.graph import StateGraph, END from typing import TypedDict, List class TutorState(TypedDict): user_input: str current_topic_id: int current_question: str question_history: List[Dict] # Stores QID, question text, attempts, etc. completed_questions: int answer_incorrect: bool
Step 2: Build Workflow Nodes
Each node represents a step in your tutor’s workflow:
def select_topic(state: TutorState) -> TutorState: # Logic to let user select a topic or pick one automatically # Insert new topic into TOPICS table if needed state["current_topic_id"] = 1 # Replace with actual TopicID from DB return state def generate_questions(state: TutorState) -> TutorState: # Generate 5 questions for the current topic and insert into QUESTIONS # Populate question_history with the new QIDs and questions state["question_history"] = [{"qid": 1, "text": "What is a Python function?"}, ...] return state def ask_question(state: TutorState) -> TutorState: # Get the next unasked question from question_history next_question = [q for q in state["question_history"] if q.get("asked") is False][0] state["current_question"] = next_question["text"] # Mark question as asked in DB return state def evaluate_response(state: TutorState) -> TutorState: # Check user's answer against expected correct answer # Update QUESTIONS table with Response, Feedback, Attempts state["answer_incorrect"] = "wrong" in state["user_input"].lower() # Replace with actual logic return state def provide_feedback(state: TutorState) -> TutorState: # Generate feedback for the user # Increment completed_questions if answer was correct if not state["answer_incorrect"]: state["completed_questions"] += 1 return state def suggest_next_topic(state: TutorState) -> TutorState: # Suggest a new topic to the user # Update TOPICS table's Understood field to 1 for the current topic state["completed_questions"] = 0 return state
Step 3: Wire Up the State Machine
Define edges and conditional logic to enforce your workflow:
# Initialize the graph graph = StateGraph(TutorState) # Add nodes graph.add_node("select_topic", select_topic) graph.add_node("generate_questions", generate_questions) graph.add_node("ask_question", ask_question) graph.add_node("evaluate_response", evaluate_response) graph.add_node("provide_feedback", provide_feedback) graph.add_node("suggest_next_topic", suggest_next_topic) # Set entry point graph.set_entry_point("select_topic") # Define linear edges graph.add_edge("select_topic", "generate_questions") graph.add_edge("generate_questions", "ask_question") graph.add_edge("ask_question", "evaluate_response") # Conditional edge: Retry or proceed after evaluation def should_retry(state: TutorState) -> str: return "ask_question" if state["answer_incorrect"] else "provide_feedback" graph.add_conditional_edges("evaluate_response", should_retry) # Conditional edge: Move to new topic or next question def should_switch_topic(state: TutorState) -> str: return "suggest_next_topic" if state["completed_questions"] >= 5 else "ask_question" graph.add_conditional_edges("provide_feedback", should_switch_topic) # Loop back to topic selection after suggesting next topic graph.add_edge("suggest_next_topic", "select_topic") # Compile the graph app = graph.compile()
With LangGraph, the agent will never drift from the workflow—each step is enforced by the state machine, so the LLM can’t misinterpret a user’s answer as a new topic request.
Quick System Prompt Tweaks to Reinforce State
Even with state tracking, a trimmed, focused prompt helps:
- Move state-related rules to the very top with bold formatting
- Remove redundant database schema details (keep those in your tool descriptions instead)
- Simplify workflow instructions to only what the LLM needs to know for the current state
Example trimmed prompt:
**RULES FIRST: ALWAYS CHECK THE CURRENT AGENT STATE BEFORE RESPONDING.** 1. If in ASKING_QUESTION stage: Evaluate the user's answer to the current question, give clear feedback, and either ask for a retry or move to the next question in the same topic. 2. If in WAITING_RETRY stage: Only ask the user to retry the current question. 3. NEVER assume a user's answer is a new topic request—stick to the current topic until all 5 questions are completed. {agent_state} User input: {user_input}
内容的提问来源于stack exchange,提问作者RF2

