You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Langchain辅导Agent状态留存方案咨询:确保LLM按预期响应的实现方法

Fixing LLM Topic Drift in Your Python Tutor Agent: State Persistence Workflows

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, set memory.state["current_topic_id"] = new_topic_id
  • After marking a question as correct, increment memory.state["completed_questions"]
  • When completed_questions hits 5, set memory.state["stage"] = "SUGGESTING_NEXT_TOPIC" and reset completed_questions to 0

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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.04.27 09:17:35