LangGraph 用户需求澄清与研究规划详细解读
📚 概述
本文档详细解读 Scoping(需求澄清与研究规划) 在 Deep Research 系统中的关键作用。这是整个研究流程的第一步,也是最重要的一步——如果需求理解错误,后续所有研究都将偏离方向。
核心问题: 用户的初始请求往往模糊、不完整,缺少关键信息。
示例:
- ❌ "研究最好的咖啡店" → 最好指什么?咖啡质量?氛围?价格?
- ❌ "对比 A 和 B" → 对比哪些维度?技术?价格?用户体验?
- ❌ "分析特斯拉" → 分析什么?股票?技术?市场?
Scoping 的目标: 通过智能对话,将模糊请求转化为清晰、结构化的研究简报。
📚 术语表
| 术语名称 | LangGraph 定义和解读 | Python 定义和说明 | 重要程度 |
|---|---|---|---|
| Scoping | 需求澄清与研究规划阶段,通过对话明确用户真实需求并生成研究简报 | N/A (工作流概念) | ⭐⭐⭐⭐⭐ |
| ClarifyWithUser | 结构化输出 Schema,判断是否需要向用户提问 | class ClarifyWithUser(BaseModel) | ⭐⭐⭐⭐⭐ |
| ResearchQuestion | 结构化输出 Schema,用于生成研究简报 | class ResearchQuestion(BaseModel) | ⭐⭐⭐⭐⭐ |
| Command | LangGraph 控制流对象,用于动态路由和状态更新 | Command(goto="node", update={...}) | ⭐⭐⭐⭐⭐ |
| Structured Output | 强制 LLM 输出符合预定义 Schema 的技术 | model.with_structured_output(Schema) | ⭐⭐⭐⭐⭐ |
| get_buffer_string | 将消息历史转化为字符串格式 | from langchain_core.messages import get_buffer_string | ⭐⭐⭐⭐ |
| MessagesState | LangGraph 预定义状态类,管理消息历史 | class MessagesState(TypedDict) | ⭐⭐⭐⭐ |
| Research Brief | 详细的研究规划文档,指导后续研究方向 | 字符串,包含研究主题、范围、标准等 | ⭐⭐⭐⭐⭐ |
| LLM-as-judge | 使用 LLM 评估其他 LLM 输出质量的技术 | 评估函数,调用 LLM 并返回评分 | ⭐⭐⭐⭐ |
🎯 核心概念
什么是 Scoping?
Scoping 是 Deep Research 系统的第一阶段,包含两个子步骤:
- User Clarification(用户澄清) - 判断是否需要向用户提问以澄清需求
- Brief Generation(简报生成) - 将对话历史转化为详细的研究简报
完整流程:
用户输入: "研究最好的咖啡店"
↓
┌─────────────────────────────────────┐
│ Step 1: 用户澄清 │
│ LLM 分析: "最好"的标准不明确 │
│ 生成问题: "您关注咖啡质量、氛围还是价格?"│
└─────────────────────────────────────┘
↓
用户回复: "咖啡质量"
↓
┌─────────────────────────────────────┐
│ Step 2: 简报生成 │
│ 将对话转化为研究简报: │
│ "研究旧金山地区以咖啡质量著称的咖啡店,│
│ 重点关注专业评分、豆源、烘焙技术..." │
└─────────────────────────────────────┘为什么需要 Scoping?
问题 1:用户请求缺少关键信息
常见缺失信息:
- 范围和边界 - 应该包括/排除什么?
- 受众和目的 - 这个研究是给谁看的,为什么需要?
- 具体要求 - 有特定的来源、时间范围、约束吗?
- 术语澄清 - 领域术语或缩写是什么意思?
示例对比:
| 模糊请求 | 缺失信息 | 澄清问题 |
|---|---|---|
| "研究最好的咖啡店" | 地点、标准 | "哪个城市?关注什么标准?" |
| "对比 OpenAI 和 Anthropic" | 对比维度 | "对比技术能力、价格还是使用体验?" |
| "分析 TSLA" | 分析角度 | "分析股票表现、技术创新还是市场竞争?" |
问题 2:避免错误假设
如果不澄清需求,LLM 可能基于自身偏好做出假设:
# ❌ 错误假设示例
User: "研究最好的咖啡店"
LLM (假设): "用户可能关心价格和便利性"
→ 研究结果: 连锁咖啡店(Starbucks, Peet's)
# ✅ 正确流程
User: "研究最好的咖啡店"
LLM: "您关注咖啡质量、氛围还是价格?"
User: "咖啡质量"
→ 研究结果: 精品咖啡店(Blue Bottle, Sightglass)🔧 技术实现详解
1. 状态定义
from typing_extensions import Optional, Annotated, Sequence
from langchain_core.messages import BaseMessage
from langgraph.graph import MessagesState
from langgraph.graph.message import add_messages
from pydantic import BaseModel, Field
# ===== 状态定义 =====
class AgentState(MessagesState):
"""
主状态类,继承自 MessagesState
MessagesState 提供:
- messages: 消息历史(自动使用 add_messages reducer)
新增字段:
- research_brief: 生成的研究简报
- supervisor_messages: 与 supervisor 的通信
"""
research_brief: Optional[str]
supervisor_messages: Annotated[Sequence[BaseMessage], add_messages]关键点:
MessagesState- LangGraph 预定义基类,自动管理消息历史add_messages- Reducer,自动处理消息追加/更新/删除Optional[str]- 研究简报初始为 None,生成后才有值
2. Structured Output Schema
为什么使用 Structured Output?
传统方式(字符串输出)的问题:
# ❌ 不可靠的字符串解析
response = llm.invoke("判断是否需要澄清,回答 yes/no")
# 可能输出: "Yes, I think..." "YES" "yes." "需要澄清"
# → 难以可靠解析Structured Output 的优势:
# ✅ 可靠的结构化输出
class ClarifyWithUser(BaseModel):
need_clarification: bool # 强制 bool 类型
question: str
verification: str
response = llm.with_structured_output(ClarifyWithUser).invoke(...)
# 保证返回: response.need_clarification 是 True/False完整 Schema 定义:
class ClarifyWithUser(BaseModel):
"""用户澄清决策和问题 Schema"""
need_clarification: bool = Field(
description="是否需要向用户提问以澄清需求",
)
question: str = Field(
description="如果需要澄清,向用户提出的问题(使用 Markdown 格式)",
)
verification: str = Field(
description="如果不需要澄清,确认开始研究的消息",
)工作流程:
if response.need_clarification:
# 返回问题给用户
return {"messages": [AIMessage(content=response.question)]}
else:
# 确认开始研究
return {"messages": [AIMessage(content=response.verification)]}3. Prompt 设计
核心 Prompt(clarify_with_user_instructions):
clarify_with_user_instructions = """
这是用户与你的对话历史:
<Messages>
{messages}
</Messages>
今天的日期是 {date}。
评估是否需要提问澄清,或者用户已经提供了足够信息。
重要提示: 如果你已经在对话历史中问过澄清问题,几乎总是不需要再问。
只在绝对必要时才提出新问题。
如果有缩写、简称或未知术语,请用户澄清。
如果需要提问,遵循以下准则:
- 简洁的同时收集所有必要信息
- 确保收集执行研究任务所需的全部信息
- 适当时使用项目符号或编号列表以提高清晰度
- 确保使用 Markdown 格式,可被 Markdown 渲染器正确显示
- 不要询问不必要的信息或用户已经提供的信息
以有效 JSON 格式响应,包含以下键:
"need_clarification": boolean,
"question": "<向用户提问以澄清研究范围>",
"verification": "<确认我们将开始研究的消息>"
如果需要澄清问题,返回:
"need_clarification": true,
"question": "<你的澄清问题>",
"verification": ""
如果不需要澄清问题,返回:
"need_clarification": false,
"question": "",
"verification": "<确认消息:你将基于提供的信息开始研究>"
对于不需要澄清时的确认消息:
- 确认你有足够信息继续
- 简要总结你从请求中理解的关键内容
- 确认你现在将开始研究过程
- 保持简洁和专业
"""Prompt 设计原则:
防止重复提问
重要提示: 如果你已经在对话历史中问过澄清问题,几乎总是不需要再问。明确输出格式
- 使用 JSON Schema
- 提供正负示例(需要澄清 vs 不需要澄清)
提供上下文
- 包含完整对话历史
- 提供当前日期(对时间敏感的研究很重要)
质量要求
- 使用 Markdown 格式
- 简洁但完整
- 项目符号提高可读性
4. 节点实现:clarify_with_user
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AIMessage, get_buffer_string
from langgraph.types import Command
# 初始化模型
model = init_chat_model(model="openai:gpt-4.1", temperature=0.0)
def clarify_with_user(state: AgentState) -> Command:
"""
判断是否需要向用户提问以澄清需求
使用 Structured Output 确保可靠的决策
Returns:
Command 对象,指向下一个节点:
- 如果需要澄清 → goto=END (返回问题给用户)
- 如果不需要澄清 → goto="write_research_brief"
"""
# 设置 Structured Output 模型
structured_output_model = model.with_structured_output(ClarifyWithUser)
# 调用模型进行判断
response = structured_output_model.invoke([
HumanMessage(content=clarify_with_user_instructions.format(
messages=get_buffer_string(messages=state["messages"]),
date=get_today_str()
))
])
# 基于判断结果路由
if response.need_clarification:
# 需要澄清:结束当前执行,返回问题给用户
return Command(
goto=END,
update={"messages": [AIMessage(content=response.question)]}
)
else:
# 不需要澄清:继续到简报生成
return Command(
goto="write_research_brief",
update={"messages": [AIMessage(content=response.verification)]}
)关键技术:Command 对象
Command(
goto="next_node", # 下一个要执行的节点(或 END)
update={...} # 要更新的状态字段
)Command 的优势:
- 🎯 灵活路由 - 节点可以动态决定下一步
- 🔄 状态更新 - 同时更新状态和跳转
- 📊 清晰语义 - 代码意图一目了然
对比传统条件边:
# ❌ 传统方式:需要单独的路由函数
def should_continue(state):
if needs_clarification(state):
return "end"
return "write_brief"
builder.add_conditional_edges("clarify", should_continue, {...})
# ✅ 使用 Command:决策和路由在一个节点中
def clarify_with_user(state):
if response.need_clarification:
return Command(goto=END, update={...})
return Command(goto="write_research_brief", update={...})5. 节点实现:write_research_brief
class ResearchQuestion(BaseModel):
"""研究简报 Schema"""
research_brief: str = Field(
description="详细的研究简报,将指导后续研究",
)
def write_research_brief(state: AgentState):
"""
将对话历史转化为详细的研究简报
使用 Structured Output 确保简报格式符合要求
"""
# 设置 Structured Output 模型
structured_output_model = model.with_structured_output(ResearchQuestion)
# 生成研究简报
response = structured_output_model.invoke([
HumanMessage(content=transform_messages_into_research_topic_prompt.format(
messages=get_buffer_string(state.get("messages", [])),
date=get_today_str()
))
])
# 更新状态
return {
"research_brief": response.research_brief,
"supervisor_messages": [HumanMessage(content=f"{response.research_brief}.")]
}研究简报的质量要求:
好的研究简报应该包含:
- 明确的研究主题 - 要研究什么?
- 具体的范围 - 包括/排除什么?
- 评估标准 - 如何判断"好"?
- 优先信息源 - 首选什么类型的来源?
- 时间要求 - 截止到什么时候的信息?
示例对比:
| 质量 | 研究简报内容 |
|---|---|
| ❌ 差 | "研究旧金山的咖啡店" |
| ⚠️ 一般 | "研究旧金山的精品咖啡店,关注咖啡质量" |
| ✅ 好 | "研究旧金山以咖啡质量著称的精品咖啡店。重点关注豆源、烘焙技术、专业评分(如 Coffee Review)和用户评价。优先使用咖啡店官网、第三方专业评测(Specialty Coffee Association)和评价聚合网站(Yelp, Google Reviews)。截止到 2025年7月的最新数据。" |
6. 图构建
from langgraph.graph import StateGraph, START, END
# 构建 Scoping 工作流
scope_builder = StateGraph(AgentState, input_schema=AgentInputState)
# 添加节点
scope_builder.add_node("clarify_with_user", clarify_with_user)
scope_builder.add_node("write_research_brief", write_research_brief)
# 添加边
scope_builder.add_edge(START, "clarify_with_user")
scope_builder.add_edge("write_research_brief", END)
# 编译
scope_research = scope_builder.compile()
# 🎨 可视化图结构
from IPython.display import Image, display
display(Image(scope_research.get_graph().draw_mermaid_png()))图的执行流程:
START
↓
clarify_with_user
↓ (decision)
├─ need_clarification? → END (返回问题)
└─ sufficient info? → write_research_brief → END🎭 实战案例:咖啡店研究
示例 1:需要澄清
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver
# 编译带 checkpointer 的图(用于测试)
checkpointer = MemorySaver()
scope = scope_builder.compile(checkpointer=checkpointer)
# 第一次调用:模糊请求
thread = {"configurable": {"thread_id": "1"}}
result = scope.invoke({
"messages": [HumanMessage(content="我想研究旧金山最好的咖啡店。")]
}, config=thread)
print(result['messages'][-1].content)输出:
您能具体说明什么标准对您来说最重要吗?例如:
- 咖啡质量
- 氛围和环境
- Wi-Fi 可用性
- 食物选项
- 其他因素?第二次调用:提供澄清
result = scope.invoke({
"messages": [HumanMessage(content="咖啡质量是最重要的标准。")]
}, config=thread)
print(result['messages'][-1].content)输出:
感谢澄清,咖啡质量是您的首要标准。我有足够信息继续,
现在将开始研究旧金山基于咖啡质量的最佳咖啡店。生成的研究简报:
print(result["research_brief"])输出:
我想识别和评估旧金山被认为是基于咖啡质量的最佳咖啡店。
我的研究应专注于分析和比较旧金山地区的咖啡店,以咖啡质量
作为主要标准。我对评估咖啡质量的方法持开放态度(例如,
专家评测、客户评分、精品咖啡认证),对氛围、地点、Wi-Fi
或食物选项没有约束,除非它们直接影响感知的咖啡质量。
请优先考虑主要来源,如咖啡店官网、信誉良好的第三方咖啡
评测组织(如 Coffee Review 或 Specialty Coffee Association)
以及知名评价聚合网站(如 Google 或 Yelp),在这些网站可以
找到关于咖啡质量的直接客户反馈。
研究应产生一个有充分支持的列表或排名,强调截止到 2025年7月
的最新可用数据所显示的咖啡质量。示例 2:立即满足(无需澄清)
thread2 = {"configurable": {"thread_id": "2"}}
result = scope.invoke({
"messages": [HumanMessage(content="""
我想研究旧金山基于咖啡质量的最佳咖啡店。
请关注专业评分(Coffee Review)、豆源和烘焙技术。
优先使用官网和专业评测网站。
""")]
}, config=thread2)
print(result['messages'][-1].content)输出:
感谢您提供具体的研究参数。您已经明确指定了地点(旧金山)、
主要标准(咖啡质量)以及评估方法(专业评分、豆源、烘焙技术)。
我现在将开始基于提供的标准进行研究。直接生成简报,无需额外澄清。
🎓 核心知识点总结
LangGraph 特有概念
1. Structured Output 的强大之处
传统字符串输出的问题:
# ❌ 不可靠
response = llm.invoke("判断是否需要澄清,回答 true/false")
# 可能返回: "True" "true" "TRUE" "yes" "I think true"
if "true" in response.lower(): # 脆弱的解析
...Structured Output 的优势:
# ✅ 可靠
class Decision(BaseModel):
need_clarification: bool
response = llm.with_structured_output(Decision).invoke(...)
if response.need_clarification: # 类型安全
...重要提示: Structured Output 使用 LLM 的 function calling 能力,确保输出符合 Schema。
2. Command 对象的灵活性
Command vs 条件边对比:
| 特性 | 条件边 | Command |
|---|---|---|
| 路由决策 | 单独的路由函数 | 节点内部决策 |
| 状态更新 | 分离的逻辑 | 一起处理 |
| 代码组织 | 分散在多处 | 集中在节点中 |
| 灵活性 | 较低 | 高 |
使用建议:
- ✅ 简单静态路由 - 使用条件边
- ✅ 复杂动态路由 - 使用 Command
3. MessagesState 的便利性
手动管理 vs MessagesState:
# ❌ 手动管理消息
class MyState(TypedDict):
messages: list
def node(state):
new_messages = state["messages"] + [new_msg] # 手动追加
return {"messages": new_messages}
# ✅ 使用 MessagesState
class MyState(MessagesState):
pass # 自动包含 messages + add_messages reducer
def node(state):
return {"messages": [new_msg]} # 自动追加add_messages Reducer 功能:
- 🔄 追加消息 - 默认行为
- 🔄 覆盖消息 - 如果提供相同 ID
- 🔄 删除消息 - 使用 RemoveMessage
Python 特有知识点
1. Pydantic BaseModel
from pydantic import BaseModel, Field
class ClarifyWithUser(BaseModel):
need_clarification: bool = Field(
description="是否需要澄清" # 这个描述会传递给 LLM
)Field 的作用:
- 📝 description - 告诉 LLM 这个字段的用途
- 📝 examples - 提供示例值
- 📝 constraints - 添加验证规则
2. get_buffer_string 工具函数
from langchain_core.messages import get_buffer_string
messages = [
HumanMessage("你好"),
AIMessage("你好!有什么可以帮您?"),
HumanMessage("告诉我关于咖啡的信息")
]
buffer = get_buffer_string(messages)
# 输出:
# Human: 你好
# AI: 你好!有什么可以帮您?
# Human: 告诉我关于咖啡的信息用途: 将消息列表格式化为 LLM 可读的字符串。
💡 最佳实践
1. 何时需要澄清?
需要澄清的场景:
- ✅ 请求包含模糊术语("最好"、"顶级"、"优秀")
- ✅ 缺少关键参数(地点、时间、标准)
- ✅ 使用缩写或专业术语
- ✅ 对比任务但未指定维度
不需要澄清的场景:
- ❌ 请求非常具体("2024年Q4 TSLA 股票价格走势")
- ❌ 已经在对话中澄清过
- ❌ 常识性请求("Python 如何定义函数")
2. Prompt 设计技巧
技巧 1:提供正负示例
prompt = """
如果需要澄清,返回:
{
"need_clarification": true,
"question": "您关注的是咖啡质量、氛围还是价格?",
"verification": ""
}
如果不需要澄清,返回:
{
"need_clarification": false,
"question": "",
"verification": "我将基于您提供的标准开始研究。"
}
"""技巧 2:防止重复提问
# 在 Prompt 中明确指出
"如果你已经在对话历史中问过澄清问题,几乎总是不需要再问。"技巧 3:使用 XML 标签组织上下文
prompt = """
<Messages>
{messages}
</Messages>
<Instructions>
...
</Instructions>
<Output Format>
...
</Output Format>
"""3. 研究简报的质量标准
评估标准:
| 标准 | 说明 | 示例 |
|---|---|---|
| 完整性 | 包含所有用户提到的要点 | 如果用户提到"咖啡质量",简报必须包含 |
| 无假设 | 不添加用户未提及的偏好 | 用户未提到"价格",不应假设价格重要 |
| 具体性 | 明确的评估标准和来源 | "专业评分(Coffee Review)"而非"好的评价" |
| 可执行 | 提供足够细节供研究 Agent 执行 | 明确时间范围、地理范围、数据源 |
4. 错误处理
常见问题:
问题 1:LLM 拒绝使用 Structured Output
# ❌ LLM 返回自然语言而非 JSON
try:
response = llm.with_structured_output(ClarifyWithUser).invoke(...)
except Exception as e:
# 降级策略:使用正则表达式解析
...解决方案:
- 使用更强的模型(GPT-4 而非 GPT-3.5)
- 在 Prompt 中强调输出格式要求
- 提供更多示例
问题 2:对话历史太长
# 如果对话超过 100 轮
if len(state["messages"]) > 100:
# 只保留最近 50 轮
recent_messages = state["messages"][-50:]
buffer = get_buffer_string(recent_messages)🔍 评估方法
设计评估器
评估目标:
- 研究简报是否包含所有用户提到的标准?
- 研究简报是否避免了用户未提及的假设?
评估器 1:Success Criteria(成功标准)
from pydantic import BaseModel, Field
class Criteria(BaseModel):
criteria_text: str = Field(description="具体的成功标准")
reasoning: str = Field(description="评估推理")
is_captured: bool = Field(description="是否在简报中体现")
def evaluate_success_criteria(outputs: dict, reference_outputs: dict):
"""
评估研究简报是否包含所有必需标准
"""
research_brief = outputs["research_brief"]
success_criteria = reference_outputs["criteria"]
model = ChatOpenAI(model="gpt-4.1", temperature=0)
structured_model = model.with_structured_output(Criteria)
# 批量评估每个标准
individual_evaluations = []
for criterion in success_criteria:
response = structured_model.invoke([
HumanMessage(content=f"""
研究简报: {research_brief}
评估标准: {criterion}
判断这个标准是否在研究简报中得到充分体现。
""")
])
individual_evaluations.append(response)
# 计算总分
captured_count = sum(1 for eval in individual_evaluations if eval.is_captured)
total_count = len(individual_evaluations)
return {
"key": "success_criteria_score",
"score": captured_count / total_count,
"individual_evaluations": individual_evaluations
}评估器 2:No Assumptions(无假设)
class NoAssumptions(BaseModel):
no_assumptions: bool = Field(
description="研究简报是否避免了未经用户明确的假设"
)
reasoning: str = Field(description="评估推理")
def evaluate_no_assumptions(outputs: dict, reference_outputs: dict):
"""
评估研究简报是否避免了错误假设
"""
research_brief = outputs["research_brief"]
success_criteria = reference_outputs["criteria"]
model = ChatOpenAI(model="gpt-4.1", temperature=0)
structured_model = model.with_structured_output(NoAssumptions)
response = structured_model.invoke([
HumanMessage(content=f"""
研究简报: {research_brief}
用户明确的标准: {success_criteria}
评估研究简报是否包含了用户未明确提及的假设或偏好。
""")
])
return {
"key": "no_assumptions_score",
"score": response.no_assumptions,
"reasoning": response.reasoning
}运行评估:
from langsmith import Client
client = Client()
# 创建数据集
dataset_name = "scoping_quality"
dataset = client.create_dataset(dataset_name, description="Scoping 质量评估")
# 添加测试用例
client.create_examples(
dataset_id=dataset.id,
examples=[
{
"inputs": {"messages": conversation_1},
"outputs": {"criteria": ["当前年龄 25", "目标退休年龄 45", "高风险承受", ...]}
}
]
)
# 运行评估
client.evaluate(
lambda inputs: scope.invoke(inputs),
data=dataset_name,
evaluators=[evaluate_success_criteria, evaluate_no_assumptions],
experiment_prefix="Scoping Quality"
)📖 扩展阅读
🎉 总结
Scoping 是 Deep Research 的关键第一步:
- 智能澄清 - 使用 Structured Output 可靠判断是否需要提问
- Command 控制流 - 灵活路由到不同节点
- 研究简报生成 - 将对话转化为结构化规划
- 评估驱动 - 使用 LLM-as-judge 持续改进质量
核心技巧:
ClarifyWithUserSchema 确保可靠决策Command对象实现动态路由get_buffer_string格式化消息历史- 评估器验证简报质量
通过 Scoping,我们避免了基于模糊需求进行研究的陷阱,为后续的深度研究打下坚实基础!
🎯 下一步: 让我们继续到 9.2 研究智能体基础 — 学习如何构建能自主搜索、反思、决策的研究 Agent!