智能体设计模式全面解析：从基础到高级

from typing import Annotated, TypedDict, Literal
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI


class AgentState(TypedDict):
    """ReAct 智能体状态定义"""
    messages: Annotated[list, "消息历史，包含所有对话和工具调用记录"]
    iteration: int  # 当前迭代次数，用于熔断


def create_react_agent(tools: list, max_iterations: int = 10):
    """
    创建 ReAct 模式智能体
    
    Args:
        tools: 可用工具列表
        max_iterations: 最大迭代次数，防止死循环
    """
    llm = ChatOpenAI(model="gpt-4").bind_tools(tools)
    tool_node = ToolNode(tools)

    def reasoning_node(state: AgentState) -> AgentState:
        """Thought 阶段：分析上下文并决定下一步"""
        response = llm.invoke(state["messages"])
        return {
            "messages": state["messages"] + [response],
            "iteration": state["iteration"] + 1,
        }

    def should_continue(state: AgentState) -> Literal["tools", "end"]:
        """路由：有 tool_calls 则继续，否则结束"""
        if state["iteration"] >= max_iterations:
            return "end"
        last = state["messages"][-1]
        if getattr(last, "tool_calls", None):
            return "tools"
        return "end"

    workflow = StateGraph(AgentState)
    workflow.add_node("reasoning", reasoning_node)
    workflow.add_node("tools", tool_node)
    workflow.set_entry_point("reasoning")
    workflow.add_conditional_edges(
        "reasoning", should_continue, {"tools": "tools", "end": END}
    )
    workflow.add_edge("tools", "reasoning")  # 观察后回到推理，形成闭环
    return workflow.compile()

from typing import TypedDict, List, Optional, Literal
from enum import Enum
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
import json


class StepStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"


class PlanExecuteState(TypedDict):
    task: str                    # 原始任务
    plan: List[dict]             # 执行计划列表
    current_step: int            # 当前步骤索引
    results: List[str]           # 各步骤执行结果
    final_result: Optional[str]  # 最终综合结果


def create_plan_execute_agent(tools: list):
    planner_llm = ChatOpenAI(model="gpt-4", temperature=0)   # 规划需要稳定
    executor_llm = ChatOpenAI(model="gpt-4").bind_tools(tools)

    def planning_node(state: PlanExecuteState) -> PlanExecuteState:
        """规划阶段：生成 3~8 个可执行原子步骤"""
        prompt = f"""
        请为以下任务生成执行计划：
        任务：{state['task']}
        
        要求：
        1. 分解为 3-8 个可执行的原子步骤
        2. 明确步骤之间的依赖关系
        3. 以 JSON 数组输出，每个元素包含 id/description/dependencies
        """
        plan = json.loads(planner_llm.invoke([HumanMessage(content=prompt)]).content)
        return {**state, "plan": plan, "current_step": 0, "results": []}

    def execution_node(state: PlanExecuteState) -> PlanExecuteState:
        """执行阶段：执行当前步骤，结果追加到 results"""
        if state["current_step"] >= len(state["plan"]):
            return state
        step = state["plan"][state["current_step"]]
        prev = "\n".join(f"步骤{i+1}: {r}" for i, r in enumerate(state["results"]))
        prompt = f"步骤描述：{step['description']}\n\n之前结果：\n{prev}"
        resp = executor_llm.invoke([HumanMessage(content=prompt)])
        return {
            **state,
            "results": state["results"] + [resp.content],
            "current_step": state["current_step"] + 1,
        }

    def replan_node(state: PlanExecuteState) -> PlanExecuteState:
        """重规划：根据进度判断是否需要调整剩余计划"""
        prompt = f"""
        原始任务：{state['task']}
        已完成：{state['results']}
        
        请评估是否需要调整剩余计划。需要则输出新计划 JSON，不需要则输出 "NO_REPLAN"。
        """
        resp = planner_llm.invoke([HumanMessage(content=prompt)])
        if "NO_REPLAN" not in resp.content:
            new_plan = json.loads(resp.content)
            return {**state, "plan": new_plan, "current_step": 0}
        return state

    def synthesize_node(state: PlanExecuteState) -> PlanExecuteState:
        """综合阶段：把所有中间结果整合为最终答案"""
        prev = "\n".join(f"步骤{i+1}: {r}" for i, r in enumerate(state["results"]))
        prompt = f"任务：{state['task']}\n\n执行结果：\n{prev}\n\n请综合给出最终答案。"
        resp = planner_llm.invoke([HumanMessage(content=prompt)])
        return {**state, "final_result": resp.content}

    def should_continue(state) -> Literal["execute", "replan", "synthesize"]:
        if state["current_step"] >= len(state["plan"]):
            return "synthesize"
        if state["current_step"] > 0 and state["current_step"] % 2 == 0:
            return "replan"  # 每执行 2 步做一次中间校验
        return "execute"

    wf = StateGraph(PlanExecuteState)
    wf.add_node("plan", planning_node)
    wf.add_node("execute", execution_node)
    wf.add_node("replan", replan_node)
    wf.add_node("synthesize", synthesize_node)
    wf.set_entry_point("plan")
    wf.add_edge("plan", "execute")
    wf.add_conditional_edges(
        "execute", should_continue,
        {"execute": "execute", "replan": "replan", "synthesize": "synthesize"},
    )
    wf.add_edge("replan", "execute")
    wf.add_edge("synthesize", END)
    return wf.compile()

from typing import TypedDict, List, Optional, Literal
from datetime import datetime
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
import json


class RalphState(TypedDict):
    task: str
    episode_count: int
    max_episodes: int
    reasoning_result: Optional[dict]
    action_result: Optional[dict]
    learning_result: Optional[dict]
    planning_result: Optional[dict]
    task_completed: bool
    final_answer: Optional[str]
    knowledge_base: List[dict]       # 语义记忆
    experience_memory: List[dict]    # 情景记忆


def create_ralph_agent(tools: list, max_episodes: int = 10):
    llm = ChatOpenAI(model="gpt-4", temperature=0)

    def reasoning_node(state: RalphState) -> RalphState:
        # 检索相关知识（可替换为向量检索）
        kb = state["knowledge_base"][-5:]
        exp = state["experience_memory"][-3:]
        prompt = f"""
        任务：{state['task']}
        相关知识：{json.dumps(kb, ensure_ascii=False)}
        类似经验：{json.dumps(exp, ensure_ascii=False)}
        
        输出 JSON：
        {{
            "state_assessment": "当前状态评估",
            "problem_analysis": "问题核心分析",
            "hypotheses": ["假设1", "假设2"],
            "recommended_action": "推荐的行动策略"
        }}
        """
        result = json.loads(llm.invoke([HumanMessage(content=prompt)]).content)
        return {**state, "reasoning_result": result}

    def acting_node(state: RalphState) -> RalphState:
        prompt = f"""
        任务：{state['task']}
        推理：{json.dumps(state['reasoning_result'], ensure_ascii=False)}
        
        请执行推荐动作，输出 JSON：
        {{"action_taken": "...", "parameters": {{}}, "outcome": "...",
          "success": true, "context": "..."}}
        """
        result = json.loads(llm.invoke([HumanMessage(content=prompt)]).content)
        return {**state, "action_result": result}

    def learning_node(state: RalphState) -> RalphState:
        """从本次行动中提取知识、更新 KB"""
        ar = state["action_result"]
        kb = state["knowledge_base"].copy()
        memory = state["experience_memory"].copy()

        experience = {
            "context": ar.get("context", ""),
            "action": ar.get("action_taken", ""),
            "outcome": ar.get("outcome", ""),
            "success": ar.get("success", False),
            "timestamp": datetime.now().isoformat(),
        }
        memory.append(experience)

        prompt = f"""
        根据经验提取可学习的模式：
        {json.dumps(experience, ensure_ascii=False)}
        
        输出 JSON：
        {{"patterns": [...], "new_knowledge": {{...}}, "strategy_updates": [...]}}
        """
        learning = json.loads(llm.invoke([HumanMessage(content=prompt)]).content)
        if learning.get("new_knowledge"):
            kb.append(learning["new_knowledge"])

        return {
            **state,
            "learning_result": learning,
            "knowledge_base": kb,
            "experience_memory": memory,
        }

    def planning_node(state: RalphState) -> RalphState:
        prompt = f"""
        任务：{state['task']}
        学习结果：{json.dumps(state['learning_result'], ensure_ascii=False)}
        
        输出 JSON：
        {{"goal": "...", "strategy": "...", "expected_outcome": "...",
          "task_completed": false}}
        """
        planning = json.loads(llm.invoke([HumanMessage(content=prompt)]).content)
        return {
            **state,
            "planning_result": planning,
            "task_completed": planning.get("task_completed", False),
        }

    def finalize_node(state: RalphState) -> RalphState:
        prompt = f"任务：{state['task']}\n完整历史：{json.dumps(state['experience_memory'], ensure_ascii=False)}\n\n综合所有经验，给出最终答案。"
        ans = llm.invoke([HumanMessage(content=prompt)]).content
        return {**state, "final_answer": ans}

    def should_continue(state) -> Literal["reasoning", "finalize"]:
        if state["task_completed"] or state["episode_count"] >= state["max_episodes"]:
            return "finalize"
        return "reasoning"

    def increment(state): return {**state, "episode_count": state["episode_count"] + 1}

    wf = StateGraph(RalphState)
    wf.add_node("reasoning", reasoning_node)
    wf.add_node("acting", acting_node)
    wf.add_node("learning", learning_node)
    wf.add_node("planning", planning_node)
    wf.add_node("increment", increment)
    wf.add_node("finalize", finalize_node)
    wf.set_entry_point("reasoning")
    wf.add_edge("reasoning", "acting")
    wf.add_edge("acting", "learning")
    wf.add_edge("learning", "planning")
    wf.add_edge("planning", "increment")
    wf.add_conditional_edges(
        "increment", should_continue,
        {"reasoning": "reasoning", "finalize": "finalize"},
    )
    wf.add_edge("finalize", END)
    return wf.compile()

from typing import TypedDict, List, Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage


class CoTState(TypedDict):
    problem: str
    steps: List[str]
    current_step: int
    final_answer: str


def create_cot_agent(max_steps: int = 5):
    llm = ChatOpenAI(model="gpt-4", temperature=0)

    # 预定义的推理脚手架（Step Scaffolding）
    STEP_PROMPTS = [
        "分析问题的关键信息和约束条件",
        "确定解决问题的方法和策略",
        "执行具体的计算或推理过程",
        "验证结果的合理性和完整性",
        "总结并给出最终答案",
    ]

    def analyze_step(state: CoTState) -> CoTState:
        cur = state["current_step"]
        guide = STEP_PROMPTS[cur] if cur < len(STEP_PROMPTS) else "继续推理"
        history = "\n".join(f"步骤{i+1}: {s}" for i, s in enumerate(state["steps"]))
        prompt = f"""
        问题：{state['problem']}
        历史推理：
        {history}
        
        当前任务：{guide}
        请给出该步详细推理。
        """
        resp = llm.invoke([HumanMessage(content=prompt)])
        return {**state, "steps": state["steps"] + [resp.content],
                "current_step": cur + 1}

    def finalize_answer(state: CoTState) -> CoTState:
        history = "\n".join(f"步骤{i+1}: {s}" for i, s in enumerate(state["steps"]))
        prompt = f"问题：{state['problem']}\n完整推理：\n{history}\n\n请给出简洁最终答案。"
        ans = llm.invoke([HumanMessage(content=prompt)]).content
        return {**state, "final_answer": ans}

    def should_continue(state) -> Literal["analyze", "finalize"]:
        return "finalize" if state["current_step"] >= max_steps else "analyze"

    wf = StateGraph(CoTState)
    wf.add_node("analyze", analyze_step)
    wf.add_node("finalize", finalize_answer)
    wf.set_entry_point("analyze")
    wf.add_conditional_edges(
        "analyze", should_continue, {"analyze": "analyze", "finalize": "finalize"}
    )
    wf.add_edge("finalize", END)
    return wf.compile()

from typing import TypedDict, Literal, Optional
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
import json


class MultiAgentState(TypedDict):
    task: str
    research_result: Optional[str]
    analysis_result: Optional[str]
    draft: Optional[str]
    review_result: Optional[dict]
    final_result: Optional[str]
    revision_count: int


def create_multi_agent_system():
    # 每个角色使用不同的温度策略，匹配其认知特征
    research_llm = ChatOpenAI(model="gpt-4", temperature=0.7)  # 探索性
    analysis_llm = ChatOpenAI(model="gpt-4", temperature=0)    # 严谨性
    writing_llm = ChatOpenAI(model="gpt-4", temperature=0.5)   # 平衡
    review_llm = ChatOpenAI(model="gpt-4", temperature=0)      # 严格

    def research_agent(state):
        prompt = f"你是专业研究员。请针对以下任务调研：{state['task']}\n输出：背景/概念/数据/参考资源。"
        return {**state, "research_result": research_llm.invoke([HumanMessage(content=prompt)]).content}

    def analysis_agent(state):
        prompt = f"你是分析师。基于研究资料进行分析。\n任务：{state['task']}\n资料：{state['research_result']}"
        return {**state, "analysis_result": analysis_llm.invoke([HumanMessage(content=prompt)]).content}

    def writing_agent(state):
        review = state.get("review_result")
        if review and not review.get("approved"):
            prompt = f"根据审核意见修改：\n原文：{state['draft']}\n意见：{review.get('feedback')}"
        else:
            prompt = f"基于资料撰写：\n任务：{state['task']}\n研究：{state['research_result']}\n分析：{state['analysis_result']}"
        return {**state, "draft": writing_llm.invoke([HumanMessage(content=prompt)]).content}

    def review_agent(state):
        prompt = f"""
        审核内容：
        任务：{state['task']}
        文稿：{state['draft']}
        输出 JSON：{{"approved": true|false, "score": 0-10, "feedback": "..."}}
        """
        result = json.loads(review_llm.invoke([HumanMessage(content=prompt)]).content)
        return {**state, "review_result": result,
                "revision_count": state["revision_count"] + 1}

    def should_revise(state) -> Literal["writing", "finalize"]:
        if state["revision_count"] >= 3:
            return "finalize"
        return "finalize" if state["review_result"].get("approved") else "writing"

    wf = StateGraph(MultiAgentState)
    wf.add_node("research", research_agent)
    wf.add_node("analysis", analysis_agent)
    wf.add_node("writing", writing_agent)
    wf.add_node("review", review_agent)
    wf.add_node("finalize", lambda s: {**s, "final_result": s["draft"]})
    wf.set_entry_point("research")
    wf.add_edge("research", "analysis")
    wf.add_edge("analysis", "writing")
    wf.add_edge("writing", "review")
    wf.add_conditional_edges(
        "review", should_revise,
        {"writing": "writing", "finalize": "finalize"},
    )
    wf.add_edge("finalize", END)
    return wf.compile()

from typing import TypedDict, List, Optional, Dict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
import json


class HierarchicalState(TypedDict):
    task: str
    subtasks: Dict[str, List[str]]         # supervisor -> [子任务]
    worker_results: Dict[str, str]
    supervisor_summaries: Dict[str, str]
    final_result: Optional[str]


def create_hierarchical_agent():
    # 层级化模型配置：上层精密，下层廉价
    master_llm = ChatOpenAI(model="gpt-4", temperature=0)
    supervisor_llm = ChatOpenAI(model="gpt-4", temperature=0)
    worker_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.5)

    def master_node(state):
        prompt = f"""
        将任务分解给三个管理者（JSON 输出）：
        - task_supervisor（任务执行）
        - resource_supervisor（资源管理）
        - quality_supervisor（质量保证）
        
        任务：{state['task']}
        """
        subtasks = json.loads(master_llm.invoke([HumanMessage(content=prompt)]).content)
        return {**state, "subtasks": subtasks}

    def make_supervisor(name: str):
        def node(state):
            subtasks = state["subtasks"].get(name, [])
            results = state["worker_results"].copy()
            for i, sub in enumerate(subtasks):
                resp = worker_llm.invoke([HumanMessage(content=f"执行子任务：{sub}")])
                results[f"{name}_worker_{i}"] = resp.content
            own = {k: v for k, v in results.items() if k.startswith(name)}
            summary = supervisor_llm.invoke(
                [HumanMessage(content=f"{name} 请汇总：{json.dumps(own, ensure_ascii=False)}")]
            ).content
            summaries = state["supervisor_summaries"].copy()
            summaries[name] = summary
            return {**state, "worker_results": results, "supervisor_summaries": summaries}
        return node

    def aggregation_node(state):
        s = state["supervisor_summaries"]
        prompt = f"""
        主控综合汇报：
        任务：{state['task']}
        任务管理：{s.get('task_supervisor', '')}
        资源管理：{s.get('resource_supervisor', '')}
        质量管理：{s.get('quality_supervisor', '')}
        """
        return {**state, "final_result": master_llm.invoke([HumanMessage(content=prompt)]).content}

    wf = StateGraph(HierarchicalState)
    wf.add_node("master", master_node)
    wf.add_node("task_supervisor", make_supervisor("task_supervisor"))
    wf.add_node("resource_supervisor", make_supervisor("resource_supervisor"))
    wf.add_node("quality_supervisor", make_supervisor("quality_supervisor"))
    wf.add_node("aggregation", aggregation_node)
    wf.set_entry_point("master")
    wf.add_edge("master", "task_supervisor")
    wf.add_edge("task_supervisor", "resource_supervisor")
    wf.add_edge("resource_supervisor", "quality_supervisor")
    wf.add_edge("quality_supervisor", "aggregation")
    wf.add_edge("aggregation", END)
    return wf.compile()

from typing import TypedDict, List, Literal, Optional
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
import json


class ReflectionState(TypedDict):
    task: str
    current_solution: Optional[str]
    reflection: Optional[dict]
    improvements: List[str]
    iteration: int
    max_iterations: int
    final_result: Optional[str]
    quality_score: float


def create_reflection_agent(max_iterations: int = 3, quality_threshold: float = 8.0):
    actor_llm = ChatOpenAI(model="gpt-4", temperature=0.7)
    reflector_llm = ChatOpenAI(model="gpt-4", temperature=0)

    def action_node(state):
        if state.get("current_solution") and state["improvements"]:
            prompt = f"""
            根据改进建议优化方案：
            任务：{state['task']}
            当前：{state['current_solution']}
            建议：{chr(10).join(state['improvements'])}
            """
        else:
            prompt = f"请完成任务：{state['task']}\n输出完整方案。"
        return {**state, "current_solution": actor_llm.invoke([HumanMessage(content=prompt)]).content}

    def reflect_node(state):
        prompt = f"""
        评估方案（JSON 输出）：
        任务：{state['task']}
        方案：{state['current_solution']}
        
        {{"effectiveness": {{"score": 0-10, "analysis": "..."}},
          "methodology": {{"score": 0-10, "analysis": "..."}},
          "goal_alignment": {{"score": 0-10, "analysis": "..."}},
          "overall_score": 0-10,
          "strengths": [...],
          "weaknesses": [...]}}
        """
        refl = json.loads(reflector_llm.invoke([HumanMessage(content=prompt)]).content)
        return {**state, "reflection": refl, "quality_score": refl.get("overall_score", 0)}

    def improve_node(state):
        prompt = f"基于评估制定 3-5 条具体改进建议：\n{json.dumps(state['reflection'], ensure_ascii=False)}"
        resp = reflector_llm.invoke([HumanMessage(content=prompt)])
        imp = [l.strip() for l in resp.content.split('\n') if l.strip() and l.strip()[0].isdigit()]
        return {**state, "improvements": imp}

    def plan_node(state):
        return {**state, "iteration": state["iteration"] + 1}

    def should_continue(state) -> Literal["action", "finalize"]:
        if state["quality_score"] >= quality_threshold: return "finalize"
        if state["iteration"] >= state["max_iterations"]: return "finalize"
        return "action"

    wf = StateGraph(ReflectionState)
    wf.add_node("action", action_node)
    wf.add_node("reflect", reflect_node)
    wf.add_node("improve", improve_node)
    wf.add_node("plan", plan_node)
    wf.add_node("finalize", lambda s: {**s, "final_result": s["current_solution"]})
    wf.set_entry_point("action")
    wf.add_edge("action", "reflect")
    wf.add_edge("reflect", "improve")
    wf.add_edge("improve", "plan")
    wf.add_conditional_edges(
        "plan", should_continue, {"action": "action", "finalize": "finalize"}
    )
    wf.add_edge("finalize", END)
    return wf.compile()