AI Agent六大设计模式

# ReAct 核心循环
for iteration in range(max_iterations):
    # 1. Thought - 分析当前状态，决定下一步行动
    thought = llm.invoke(f"分析: {context}\n决定下一步:")
    
    # 2. Action - 调用工具执行操作
    if "Action:" in thought:
        action = parse_action(thought)
        observation = execute_tool(action)
        context += f"\n行动: {action}\n观察: {observation}"
    
    # 3. Observation - 获取结果作为下一轮输入
    # 循环持续，直到任务完成或达到最大步数
    
    # 4. 判断是否结束
    if "Answer:" in thought:
        return extract_answer(thought)

import { StateGraph, END, Annotation } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
import { ToolNode } from "@langchain/langgraph/prebuilt";

// 定义网络搜索工具
const webSearch = tool(async ({ query }) => {
    return `搜索结果: ${query}`;
}, {
    name: "web_search",
    description: "搜索网络",
    schema: z.object({ query: z.string() })
});

// 定义数据库查询工具（安全限制）
const dbQuery = tool(async ({ sql }) => {
    // 安全：仅允许SELECT语句
    if (!ALLOWED_SQL_PATTERN.test(sql.trim())) {
        return "Error: 仅允许简单SELECT查询";
    }
    return `查询结果: ${sql}`;
}, {
    name: "db_query",
    description: "查询数据库(仅SELECT)",
    schema: z.object({ sql: z.string().max(500) })
});

// 绑定工具到LLM
const tools = [webSearch, dbQuery];
const model = new ChatOpenAI({ model: "gpt-4o" })
    .bindTools(tools);

// 创建工具节点
const toolNode = new ToolNode(tools);

// 判断是否使用工具
function shouldUseTool(state) {
    const last = state.messages[state.messages.length - 1];
    return last?.tool_calls?.length > 0 ? "tools" : END;
}

// 构建状态图
const graph = new StateGraph(AgentAnnotation)
    .addNode("agent", agent)
    .addNode("tools", toolNode)
    .setEntryPoint("agent")
    .addConditionalEdges("agent", shouldUseTool, 
        { tools: "tools", [END]: END })
    .addEdge("tools", "agent")
    .compile();

# Reflection 核心循环
# Generate → Critique → Refine

class ReflectionState:
    task: str          # 任务描述
    draft: str         # 当前草稿
    critique: str     # 评审意见
    score: int         # 质量评分(1-10)
    iteration: int     # 当前迭代次数

def generate(state):
    # 生成器：基于反馈生成/改进内容
    feedback = f"Feedback: {state['critique']}" if state["critique"] else ""
    msg = llm.invoke(f"Write: {state['task']}{feedback}")
    return {"draft": msg.content, "iteration": state["iteration"] + 1}

def critique(state):
    # 评审器：评分并提供改进建议
    msg = llm.invoke(f"Score 1-10 and critique: {state['draft']}")
    score = extract_score(msg.content)
    return {"critique": msg.content, "score": score}

def should_continue(state):
    # 终止条件：分数达标或达到最大迭代
    if state["score"] >= 8 or state["iteration"] >= 3:
        return END
    return "generate"

class PlanState(TypedDict):
    objective: str              # 目标
    steps: List[dict]           # 步骤列表
    current_index: int           # 当前步骤索引
    final_result: str           # 最终结果

def planner(state):
    # 规划器：生成任务计划
    msg = llm.invoke(
        f"Break into 3-5 steps. Return ONLY numbered list.\n{state['objective']}"
    )
    # 解析步骤（只接受编号行）
    steps = [{"description": line, "status": "pending"} 
             for line in msg.content.strip().split("\n") 
             if re.match(r'^\d+[\.\)]', line)]
    return {"steps": steps, "current_index": 0}

def executor(state):
    # 执行器：执行当前步骤
    idx = state["current_index"]
    step = state["steps"][idx]
    # 累积之前的上下文
    context = "\n".join(s["result"] for s in 
                   state["steps"][:idx] if s["result"])
    msg = llm.invoke(f"Execute: {step['description']}\nContext: {context}")
    # 更新步骤状态
    steps = list(state["steps"])
    steps[idx] = {"status": "done", "result": msg.content}
    return {"steps": steps, "current_index": idx + 1}

def replanner(state):
    # 重规划器：失败时调整计划
    completed = [s for s in state["steps"] if s["status"] == "done"]
    msg = llm.invoke(f"Revise remaining steps:\n{formatted(completed)}")
    # ... 解析并更新步骤
    return new_state

def route_after_execute(state):
    if state["current_index"] >= len(state["steps"]):
        return "aggregator"  # 完成
    if state["steps"][state["current_index" - 1].get("status") == "failed":
        return "replanner"   # 失败，重规划
    return "executor"      # 继续执行

from crewai import Agent, Task, Crew, Process

# 定义专业Agent
researcher = Agent(
    role="资深研究员",
    goal="为给定的主题进行深入调研，发现准确可靠的信息",
    backstory="你是在信息检索和事实核查方面经验丰富的专家",
    verbose=True
)

analyst = Agent(
    role="数据分析师",
    goal="分析研究员提供的数据和信息，提炼洞见和趋势",
    backstory="你擅长从复杂信息中提取关键点并进行逻辑推理",
    verbose=True
)

writer = Agent(
    role="技术作家",
    goal="根据研究员和分析师的工作，撰写清晰、引人入胜的报告",
    backstory="你擅长将技术内容转化为易于理解的文章",
    verbose=True
)

# 定义任务
research_task = Task(
    description="调查人工智能在2026年的主要突破",
    expected_output="一份包含关键事实和来源的调研笔记",
    agent=researcher
)

analysis_task = Task(
    description="分析这些突破对医疗、金融、教育行业的潜在影响",
    expected_output="一份趋势分析和风险评估",
    agent=analyst,
    context=[research_task]  # 依赖researcher
)

write_task = Task(
    description="撰写一篇面向科技爱好者的博客文章",
    expected_output="一篇约500字的博客文章",
    agent=writer,
    context=[research_task, analysis_task]
)

# 组建团队
tech_crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, write_task],
    process=Process.sequential,  # 顺序执行
    verbose=2
)

# 启动
result = tech_crew.kickoff()
print(result)

class MultiAgentState(TypedDict):
    messages: Annotated[List[dict], operator.add]  # 累积消息
    turn_count: int
    consensus: bool

def researcher(state):
    history = format_messages(state["messages"])
    msg = llm.invoke(f"You are a Researcher. Provide analysis.\n{history}")
    return {"messages": [{"role": "researcher", "content": msg.content}]}

def critic(state):
    history = format_messages(state["messages"])
    msg = llm.invoke(f"You are a Critic. Challenge weaknesses.\n{history}")
    return {"messages": [{"role": "critic", "content": msg.content}]}

def supervisor(state):
    # 评估是否达成共识
    msg = llm.invoke(f"Have these agents reached consensus? Reply YES or NO.\n{history}")
    consensus = "YES" in msg.content.upper()
    return {"consensus": consensus}

def route(state):
    if state["consensus"] or state["turn_count"] >= 6:
        return END
    return "researcher"

# 构建图
graph = StateGraph(MultiAgentState)
graph.add_node("researcher", researcher)
graph.add_node("critic", critic)
graph.add_node("supervisor", supervisor)
graph.set_entry_point("researcher")
graph.add_edge("researcher", "critic")
graph.add_edge("critic", "supervisor")
graph.add_conditional_edges("supervisor", route, {END: END, "researcher": "researcher"})

# 静态中断：在预定节点前暂停
from langgraph.checkpoint.memory import MemorySaver

# 编译图时设置中断点
graph = builder.compile(
    interrupt_before=["assistant"],  # 在assistant节点前中断
    checkpointer=MemorySaver()
)

# 人工干预：更新状态
graph.update_state(
    config,
    {"messages": [HumanMessage(content="补充：查找量子计算专利")]}
)

# 动态中断：在节点内基于状态暂停
from langgraph.types import interrupt

def human_in_the_loop(state):
    # 从用户获取输入
    value = interrupt("您想修改输入还是继续？")
    return {"messages": value}

# 构建新图
new_builder = StateGraph(AgentState)
new_builder.add_node("human_in_the_loop", human_in_the_loop)
# ... 添加其他节点
new_builder.add_edge("guardian", "human_in_the_loop")
new_builder.add_edge("human_in_the_loop", "assistant")

# 模式选择决策树
if 任务需要外部数据:
    → 使用 Tool Use
    
elif 输出质量是瓶颈:
    → 使用 Reflection
    
elif 任务流程明确固定:
    → 使用 Planning
    
elif 需要多领域专家协作:
    → 使用 Multi-Agent
    
elif 高风险/高价值任务:
    → 使用 Human-in-the-Loop
    
else:
    → 使用 ReAct 作为起点

# 看宝AI 基础Agent架构
# 组合ReAct + Tool Use

# 工具定义
tools = [
    kanbao_search,      # 知识库搜索
    web_search,         # 网络搜索
    user_profile,       # 用户画像查询
    conversation_history # 历史对话检索
]

# ReAct Agent
react_agent = create_react_agent(llm, tools)
agent_executor = AgentExecutor(
    agent=react_agent,
    tools=tools,
    max_iterations=10,
    handle_parsing_errors=True
)

# 看宝AI 内容生成Agent
# 使用Reflection提升质量

class ContentAgent:
    def generate(self, prompt):
        # 生成 → 评审 → 改进循环
        draft = self.first_draft(prompt)
        for i in range(3):
            critique = self.critique(draft)
            if critique.score >= 8:
                break
            draft = self.improve(draft, critique.feedback)
        return draft

# 看宝AI 复杂任务处理
# 学习计划制定 → Planning模式

# Plan: [了解水平] → [推荐课程] → [制定计划] → [设置提醒]

class LearningPlanWorkflow:
    def create_plan(self, user_goal):
        # 1. 评估用户当前水平
        current_level = self.assess_user(user_goal)
        
        # 2. 推荐学习路径
        learning_path = self.recommend_path(current_level, user_goal)
        
        # 3. 分解为周计划
        weekly_plan = self.break_down(learning_path)
        
        # 4. 设置里程碑和提醒
        milestones = self.create_milestones(weekly_plan)
        
        return {
            "level": current_level,
            "path": learning_path,
            "plan": weekly_plan,
            "milestones": milestones
        }

# 看宝AI 多Agent协作

# 角色定义
researcher = Agent(role="学习研究员", goal="收集最新AI知识")
teacher = Agent(role="教学设计师", goal="设计学习内容")
coach = Agent(role="学习教练", goal="跟踪进度和激励")

# 协作流程
def kanbao_team(user_goal):
    # 研究员：调研最新资料
    resources = researcher.research(user_goal)
    
    # 教师：设计学习路径
    curriculum = teacher.design(resources, user_goal)
    
    # 教练：制定执行计划
    plan = coach.plan(curriculum)
    
    return plan

# 看宝AI 高风险操作

# 需要人工确认的场景
high_risk_operations = [
    "删除用户数据",
    "发送外部邮件",
    "修改系统设置",
    "生成公开内容"
]

def check_with_human(operation):
    if operation.type in high_risk_operations:
        # 触发人工确认
        confirm = interrupt(f"即将执行: {operation.description}")
        if not confirm:
            return "CANCELLED"
    return execute(operation)