1、对话式工作流

持续追踪 “用户 - 助手 - 工具” 对话历史的场景

LangGraph 的工作流(Graph)运行时,需通过graph.invoke(initial_state)传入初始状态

State Schema

通常使用TypedDict(类型字典)或pydantic.BaseModel明确状态结构,核心字段是messages(存储对话历史):

from langchain_core.messages import BaseMessage
from typing import List, TypedDict

# 定义State结构:包含messages字段(BaseMessage类型的列表)
class ChatState(TypedDict):
    messages: List[BaseMessage]  # 对话历史,需是BaseMessage子类(如HumanMessage、AIMessage)

对应输入格式

# 格式1:直接传入BaseMessage列表(无需convert_to_messages)
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage

initial_state = {
    "messages": [
        HumanMessage(content="What does Lilian Weng say about reward hacking?"),  # 用户消息
        AIMessage(tool_calls=[{"id": "1", "name": "retrieve", "args": {"query": "..."}}]),  # 助手工具调用
        ToolMessage(content="...", tool_call_id="1")  # 工具返回
    ]
}

# 格式2:用convert_to_messages转换字典为BaseMessage(更简洁)
initial_state = {
    "messages": convert_to_messages([
        {"role": "user", "content": "..."},
        {"role": "assistant", "tool_calls": [...]},
        {"role": "tool", "content": "...", "tool_call_id": "..."}
    ])
}

2. 任务型工作流(非对话式 State)

无需对话历史,仅需追踪 “任务参数、中间结果、执行状态” 的场景

State Schema 

from typing import TypedDict

class DocProcessingState(TypedDict):
    doc_content: str  # 输入的文档内容(初始状态需传入)
    summary: str | None  # 中间结果:摘要(初始为None,后续节点赋值)
    keywords: List[str] | None  # 中间结果:关键词(初始为None,后续节点赋值)

对应输入格式

# 初始输入:仅需传入文档内容(其他字段为None,可省略)
initial_state = {
    "doc_content": "LangGraph is a stateful workflow framework for LLMs. It supports conditional edges and cyclic graphs..."
}

3、混合式工作流(多字段 State)

同时追踪 “对话历史” 和 “其他业务数据” 的场景

State Schema 

from langchain_core.messages import BaseMessage
from typing import List, TypedDict

class CustomerServiceState(TypedDict):
    user_id: str  # 业务字段:用户ID(用于关联用户信息)
    doc_context: str  # 业务字段:当前咨询的文档上下文(如产品手册片段)
    messages: List[BaseMessage]  # 对话字段:用户-客服-工具的消息历史

对应输入格式

initial_state = {
    "user_id": "user_12345",  # 初始业务数据
    "doc_context": "Product X has a 1-year warranty. Repairs are free within warranty.",  # 初始业务数据
    "messages": convert_to_messages([{"role": "user", "content": "Does Product X have a warranty?"}])  # 初始对话消息
}

Logo

中国智能体开发者社区,聚焦智能体与大模型开发,提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动,促进经验交流与协作,助力开发者快速构建创新智能应用。

更多推荐