1. 使用langgraph实现agent步骤

1. 确定工作流中的不同step，每个step都将成为一个 node，然后规划出这些 node 之间的连接方式(即点和边)。
2. 确认每个node内部的工作：
   • 调用模型输出文本或推理决策
   • 获取外部数据信息
   • 调用 Tool 执行操作
   • 用户输入进行人工干预
3. 设计 state 结构：状态是agent 中所有node都能访问的共享memory
   • 存储各个 **node **中都需要的固定话知识
   • state 应存储原始数据，而非格式化文本
4. 构造 node
   • node 定义：一个python函数，接受state并返回对state的更新(追加)
   • 错误处理策略
     • 暂时性错误（网络问题、速率限制）：设置 retry_policy
     • 大语言模型可恢复错误（工具故障、解析问题）: 存储到state LLM循环解决
     • 用户可修复的错误（信息缺失、指令不明确）：Human interrupt()
     • 意外错误 ：开发者修复agent
   • node 尽量功能单一不要组合多功能，越小越好
5. 连接 node
   • 为了通过 interrupt() 实现 human-in-the-loop，我们需要使用 checkpointer 进行编译，这样才能在中断前后保存 state
   • 通过 Command goto 方式在 node 内部实现路由，流程清晰可追踪

2. Workflows and agents

Workflow: 具有预先确定的代码路径，并且设计为按特定顺序运行

Agent: 动态性，能自主定义自身流程和工具使用方式

2.1. Workflow demo

当一项任务有特定的成功标准，但需要通过迭代来满足该标准时，评估器-优化器工作流通常会被采用

模型生成，模型评估，通过则结束，不通过则返回模型生成继续

import yaml
from pathlib import Path
from typing import Literal

from openai import BaseModel
from pydantic import Field
from typing_extensions import TypedDict

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END


config_path = Path(__file__).parent.parent.parent / "config.yaml"
with open(config_path, 'r', encoding='utf-8') as f:
    config = yaml.safe_load(f)

# 初始化模型（从配置文件读取模型配置）
llm = ChatOpenAI(
    model=config['model'],
    api_key=config['api_key'],
    base_url=config['base_url'],
    temperature=config['temperature']
)

class State(TypedDict):
    joke: str
    topic: str
    feedback: str
    funny_or_not: str


# 用于评估的结构化输出 Schema
class Feedback(BaseModel):
    grade: Literal["funny", "not funny"] = Field(
        description="判断笑话是否有趣",
    )
    feedback: str = Field(
        description="如果笑话不有趣，提供改进建议",
    )


# 使用 schema 增强 LLM 实现结构化输出
evaluator = llm.with_structured_output(
    Feedback,
    method="function_calling"
)


# Nodes
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}


# 根据 llm_call_evaluator 的反馈决定是返回 llm_call_generator 还是 END
def route_joke(state: State):
    """根据 llm_call_evaluator 的反馈决定是返回 llm_call_generator 还是 END"""

    if state["funny_or_not"] == "funny":
        return "Accepted"
    elif state["funny_or_not"] == "not funny":
        return "Rejected Feedback"


# Build workflow
optimizer_builder = StateGraph(State)

# Add the nodes
optimizer_builder.add_node("llm_call_generator", llm_call_generator)
optimizer_builder.add_node("llm_call_evaluator", llm_call_evaluator)

# Add edges to connect nodes
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,
    {
        "Accepted": END,
        "Rejected Feedback": "llm_call_generator",
    },
)

optimizer_workflow = optimizer_builder.compile()

state = optimizer_workflow.invoke({"topic": "狗"})

print(state["joke"])

2.2. Agents demo

Agent 通常是通过大型语言模型（LLM）借助 tools 来执行操作的

两个 Node, 一个模型决策，一个 Tools 调用

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()

messages = [HumanMessage(content="Add 3 and 4.")]
messages = agent.invoke({"messages": messages})

for m in messages["messages"]:
    m.pretty_print()

参考文档：
Thinking in LangGraph
Workflows and agents