AI 智能体 - 目标设定与监控模式

# pip install langgraph langchain-openai pydanticfrom typing import TypedDict, List, Annotated, Dictimport operatorfrom langchain_core.messages import BaseMessage, HumanMessage, AIMessagefrom langchain_openai import ChatOpenAIfrom langgraph.graph import StateGraph, ENDfrom pydantic import BaseModel, Field
# 1. 定义智能体状态 (State)class CodeAgentState(TypedDict):    """用于在 LangGraph 节点间传递的状态"""    messages: Annotated[List[BaseMessage], operator.add] # 聊天记录和历史    code: str         # 当前生成的代码    goals: List[str]  # 智能体要达成的目标列表    review_feedback: str # 评审员的详细反馈    is_met: bool      # 目标是否已达成 (由 Judge 节点设置)    iterations: int   # 当前迭代次数
# 2. 定义确定性工具：代码执行器 (用于监控)# 🚨 警告：生产环境请务必使用沙箱执行代码！def execute_python_code(code: str) -> str:    """执行 Python 代码，返回 stdout 或错误信息"""    try:        import io, sys        old_stdout = sys.stdout        redirected_output = io.StringIO()        sys.stdout = redirected_output                # 移除可能破坏环境的危险代码（简单的安全检查）        if "os.remove" in code or "shutil." in code:            return "Execution Error: Forbidden operations detected."
        exec(code, {})        sys.stdout = old_stdout # 恢复 stdout        return f"Execution Success. Output:\n{redirected_output.getvalue()}"    except Exception as e:        sys.stdout = old_stdout        return f"Execution Failure: {str(e)}"

def generate_code(state: CodeAgentState) -> CodeAgentState:    """节点：根据目标和反馈生成或修复代码"""        current_goals = state['goals']    feedback = state['review_feedback']        # 构造 LLM 提示词    goals_list = "\n".join(f"- {g}" for g in current_goals)        system_prompt = f"""    你是一个专业的 Python 程序员。你的目标是编写完全符合以下要求的代码：        【目标列表】:    {goals_list}        {'---' if feedback else ''}    {'【上次代码及反馈】:' + feedback if feedback else ''}    {'---' if feedback else ''}        请只返回完整的、可执行的 Python 代码块。不要包含任何解释或额外文字。    """
    model = ChatOpenAI(model="gpt-4o", temperature=0.3)        # 如果有反馈，将反馈添加到历史消息中进行引导    if feedback:        input_message = HumanMessage(            content=f"请根据以上反馈和目标，重写并优化代码。"        )    else:        # 初始任务        input_message = HumanMessage(            content="开始编写代码。请生成初始版本。"        )            messages_in = [HumanMessage(content=system_prompt), input_message]        response = model.invoke(messages_in)        # 清理代码块（与原示例类似）    code_content = response.content.replace("```python", "").replace("```", "").strip()
    print(f"🔄 Coder 节点：生成代码（Iteration {state['iterations'] + 1}）")        return {        "messages": [response],        "code": code_content,        "iterations": state['iterations'] + 1    }

def review_and_execute(state: CodeAgentState) -> CodeAgentState:    """节点：执行代码并进行 LLM 评审"""        code = state['code']    goals = state['goals']        # 1. 确定性监控：执行代码并捕获错误    execution_result = execute_python_code(code)    print(f"✅ Executor 节点：执行结果：{'Success' if 'Success' in execution_result else 'Failure'}")        # 2. 非确定性监控：LLM 给出详细的、基于目标的反馈    review_prompt = f"""    你是一个专业的代码评审员。你的任务是根据以下目标，对代码进行全面、客观的审查。        【目标列表】:    {chr(10).join(f"- {g}" for g in goals)}
    【代码执行结果】:    {execution_result}
    【待审查代码】:    {code}
    请给出详细的改进建议，重点关注：功能正确性、目标达成度、可读性和边界处理。    """        model = ChatOpenAI(model="gpt-4o", temperature=0.2)    feedback_response = model.invoke(review_prompt)        print("📝 Reviewer 节点：生成详细反馈...")        return {        "messages": [feedback_response],        "review_feedback": feedback_response.content    }

class GoalAssessment(BaseModel):    """用于 LLM 输出的 Pydantic 模型"""    is_met: bool = Field(description="如果所有目标均已满足，则为 True；否则为 False。")    reasoning: str = Field(description="解释做出此判断的简短理由。")
def judge_goals_met(state: CodeAgentState) -> CodeAgentState:    """节点：判断目标是否达成，并返回结构化结果"""        goals = state['goals']    feedback = state['review_feedback']        judge_prompt = f"""    你是一个最终的决策者。请根据以下目标和代码评审反馈，严格判断目标是否全部达成。        【目标列表】:    {chr(10).join(f"- {g}" for g in goals)}
    【代码评审反馈】:    {feedback}        如果反馈中仍有任何明显的、需要修复的建议，即使代码可以运行，也请判定为 False。    请按照 JSON 格式返回结果，并严格遵循提供的 Pydantic 模型。    """        model = ChatOpenAI(model="gpt-4o", temperature=0)        # 强制模型返回 Pydantic 结构    response = model.with_structured_output(GoalAssessment).invoke(judge_prompt)        print(f"⚖️ Judge 节点：目标达成？ {response.is_met}。原因: {response.reasoning}")        return {        "messages": [AIMessage(content=response.reasoning)],        "is_met": response.is_met    }

def route_next_step(state: CodeAgentState) -> str:    """路由函数：根据 Judge 的判断决定下一步"""    if state['is_met']:        return "end"    if state['iterations'] >= 5: # 设定最大迭代次数，防止无限循环 (SMART-T)        print("🛑 达到最大迭代次数，停止优化。")        return "end"        # 否则，回到 Coder 节点进行重写 (反馈驱动)    return "replan"
# 1. 初始化图workflow = StateGraph(CodeAgentState)
# 2. 添加节点workflow.add_node("coder", generate_code)workflow.add_node("reviewer", review_and_execute)workflow.add_node("judge", judge_goals_met)workflow.add_node("finalizer", lambda state: state) # 最终处理节点
# 3. 设置起点workflow.set_entry_point("coder")
# 4. 定义边 (顺序执行)workflow.add_edge("coder", "reviewer")workflow.add_edge("reviewer", "judge")
# 5. 定义条件边 (反馈循环)workflow.add_conditional_edges(    "judge",    route_next_step,    {        "replan": "coder",    # 目标未达成 -> 循环回 Coder (规划/调整)        "end": "finalizer"    # 目标达成或达到限制 -> 结束    })
# 6. 结束图workflow.add_edge("finalizer", END)
# 7. 编译app = workflow.compile()
# --- 运行测试 ---if __name__ == "__main__":    # 高层目标：计算斐波那契数列，并包含严格的质量标准    initial_goals = [        "功能正确，能计算第 N 个斐波那契数。",        "代码必须使用迭代法（非递归），以优化性能。",        "必须包含清晰的类型注解和文档字符串。",        "必须能处理 N <= 0 的边界情况，返回异常。",    ]
    initial_state = {        "messages": [],        "code": "",        "goals": initial_goals,        "review_feedback": "",        "is_met": False,        "iterations": 0    }
    print("🚀 启动目标驱动代码智能体...")    final_state = app.invoke(initial_state)
    print("\n" + "="*80)    print("✨ 最终输出结果:")    print("="*80)    print(final_state['code'])    print(f"\n目标达成状态: {'成功' if final_state['is_met'] else '未完全达成 (最大迭代限制)'}")