場景：

用戶指定冷笑話主題，生成冷笑話后，進行評估，如果不搞笑就需要重新生成

以下代碼實現了一個基于LangGraph的冷笑話自動生成與評估工作流。

系統包含兩個核心節點：生成器根據用戶主題創作冷笑話，評估器對笑話進行幽默評級并提供改進建議。通過條件路由機制，當笑話被評為不搞笑 時會循環生成新版本，直到產生合格的笑話為止。工作流采用結構化輸出確保評估反饋的標準化，并利用大語言模型進行內容生成和質量判斷。系統實現了從初始創作到多次迭代優化的完整笑話生產閉環。

上代碼：

# 導入所需的類型和模塊
from typing import TypedDict, Literal
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langgraph.constants import START, END
from langgraph.graph import StateGraph
from pydantic import BaseModel, Field
from src.agent.aitool2 import llm# 定義狀態結構的類型提示，用于在圖中傳遞狀態
class State(TypedDict):joke: str                  # 生成的冷笑話內容topic: str                 # 用戶指定的主題feedback: str              # 改進建議funny_or_not: str          # 幽默評級（"funny" 或 "not funny"）# 定義結構化輸出模型，用于LLM評估反饋的格式化輸出
class Feedback(BaseModel):"""使用此工具來結構化你的響應"""# 定義笑話評估的等級，只能是 "funny" 或 "not funny"grade: Literal["funny", "not funny"] = Field(description="判斷笑話是否幽默",examples=["funny", "not funny"])# 提供具體的反饋建議feedback: str = Field(description="若不幽默，提供改進建議",example="可以加入雙關語或意外結局")# 節點函數：負責生成冷笑話
def generator_func(state: State):"""由大模型生成一個冷笑話的節點"""# 根據是否有反饋來構建提示詞prompt = (f"根據反饋改進笑話：{state['feedback']}\n主題：{state['topic']}"if state.get("feedback", None)  # 如果存在反饋，則使用反饋+主題的提示else f"創作一個關于{state['topic']}的笑話"  # 否則只使用主題生成新笑話)# 創建一個鏈式調用：LLM + 字符串輸出解析器chain = llm | StrOutputParser()# 執行鏈式調用并獲取響應resp = chain.invoke(prompt)# 返回更新后的狀態，包含新生成的笑話return {'joke': resp}# 節點函數：負責評估生成的冷笑話
def avaluator_func(state: State):"""評估狀態中的冷笑話"""# 綁定工具（Feedback模型）到LLM，使LLM能夠以結構化方式調用工具chain = llm.bind_tools([Feedback])# 構造評估提示并調用LLM進行評估evaluation = chain.invoke(f"評估此笑話的幽默程度：\n{state['joke']}\n""注意：幽默應包含意外性或巧妙措辭")# 從LLM響應中提取最后一次工具調用的參數evaluation = evaluation.tool_calls[-1]['args']# 返回更新后的狀態，包含評估結果和反饋return {"funny_or_not": evaluation['grade'],      # 幽默評級"feedback": evaluation['feedback']        # 改進建議}# 條件邊的路由函數：根據評估結果決定下一步流程
def route_func(state: State) -> str:"""動態路由決策函數"""# 如果笑話被評為"funny"，則接受并結束流程# 否則拒絕并返回反饋，重新生成return 'Accepted' if state.get("funny_or_not", None) == "funny" else "Rejected + Feedback"# 構建工作流圖
builder = StateGraph(State)  # 使用 State 類型創建狀態圖構建器# 添加節點到圖中
builder.add_node('generator', generator_func)   # 添加生成笑話的節點
builder.add_node('avaluator', avaluator_func)   # 添加評估笑話的節點# 定義節點間的連接關系
builder.add_edge(START, 'generator')            # 從開始節點連接到生成器節點
builder.add_edge('generator', 'avaluator')      # 從生成器連接到評估器節點# 添加條件邊：根據評估結果決定流程走向
builder.add_conditional_edges('avaluator',                                # 從評估器節點出發route_func,                                 # 使用路由函數決定走向{"Accepted": END,                        # 如果接受，則流程結束"Rejected + Feedback": "generator"      # 如果拒絕，則返回生成器節點重新生成}
)# 編譯圖結構為可執行的工作流
graph = builder.compile()

測試結果