AI Agent工作流的5种基础模式及其通用实现:评估器-优化器模式
AI Agent工作流的5种基础模式及其通用实现:评估器-优化器模式
反思(Reflection)是一种广泛应用于LLM应用中的提示与工作流模式,特别在Agent应用中,反思可以极大的提高输出质量与任务成功率。本文将探讨Anthropic的《Build effective agents》文中关于Workflows的最后一种基础反思模式:Evaluator-Optimizer(评估器-优化器)模式。**
除了基础反思模式外,还有一些扩展的反思模式。比如可以借助外部知识的强化反思模式、结合了自我评估与树搜索的任务优化模式(LATS,Language Agent Tree Search),我们将用专门的文章来介绍这些增强的反思模式。**
基础反思模式(Evaluator-Optimizer)
反思是这样一种工作模式:一个“增强LLM”调用负责生成响应,另一个则负责评估与反馈,两者之间经过多次迭代,最后输出更高质量的响应结果。
显然,这种模式也可以拓展到多Agent系统(Multi-Agent System)中。当我们的输入任务有比较明确的期望、要求与评估标准,都可以借助这种方式来不断的评估与完善任务结果。典型的场景比如:
- 长文翻译:从语言风格、专业名词、前后一致性等方面自我审查
- 代码编写:从正确性、复杂度、效率、编码风格等方面审查代码
- 文案/报告创作:从格式、完备性、目录、风格等方面审查结果
这种模式在实际应用中仍然有改进空间,否则你可能会发现,多次的简单反思并不会带来更多的改进。一些可能的改进方式比如:
- 评估时以不同的角色,从不同的角度对响应进行评价
- 借助于外部数据或知识来对已有响应做评估与增强
PydanticAI实现基础反思模式
实现基础反思模式的过程如下(基于PydanticAI):
- 定义生成器与评估器的输出类型
from pydantic import BaseModel, Field
# 生成器响应模型
class GeneratorResponse(BaseModel):
thoughts: str = Field(..., description='你对任务的理解和反馈,或者你计划如何改进。')
response: str = Field(..., description='生成的解决方案。')
# 评估器响应模型
class EvaluatorResponse(BaseModel):
thoughts: str = Field(..., description='你对提交内容的仔细和详细的审查和评估。')
evaluation: str = Field(..., description='通过, 需要改进, 或失败')
feedback: str = Field(..., description='需要改进的地方和原因。')
- 定义生成器与评估器的提示与模型
仍然采用配置的方式来完成,这里注意根据任务的不同,你可能需要调整提示词,以实现更加针对性的反思。
from pydantic_ai import Agent
steps = {
"generator": {
"prompt": """你的目标是根据用户输入完成任务。如果你之前生成的内容收到了反馈,请根据这些反馈改进你的解决方案。""",
"model": model,
"result_type": GeneratorResponse
},
"evaluator": {
"prompt": """
请对以下代码实现进行评估,重点关注以下方面:
1. **代码正确性**:是否完全按照规范无误地实现了要求的功能?
2. **时间复杂度**:实现是否满足规定的时间复杂度要求?
3. **效率**:实现是否是针对需求最有效、最优化的方案?
4. **风格与最佳实践**:代码是否遵循标准的 Python 风格和最佳实践?
5. **可读性**:代码是否易于阅读和理解?
6. **文档化**:代码是否有清晰的文档说明,包括为所有函数和类撰写的 docstrings,以及必要的内嵌注释
注意:你应该仅评估代码,而不是尝试解决任务。
请仔细且严格地评估代码,确保不会错过任何改进的机会。
如果所有评估标准都完全满足且没有进一步改进建议,请输出“PASS”。否则,请输出“NEEDS_IMPROVEMENT”或“FAIL”,以便编码者能够学习和改进。
""",
"model": model,
"result_type": EvaluatorResponse
}
}
- 实现生成器与评估器的LLM调用
这个过程本身没有复杂性,这里借助PydanticAI的Agent组件实现,实际上也可以借助LLM API直接实现。
import asyncio
async def generate(task: str, context: str = "") -> tuple[str, str]:
"""根据反馈生成和改进解决方案。"""
config = steps["generator"]
system_prompt = config["prompt"]
if context:
system_prompt += f"\n\n{context}"
generator_agent = Agent(config["model"], system_prompt=system_prompt, result_type=config["result_type"])
response = await generator_agent.run(f'任务:\n{task}')
thoughts = response.data.thoughts
result = response.data.response
return thoughts, result
async def evaluate(content: str, task: str) -> tuple[str, str]:
"""评估解决方案是否符合要求。"""
config = steps["evaluator"]
evaluator_agent = Agent(config["model"], system_prompt=f'{config["prompt"]}\n\n任务:\n{task} ', result_type=config["result_type"])
response = await evaluator_agent.run(content)
evaluation = response.data.evaluation
feedback = response.data.feedback
return evaluation, feedback
- 主循环
借助生成器与评估器之间的配合与迭代,最终获得输出结果。在实际应用中,最好要设置最大迭代次数(max_iterations)作为终止条件,而不能完全的依赖于LLM评估器对结果的判断,否则有可能陷入死循环。
async def main(task: str, max_iterations: int = 5):
memory = []
iteration = 0
thoughts, result = await generate(task)
while iteration < max_iterations:
evaluation, feedback = await evaluate(result, task)
memory.append({"thoughts": thoughts, "result": result, "evaluation": evaluation, "feedback": feedback})
print(f"\nIteration {iteration + 1}:\nThoughts: {thoughts}\nResult: {result}\nEvaluation: {evaluation}\nFeedback: {feedback}")
if evaluation == "PASS":
return result, memory
context = "\n".join([
"之前的尝试:",
*[f"- 结果: {m['result']}\n 反馈: {m['feedback']}" for m in memory]
])
thoughts, result = await generate(task, context)
iteration += 1
以上就是基础反思模式的核心实现。如果使用输入任务进行测试,就可以看到类似如下的输出,体现了其中的评估与优化改进过程:
小结
在Anthropic的这篇《Build effective agents》文章中,比较清晰的帮我们区分了Agent系统的两种类型:Workflows与Agents,并介绍了Workflows中常见的5种基础模式:顺序、并行、路由、编排-工作、反思-优化模式。
在我们的系列文章中,对这些基础模式借助更轻量级的PydanticAI做了通用实现。在理解了这些模式的基础上,借助于类似于LangGraph/LlamaIndex Workflows这样的框架,就可以方便的组合与拓展出复杂性更高的AI工作流。相对于更“黑盒”的Agents,Workflows牺牲了一定的灵活性,但是大大提高了可控性,这对于目前在LLM的长任务推理能力尚不足的情况下,实现企业级的Agent系统是至关重要的。