在Loop Engineering在MAF中的实现-06:利用Workflow来使用LoopAgent的循环策略中，我们采用MAF的工作流实现了LoopAgent类似的循环。既然可以利用MAF的Workflow实现，自然能利用LangGraph来实现。鉴于两者之间设计上的差异，相同的功能如果采用这两种解决方案，使用LangGraph大概率会更加简单。

1. 采用LangGraph来构建诗词创作Agent

我们将类似于LoopAgent的根据评估结果循环调用指定Agent的逻辑实现在如下这个工厂方法，该方法会创建一个自动循环执行的Agent对象。

def create_loop_agent(
        agent: Runnable,
        evaluator: Callable[[LoopContext], Awaitable[LoopEvaluation]],
        max_iterations:int = 100)

class LoopContext(TypedDict):
    interation: int
    init_messages: list[AnyMessage]
    last_response: AIMessage
    feedback: list[str]

class LoopEvaluation(TypedDict):
    stop: bool
    feedback: str

create_loop_agent函数的三个参数说明如下：

  • agent: 被封装的目标Agent，即循环调用的对象；
  • evaluator：通过一个异步函数表示的评估器，输入为表示循环上下文的LoopContext类型，返回值LoopEvaluation表示评估结果，两个字段分别表示是否终止以及评估反馈；
  • max_iterations：允许的最多迭代次数。

LoopContext的四个上下文成员：

  • interation：当前迭代次数；
  • init_messages： 初始传入的消息列表；
  • last_response：最后一下响应的消息；
  • feedback： 之前的评估反馈列表。

如下定义的composer_system_message和reviewer_system_message两个变量是我们为创作Agent和调用评估模型定义的系统指令。prompt为调用loop_agent传入的提示词。

composer_system_message = SystemMessage(content="你是一个精通宋词创作的智能体，负责根据提供的主题、意境和评估反馈以词牌名**相见欢**创作一首词")
reviewer_system_message = SystemMessage(content="""\
你是一个深谙古典文化和诗词的文化大家，请按照如下的标准对创作的这首词进行评价。评语保持500字以内，尽可能简洁。

1. 格律规范 (Metrical Accuracy) —— 30分

检测AI生成的词是否真正符合该词牌的“说明书”。

- 词牌结构（20分）：总字数、分片、每句的字数长短必须与词牌名保持一致。
- 平仄规范（5分）：关键位置的平仄（平声、仄声）必须严格符合词牌要求。
- 押韵规则（5分）：检查是否在规定位置押韵，是否混淆了平仄韵，有无出韵。

2. 意境创设 (Imagery & Conception) —— 30分

评估词作是否具备古典美感，是否能画出**画面感**。

- 意象选用（15分）：是否恰当使用了符合古典美学的意象。意象之间是否逻辑自洽，没有现代感违和物。
- 流派风格（15分）：整体意境的深远程度。

3. 语言艺术 (Linguistic Artistry) —— 20分

评估遣词造句的功底和文字的流畅度。

- 用词典雅（10分）：遣词造句需有**词味**，杜绝使用现代大白话或过于生硬的拼凑词。
- 对仗与过片（10分）：若词牌要求对仗，需检查是否工整；重点评估**过片**（上下片过渡）是否自然流转，有无断层。

4. 情感寄托与创新 (Emotional Depth & Innovation) —— 20分

评估诗词的灵魂，拒绝纯粹的字词堆砌。

- 情感共鸣（10分）：词中所表达的悲欢离合、家国情怀或羁旅之思是否真挚饱满。
- 陈词翻新（10分）：是否只会堆砌**愁**、**泪**等陈词滥调。优秀的词作应当在传统框架下有独特的视角或新颖的构思。
""")    

prompt ="""\
基于如下这首《卫风·氓》的背景和情感基调，创作**一首**宋词。

原文如下：

氓之蚩蚩，抱布贸丝。匪来贸丝，来即我谋。
送子涉淇，至于顿丘。匪我愆期，子无良媒。
将子无怒，秋以为期。

乘彼垝垣，以望复关。不见复关，泣涕涟涟。
既见复关，载笑载言。尔卜尔筮，体无咎言。
以尔车来，以我贿迁。

桑之未落，其叶沃若。于嗟鸠兮，无食桑葚！
于嗟女兮，无与士耽！士之耽兮，犹可说也；
女之耽兮，不可说也。

桑之落矣，其黄而陨。自我徂尔，三岁食贫。
淇水汤汤，渐车帷裳。女也不爽，士贰其行。
士也罔极，二三其德。

三岁为妇，靡室劳矣；夙兴夜寐，靡有朝矣。
言既遂矣，至于暴怒。兄弟不知，咥其笑矣。
静言思之，躬自悼矣。

及尔偕老，老使我怨。淇则有岸，隰则有泮。
总角之宴，言笑晏晏。信誓旦旦，不思其反。
反是不思，亦已焉哉！
"""

整个演示程序如下所示。我们调用create_agent函数创建了用于宋词创作的Agent。由于评估器通过体现一个Callable[[LoopContext], Awaitable[LoopEvaluation]]对象，所以定义了与之匹配的evaluate函数来完成评估工作。evaluate函数采用结构化输出调用ChatOpenAI对象将评估结果反序列化成一个Review对象，并生成返回的LoopEvaluation对象。这里体现的评估逻辑是：如果得分低于75则通过循环重新创作，同时将评语反馈给后续迭代。

from langchain.messages import AnyMessage, SystemMessage, HumanMessage
from langchain.agents import create_agent
from typing import TypedDict
from dotenv import load_dotenv
from langchain_openai  import ChatOpenAI
from loop_agents import LoopContext, LoopEvaluation, create_loop_agent
import asyncio

load_dotenv()

"""评估结果"""
class Review(TypedDict):
    """分数（百分制）"""
    score: int
    """评语"""
    feedback:str

composer = create_agent(
    model= ChatOpenAI(model="gpt-5.4-mini"),
    system_prompt=composer_system_message)

reviewer = ChatOpenAI(model="gpt-5.4-mini")
reviewer = reviewer.with_structured_output(Review)

async def evaluate(context: LoopContext)-> LoopEvaluation:
    messages:list[AnyMessage] = [reviewer_system_message, *context["init_messages"], context["last_response"]]
    review = await reviewer.ainvoke(input=messages)
    return {
        "stop": review["score"] >= 75,
        "feedback":review["feedback"] + f"\n\n综合得分：{review["score"] }"
    }

agent = create_loop_agent(
    agent=composer,
    evaluator=evaluate,
    max_iterations=3
)

async def main():
    message = HumanMessage(content=prompt)
    result = await agent.ainvoke(input={
        "init_messages":[message],
        "messages":[message]})
    
    for message in result.get("messages"):
        message.pretty_print()   

asyncio.run(main())

我们利用创建的Agent和评估函数，调用create_loop_agent函数将具有循环调用能力的Agent创建出来，并传入提示词进行调用。输出的消息列表体现了Agent调用+评估的两次循环迭代：两次创作得分分别为70和81。

================================ Human Message =================================

基于如下这首《卫风·氓》的背景和情感基调，创作**一首**宋词。

原文如下：

氓之蚩蚩，抱布贸丝。匪来贸丝，来即我谋。
送子涉淇，至于顿丘。匪我愆期，子无良媒。
将子无怒，秋以为期。

乘彼垝垣，以望复关。不见复关，泣涕涟涟。
既见复关，载笑载言。尔卜尔筮，体无咎言。
以尔车来，以我贿迁。

桑之未落，其叶沃若。于嗟鸠兮，无食桑葚！
于嗟女兮，无与士耽！士之耽兮，犹可说也；
女之耽兮，不可说也。

桑之落矣，其黄而陨。自我徂尔，三岁食贫。
淇水汤汤，渐车帷裳。女也不爽，士贰其行。
士也罔极，二三其德。

三岁为妇，靡室劳矣；夙兴夜寐，靡有朝矣。
言既遂矣，至于暴怒。兄弟不知，咥其笑矣。
静言思之，躬自悼矣。

及尔偕老，老使我怨。淇则有岸，隰则有泮。
总角之宴，言笑晏晏。信誓旦旦，不思其反。
反是不思，亦已焉哉！

================================== Ai Message ==================================

【相见欢】  
当年初见斜阳，过横塘。  
一诺轻轻，笑语付疏妆。  

桑叶落，心先薄，泪成行。  
淇水汤汤，流尽旧时光。
================================ Human Message =================================

词意能承《卫风·氓》之怨悔与决绝，情感脉络较清楚，淇水、桑叶等意象亦合古典语境，结句有余味。但就“宋词”而言，词牌未明，结构、字数、句式、平仄与押韵均无从称合，严格说不合格律规范。语言尚有词味，但“斜阳、横塘、泪成行、旧时光”等语稍显泛化，略近今人语感，且新意不足，仍多沿用闺怨陈辞。整体可算有情而未入法，重在意境传达，格律功夫明显欠缺。

综合得分：70
================================== Ai Message ==================================

【相见欢】  
当年一笑相逢。  
过淇东。  
布贸无非，来语问深衷。  

秋已约，心难托，誓成空。  
桑落人归，回首怨东风。
================================ Human Message =================================

此词取《氓》之怨与悔，情绪线索尚明，但格律上问题较大：词牌结构更接近小令却未严守《相见欢》定式，句式、字数、分片与押韵均不够规范，平仄亦难尽合。意象方面以“淇”“桑”“秋”“东风”承接原诗，古典感尚可，但略显提炼不足，画面较单薄。语言总体较雅，未见明显白话杂语；然“布贸无非，来语问深衷”稍嫌生硬，过片亦略显跳脱。情感上能写出由欢而怨、由信而悔的转折，贴合原作精神，但创新性一般，仍多依附原诗意脉。

综合得分：81

2. 状态类型

Loop Engineering在MAF中的实现-06:利用Workflow来使用LoopAgent的循环策略的设计只要体现将涉及的状态，包括对话历史和当前迭代的循环上下文，存储在Workflow的上下文中。与之类似，如果使用LangGraph，我们就可以直接将它作为状态图的状态。这个状态类型就是如下这个继承自AgentState的TypedDict：

class State(AgentState):
    interation: Required[Annotated[int, operator.add]]
    init_messages: list[AnyMessage]
    last_response: AIMessage
    stop: bool
    feedback: Required[Annotated[list[str],operator.add]]

除了继承自AgentState的核心成员messages之外，自身定义的状态字段说明如下：

  • interation：迭代次数，应用的reducer函数实现自增；
  • init_messages：初始调用指定的消息列表；
  • last_response：当前迭代从LLM返回的AIMessage；
  • stop: 决定是否继续循环还是立即终止；
  • feedback: 历次评估收集的反馈，应用的reducer函数实现追加功能。

3. create_loop_agent的实现

create_loop_agent函数返回的是通过状态图（基于状态类型State的StateGraph对象）构建并编译生成的Agent对象。

from langgraph.graph import StateGraph
from langchain_core.runnables import Runnable
from langchain.messages import AnyMessage, AIMessage,HumanMessage
from langchain.agents import AgentState
from typing import  Any, Required,Annotated, TypedDict,Callable,Awaitable
from langgraph.graph.message import add_messages
import operator, functools
  
async def invoke_agent(agent: Runnable,state: State)->dict[str, Any]:
    messages = state.get("messages", state.get("init_messages"))
    result = await agent.ainvoke(input= {"messages":messages})
    new_messages = result["messages"][len(messages):-1]
    response: AnyMessage = result["messages"][-1]
    return  {
        "messages": new_messages,
        "interation": 1,
        "last_response": response
    }

async def evaluate(
        evaluator: Callable[[LoopContext], Awaitable[LoopEvaluation]], 
        state: State)-> dict[str, Any]:
    context: LoopContext = {
        "interation": state.get("interation"),
        "init_messages": state.get("init_messages"),
        "last_response":state.get("last_response"),
        "feedback": state.get("feedback")
    }
    evaluation = await evaluator(context)
    return {
        "stop": evaluation["stop"],
        "feedback":[evaluation["feedback"]],
        "messages":[HumanMessage(content=evaluation["feedback"])]
    }

def route(
        max_iterations:int,
        state:State)-> str:    
    if state.get("stop") or state.get("interation") >= max_iterations:
        return "__end__"
    return "invoke_agent"

def create_loop_agent(
        agent: Runnable,
        evaluator: Callable[[LoopContext], Awaitable[LoopEvaluation]],
        max_iterations:int = 100):  
    return (StateGraph(State)
        .add_node("invoke_agent", functools.partial(invoke_agent, agent))
        .add_node("evaluate", functools.partial(evaluate, evaluator))
        .set_entry_point("invoke_agent")
        .add_edge(start_key="invoke_agent", end_key="evaluate")
        .add_conditional_edges(source="evaluate", path= functools.partial(route, max_iterations) )
        .compile())

它具有两个节点，函数invoke_agent代表的初始节点用来执行指定的Agent，并通过返回的字典更新如下的状态：

  • messages：将响应消息添加到对话历史中；
  • interation: 迭代次数加一；
  • last_response：将响应的最后一条AIMessage作为本次迭代的响应。

函数evaluate代表的节点会根据当前状态创建LoopContext上下文对象，并调用指定的评估函数并得到代表评估结果的LoopEvaluation对象，然后根据评估结果更新如下的状态：

  • stop： 是否终止循环；
  • feedback：将评估反馈追加到反馈文本列表中；
  • messages：将评估反馈文本转换成HumanMessage追加到对话历史中，意味着后续的Agent调用会参考本次评估的反馈。

我们在invoke_agent和evaluate节点之间添加了双向边，从invoke_agent到evaluate节点之间是一条静态边，反向则是一条条件边，分支条件实现在函数route中：

  • 如果stop状态为true，或者迭代次数超出阈值，通过跳转到__end__节点立即停止执行；
  • 否则跳转到invoke_agent节点。
# AI
Logo
CSDN-OPC开发者社区

这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容，并连接云服务、办公空间等稀缺资源，助你专注创造，无忧运营。

更多推荐

cover

从零构建 Java 智能体 RAG 系统：Milvus 向量数据库实战指南

avatar CSDN-OPC开发者社区
cover

DeepSeek-Harness-入门指南

avatar CSDN-OPC开发者社区
cover

提升办公效率：OpenClaw 本地自动化 AI 工具搭建实战教程

avatar CSDN-OPC开发者社区