- 工作流具有预定义的代码路径，设计为按特定顺序运行。
- 智能体是动态的，可自行定义其处理流程和工具使用方式。
设置
要构建工作流或智能体，您可以使用任何支持结构化输出和工具调用的聊天模型。以下示例使用 Anthropic：- 安装依赖项：
Copy
pip install langchain_core langchain-anthropic langgraph
- 初始化 LLM：
Copy
import os
import getpass
from langchain_anthropic import ChatAnthropic
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("ANTHROPIC_API_KEY")
llm = ChatAnthropic(model="claude-3-5-sonnet-latest")
LLM 与增强功能
工作流和智能体系统基于 LLM 及您为其添加的各种增强功能。工具调用、结构化输出和短期记忆是几种可根据您的需求定制 LLM 的选项。 LLM 增强功能
Copy
# 结构化输出的模式
from pydantic import BaseModel, Field
class SearchQuery(BaseModel):
search_query: str = Field(None, description="优化用于网络搜索的查询。")
justification: str = Field(
None, description="此查询与用户请求相关的原因。"
)
# 使用模式增强 LLM 以支持结构化输出
structured_llm = llm.with_structured_output(SearchQuery)
# 调用增强后的 LLM
output = structured_llm.invoke("钙 CT 评分如何与高胆固醇相关？")
# 定义一个工具
def multiply(a: int, b: int) -> int:
return a * b
# 使用工具增强 LLM
llm_with_tools = llm.bind_tools([multiply])
# 调用 LLM，输入触发工具调用
msg = llm_with_tools.invoke("2 乘以 3 是多少？")
# 获取工具调用
msg.tool_calls
提示链式调用
提示链式调用是指每个 LLM 调用处理前一个调用的输出。它通常用于执行可分解为更小、可验证步骤的明确定义任务。一些示例包括：- 将文档翻译成不同语言
- 验证生成内容的一致性
Copy
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from IPython.display import Image, display
# 图状态
class State(TypedDict):
topic: str
joke: str
improved_joke: str
final_joke: str
# 节点
def generate_joke(state: State):
"""第一次 LLM 调用，生成初始笑话"""
msg = llm.invoke(f"写一个关于 {state['topic']} 的短笑话")
return {"joke": msg.content}
def check_punchline(state: State):
"""检查笑话是否有笑点的门控函数"""
# 简单检查 - 笑话是否包含 "?" 或 "!"
if "?" in state["joke"] or "!" in state["joke"]:
return "通过"
return "失败"
def improve_joke(state: State):
"""第二次 LLM 调用，改进笑话"""
msg = llm.invoke(f"通过添加文字游戏让这个笑话更有趣：{state['joke']}")
return {"improved_joke": msg.content}
def polish_joke(state: State):
"""第三次 LLM 调用，进行最终润色"""
msg = llm.invoke(f"给这个笑话添加一个出人意料的转折：{state['improved_joke']}")
return {"final_joke": msg.content}
# 构建工作流
workflow = StateGraph(State)
# 添加节点
workflow.add_node("generate_joke", generate_joke)
workflow.add_node("improve_joke", improve_joke)
workflow.add_node("polish_joke", polish_joke)
# 添加边以连接节点
workflow.add_edge(START, "generate_joke")
workflow.add_conditional_edges(
"generate_joke", check_punchline, {"失败": "improve_joke", "通过": END}
)
workflow.add_edge("improve_joke", "polish_joke")
workflow.add_edge("polish_joke", END)
# 编译
chain = workflow.compile()
# 显示工作流
display(Image(chain.get_graph().draw_mermaid_png()))
# 调用
state = chain.invoke({"topic": "猫"})
print("初始笑话：")
print(state["joke"])
print("\n--- --- ---\n")
if "improved_joke" in state:
print("改进后的笑话：")
print(state["improved_joke"])
print("\n--- --- ---\n")
print("最终笑话：")
print(state["final_joke"])
else:
print("笑话未通过质量检查 - 未检测到笑点！")
并行化
在并行化中，LLM 同时处理一个任务。这可以通过同时运行多个独立的子任务，或多次运行同一任务以检查不同输出来实现。并行化通常用于：- 将子任务拆分并并行运行，以提高速度
- 多次运行任务以检查不同输出，以提高置信度
- 运行一个子任务处理文档中的关键词，同时运行另一个子任务检查格式错误
- 多次运行一个基于不同标准（如引用数量、使用来源数量和来源质量）对文档准确性进行评分的任务
Copy
# 图状态
class State(TypedDict):
topic: str
joke: str
story: str
poem: str
combined_output: str
# 节点
def call_llm_1(state: State):
"""第一次 LLM 调用，生成初始笑话"""
msg = llm.invoke(f"写一个关于 {state['topic']} 的笑话")
return {"joke": msg.content}
def call_llm_2(state: State):
"""第二次 LLM 调用，生成故事"""
msg = llm.invoke(f"写一个关于 {state['topic']} 的故事")
return {"story": msg.content}
def call_llm_3(state: State):
"""第三次 LLM 调用，生成诗歌"""
msg = llm.invoke(f"写一首关于 {state['topic']} 的诗")
return {"poem": msg.content}
def aggregator(state: State):
"""将笑话和故事合并为单一输出"""
combined = f"以下是关于 {state['topic']} 的故事、笑话和诗歌！\n\n"
combined += f"故事：\n{state['story']}\n\n"
combined += f"笑话：\n{state['joke']}\n\n"
combined += f"诗歌：\n{state['poem']}"
return {"combined_output": combined}
# 构建工作流
parallel_builder = StateGraph(State)
# 添加节点
parallel_builder.add_node("call_llm_1", call_llm_1)
parallel_builder.add_node("call_llm_2", call_llm_2)
parallel_builder.add_node("call_llm_3", call_llm_3)
parallel_builder.add_node("aggregator", aggregator)
# 添加边以连接节点
parallel_builder.add_edge(START, "call_llm_1")
parallel_builder.add_edge(START, "call_llm_2")
parallel_builder.add_edge(START, "call_llm_3")
parallel_builder.add_edge("call_llm_1", "aggregator")
parallel_builder.add_edge("call_llm_2", "aggregator")
parallel_builder.add_edge("call_llm_3", "aggregator")
parallel_builder.add_edge("aggregator", END)
parallel_workflow = parallel_builder.compile()
# 显示工作流
display(Image(parallel_workflow.get_graph().draw_mermaid_png()))
# 调用
state = parallel_workflow.invoke({"topic": "猫"})
print(state["combined_output"])
路由
路由工作流处理输入，然后将其引导至特定于上下文的任务。这允许您为复杂任务定义专门的流程。例如，一个用于回答产品相关问题的工作流可能首先处理问题类型，然后将请求路由到定价、退款、退货等特定流程。 routing.png
Copy
from typing_extensions import Literal
from langchain_core.messages import HumanMessage, SystemMessage
# 用作路由逻辑的结构化输出模式
class Route(BaseModel):
step: Literal["poem", "story", "joke"] = Field(
None, description="路由过程中的下一步"
)
# 使用模式增强 LLM 以支持结构化输出
router = llm.with_structured_output(Route)
# 状态
class State(TypedDict):
input: str
decision: str
output: str
# 节点
def llm_call_1(state: State):
"""写一个故事"""
result = llm.invoke(state["input"])
return {"output": result.content}
def llm_call_2(state: State):
"""写一个笑话"""
result = llm.invoke(state["input"])
return {"output": result.content}
def llm_call_3(state: State):
"""写一首诗"""
result = llm.invoke(state["input"])
return {"output": result.content}
def llm_call_router(state: State):
"""将输入路由到适当的节点"""
# 运行带有结构化输出的增强 LLM 作为路由逻辑
decision = router.invoke(
[
SystemMessage(
content="根据用户请求将输入路由到故事、笑话或诗歌。"
),
HumanMessage(content=state["input"]),
]
)
return {"decision": decision.step}
# 条件边函数，用于路由到适当的节点
def route_decision(state: State):
# 返回您希望访问的下一个节点名称
if state["decision"] == "story":
return "llm_call_1"
elif state["decision"] == "joke":
return "llm_call_2"
elif state["decision"] == "poem":
return "llm_call_3"
# 构建工作流
router_builder = StateGraph(State)
# 添加节点
router_builder.add_node("llm_call_1", llm_call_1)
router_builder.add_node("llm_call_2", llm_call_2)
router_builder.add_node("llm_call_3", llm_call_3)
router_builder.add_node("llm_call_router", llm_call_router)
# 添加边以连接节点
router_builder.add_edge(START, "llm_call_router")
router_builder.add_conditional_edges(
"llm_call_router",
route_decision,
{ # route_decision 返回的名称 : 要访问的下一个节点名称
"llm_call_1": "llm_call_1",
"llm_call_2": "llm_call_2",
"llm_call_3": "llm_call_3",
},
)
router_builder.add_edge("llm_call_1", END)
router_builder.add_edge("llm_call_2", END)
router_builder.add_edge("llm_call_3", END)
# 编译工作流
router_workflow = router_builder.compile()
# 显示工作流
display(Image(router_workflow.get_graph().draw_mermaid_png()))
# 调用
state = router_workflow.invoke({"input": "给我写一个关于猫的笑话"})
print(state["output"])
协调器-工作者
在协调器-工作者配置中，协调器：- 将任务分解为子任务
- 将子任务委派给工作者
- 将工作者的输出综合为最终结果
Copy
from typing import Annotated, List
import operator
# 用于规划中结构化输出的模式
class Section(BaseModel):
name: str = Field(
description="本报告部分的名称。",
)
description: str = Field(
description="本部分将涵盖的主要主题和概念的简要概述。",
)
class Sections(BaseModel):
sections: List[Section] = Field(
description="报告的各个部分。",
)
# 使用结构化输出模式增强LLM
planner = llm.with_structured_output(Sections)
在LangGraph中创建工作者
协调器-工作者工作流很常见，LangGraph对其有内置支持。Send API允许你动态创建工作者节点并向它们发送特定输入。每个工作者都有自己的状态，所有工作者的输出都会写入一个共享状态键，该键可供协调器图访问。这使协调器能够访问所有工作者的输出，并允许其将它们综合为最终输出。下面的示例遍历一个部分列表，并使用Send API将每个部分发送给一个工作者。
Copy
from langgraph.types import Send
# 图状态
class State(TypedDict):
topic: str # 报告主题
sections: list[Section] # 报告部分列表
completed_sections: Annotated[
list, operator.add
] # 所有工作者并行写入此键
final_report: str # 最终报告
# 工作者状态
class WorkerState(TypedDict):
section: Section
completed_sections: Annotated[list, operator.add]
# 节点
def orchestrator(state: State):
"""生成报告计划的协调器"""
# 生成查询
report_sections = planner.invoke(
[
SystemMessage(content="生成报告计划。"),
HumanMessage(content=f"以下是报告主题：{state['topic']}"),
]
)
return {"sections": report_sections.sections}
def llm_call(state: WorkerState):
"""工作者撰写报告的一个部分"""
# 生成部分
section = llm.invoke(
[
SystemMessage(
content="根据提供的名称和描述撰写报告部分。每个部分不要包含前言。使用markdown格式。"
),
HumanMessage(
content=f"以下是部分名称：{state['section'].name} 和描述：{state['section'].description}"
),
]
)
# 将更新的部分写入已完成的部分
return {"completed_sections": [section.content]}
def synthesizer(state: State):
"""从各部分综合完整报告"""
# 已完成部分列表
completed_sections = state["completed_sections"]
# 格式化已完成部分为字符串，用作最终部分的上下文
completed_report_sections = "\n\n---\n\n".join(completed_sections)
return {"final_report": completed_report_sections}
# 条件边函数，为计划中的每个部分创建一个llm_call工作者
def assign_workers(state: State):
"""为计划中的每个部分分配一个工作者"""
# 通过Send() API并行启动部分撰写
return [Send("llm_call", {"section": s}) for s in state["sections"]]
# 构建工作流
orchestrator_worker_builder = StateGraph(State)
# 添加节点
orchestrator_worker_builder.add_node("orchestrator", orchestrator)
orchestrator_worker_builder.add_node("llm_call", llm_call)
orchestrator_worker_builder.add_node("synthesizer", synthesizer)
# 添加边以连接节点
orchestrator_worker_builder.add_edge(START, "orchestrator")
orchestrator_worker_builder.add_conditional_edges(
"orchestrator", assign_workers, ["llm_call"]
)
orchestrator_worker_builder.add_edge("llm_call", "synthesizer")
orchestrator_worker_builder.add_edge("synthesizer", END)
# 编译工作流
orchestrator_worker = orchestrator_worker_builder.compile()
# 显示工作流
display(Image(orchestrator_worker.get_graph().draw_mermaid_png()))
# 调用
state = orchestrator_worker.invoke({"topic": "创建一份关于LLM缩放定律的报告"})
from IPython.display import Markdown
Markdown(state["final_report"])
评估器-优化器
在评估器-优化器工作流中，一个LLM调用创建响应，另一个LLM调用评估该响应。如果评估器或人在回路确定响应需要改进，则提供反馈并重新创建响应。此循环持续进行，直到生成可接受的响应。 当任务有特定的成功标准，但需要迭代才能满足该标准时，通常会使用评估器-优化器工作流。例如，在两种语言之间翻译文本时，并不总是存在完美的匹配。可能需要几次迭代才能生成在两种语言中含义相同的翻译。 evaluator_optimizer.png
Copy
# 图状态
class State(TypedDict):
joke: str
topic: str
feedback: str
funny_or_not: str
# 用于评估中结构化输出的模式
class Feedback(BaseModel):
grade: Literal["funny", "not funny"] = Field(
description="判断笑话是否好笑。",
)
feedback: str = Field(
description="如果笑话不好笑，请提供改进建议。",
)
# 使用结构化输出模式增强LLM
evaluator = llm.with_structured_output(Feedback)
# 节点
def llm_call_generator(state: State):
"""LLM生成笑话"""
if state.get("feedback"):
msg = llm.invoke(
f"写一个关于{state['topic']}的笑话，但请考虑以下反馈：{state['feedback']}"
)
else:
msg = llm.invoke(f"写一个关于{state['topic']}的笑话")
return {"joke": msg.content}
def llm_call_evaluator(state: State):
"""LLM评估笑话"""
grade = evaluator.invoke(f"给笑话评分 {state['joke']}")
return {"funny_or_not": grade.grade, "feedback": grade.feedback}
# 条件边函数，根据评估器的反馈路由回笑话生成器或结束
def route_joke(state: State):
"""根据评估器的反馈路由回笑话生成器或结束"""
if state["funny_or_not"] == "funny":
return "Accepted"
elif state["funny_or_not"] == "not funny":
return "Rejected + Feedback"
# 构建工作流
optimizer_builder = StateGraph(State)
# 添加节点
optimizer_builder.add_node("llm_call_generator", llm_call_generator)
optimizer_builder.add_node("llm_call_evaluator", llm_call_evaluator)
# 添加边以连接节点
optimizer_builder.add_edge(START, "llm_call_generator")
optimizer_builder.add_edge("llm_call_generator", "llm_call_evaluator")
optimizer_builder.add_conditional_edges(
"llm_call_evaluator",
route_joke,
{ # route_joke返回的名称 : 要访问的下一个节点名称
"Accepted": END,
"Rejected + Feedback": "llm_call_generator",
},
)
# 编译工作流
optimizer_workflow = optimizer_builder.compile()
# 显示工作流
display(Image(optimizer_workflow.get_graph().draw_mermaid_png()))
# 调用
state = optimizer_workflow.invoke({"topic": "猫"})
print(state["joke"])
智能体
智能体通常被实现为使用工具执行操作的LLM。它们在持续的反馈循环中运行，用于问题和解决方案不可预测的情况。智能体比工作流具有更多的自主性，可以决定使用哪些工具以及如何解决问题。你仍然可以定义可用的工具集和智能体行为的指导方针。 agent.png
使用工具
Copy
from langchain_core.tools import tool
# 定义工具
@tool
def multiply(a: int, b: int) -> int:
"""将a和b相乘。
参数：
a：第一个整数
b：第二个整数
"""
return a * b
@tool
def add(a: int, b: int) -> int:
"""将a和b相加。
参数：
a：第一个整数
b：第二个整数
"""
return a + b
@tool
def divide(a: int, b: int) -> float:
"""将a除以b。
参数：
a：第一个整数
b：第二个整数
"""
return a / b
# 使用工具增强LLM
tools = [add, multiply, divide]
tools_by_name = {tool.name: tool for tool in tools}
llm_with_tools = llm.bind_tools(tools)
Copy
from langgraph.graph import MessagesState
from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage
# 节点
def llm_call(state: MessagesState):
"""LLM决定是否调用工具"""
return {
"messages": [
llm_with_tools.invoke(
[
SystemMessage(
content="你是一个乐于助人的助手，负责对一组输入执行算术运算。"
)
]
+ state["messages"]
)
]
}
def tool_node(state: dict):
"""执行工具调用"""
result = []
for tool_call in state["messages"][-1].tool_calls:
tool = tools_by_name[tool_call["name"]]
observation = tool.invoke(tool_call["args"])
result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
return {"messages": result}
# 条件边函数，根据LLM是否进行工具调用路由到工具节点或结束
def should_continue(state: MessagesState) -> Literal["tool_node", END]:
"""根据LLM是否进行工具调用决定是否继续循环或停止"""
messages = state["messages"]
last_message = messages[-1]
# 如果LLM进行了工具调用，则执行操作
if last_message.tool_calls:
return "tool_node"
# 否则，我们停止（回复用户）
return END
# 构建工作流
agent_builder = StateGraph(MessagesState)
# 添加节点
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)
# 添加边以连接节点
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges(
"llm_call",
should_continue,
["tool_node", END]
)
agent_builder.add_edge("tool_node", "llm_call")
# 编译智能体
agent = agent_builder.compile()
# 显示智能体
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))
# 调用
messages = [HumanMessage(content="3加4是多少。")]
messages = agent.invoke({"messages": messages})
for m in messages["messages"]:
m.pretty_print()