在构建一个大规模 AI 系统时,我们其实就是在把不同的“智能体设计模式(agentic design patterns)”组合起来。不管系统多复杂都可以拆解成有限的几种"设计模式"。这些模式各有各的用法——有的专门负责思考优化,有的处理工具调用,有的管多智能体协作。

主要就这么几类:多智能体系统让不同角色分工协作;集成决策让多个智能体投票选最优解;思维树(ToT)探索多条推理路径再择优;反思式架构能自我审视和改进;ReAct循环在思考与行动间迭代。

本文将逐一深入解析这17种不同的智能体架构,不仅阐述其核心原理和设计思想,还将通过完整的代码实现来演示其工作机制,并用真实数据验证每种架构的实际效果和适用场景。

环境配置

开始之前得把工具链搞定。现在的RAG或智能体系统,LangChain、LangGraph和LangSmith基本成了标配——前者提供基础组件,LangGraph负责编排流程,LangSmith可以进行调试和监控。

导入

 import os  
from typing import List, Dict, Any, Optional, Annotated, TypedDict  
from dotenv import load_dotenv  # 从 .env 文件加载环境变量  

# 用于数据建模/校验的 Pydantic  
from pydantic import BaseModel, Field  

# LangChain & LangGraph 组件  
from langchain_nebius import ChatNebius  # Nebius LLM 封装器  
from langchain_tavily import TavilySearch  # Tavily 搜索工具集成  
from langchain_core.prompts import ChatPromptTemplate  # 用于组织提示词  
from langgraph.graph import StateGraph, END  # 构建状态机图  
from langgraph.prebuilt import ToolNode, tools_condition  # 预置的节点与条件  

# 便于更好地输出结果  
from rich.console import Console  # 终端样式化输出  
 from rich.markdown import Markdown  # 在终端渲染 Markdown

为了作为演示,我还接了Tavily API做实时搜索,避免智能体被训练数据的时效性限制住。每月1000次调用的免费额度已经够用了了。

环境变量这样配:

# Nebius LLM 的 API key(用于 ChatNebius)  
NEBIUS_API_KEY="your_nebius_api_key_here"  

# LangSmith(LangChain 观测/遥测平台)的 API key  
LANGCHAIN_API_KEY="your_langsmith_api_key_here"  

# Tavily 搜索工具(用于 TavilySearch 集成)的 API key  
TAVILY_API_KEY="your_tavily_api_key_here"

加载和验证:

load_dotenv()  # 加载环境变量  

# 启用 LangSmith 跟踪以便监控/调试  
os.environ["LANGCHAIN_TRACING_V2"] = "true"  
os.environ["LANGCHAIN_PROJECT"] = "Implementing 17 Agentic Architectures"  # 用于分组轨迹的项目名  

# 校验所需的 API key 是否齐全  
for key in ["NEBIUS_API_KEY", "LANGCHAIN_API_KEY", "TAVILY_API_KEY"]:  
    if not os.environ.get(key):  # 若环境变量中未找到  
        print(f"{key} not found. Please create a .env file and set it.")

反思架构(Reflection)

反思可能是智能体工作流里最基础也最实用的模式。核心思路就是让智能体"退一步看自己的输出,然后想办法改进"。

这个模式特别适合那种对结果质量要求很高的场景,比如生成复杂代码、写技术文档什么的。如果只给个草稿通常质量都不好,所以需要这个模式。

流程非常简单:

生成(Generate):基于用户需求先出个初版

批判(Critique):切换视角,从代码审查员的角度挑毛病

精炼(Refine):基于批评意见,输出改进版本

实现上用Pydantic模型来约束LLM的输出格式,这样多步骤之间的数据传递会更稳定:

class DraftCode(BaseModel):  
    """Schema for the initial code draft generated by the agent."""  
    code: str = Field(description="The Python code generated to solve the user's request.")  # 原始草稿  
    explanation: str = Field(description="A brief explanation of how the code works.")  # 推理说明  

class Critique(BaseModel):  
    """Schema for the self-critique of the generated code."""  
    has_errors: bool = Field(description="Does the code have any potential bugs or logical errors?")  # 错误排查  
    is_efficient: bool = Field(description="Is the code written in an efficient and optimal way?")  # 性能/最佳实践检查  
    suggested_improvements: List[str] = Field(description="Specific, actionable suggestions for improving the code.")  # 具体改进建议  
    critique_summary: str = Field(description="A summary of the critique.")  # 评审概述  

class RefinedCode(BaseModel):  
    """Schema for the final, refined code after incorporating the critique."""  
    refined_code: str = Field(description="The final, improved Python code.")  # 打磨后的版本  
    refinement_summary: str = Field(description="A summary of the changes made based on the critique.")  # 修改说明

Critique这个模型设计得比较巧妙,它强制智能体必须从错误检查、效率评估等具体维度来审视代码,而不是泛泛地说"看起来还行",这样保证了检查的质量。

生成器节点就是标准的prompt调用:

def generator_node(state):  
    """Generates the initial draft of the code."""  
    console.print("--- 1. Generating Initial Draft ---")  
    # 初始化 LLM,并让其输出结构化 DraftCode 对象  
    generator_llm = llm.with_structured_output(DraftCode)  

    prompt = f"""You are an expert Python programmer. Write a Python function to solve the following request.  
    Provide a simple, clear implementation and an explanation.  

    Request: {state['user_request']}  
    """  

    draft = generator_llm.invoke(prompt)  
    return {"draft": draft.model_dump()}

批评节点是这个架构的核心,它让智能体作为"资深开发者"来审视自己刚写的代码:

def critic_node(state):  
    """Critiques the generated code for errors and inefficiencies."""  
    console.print("--- 2. Critiquing Draft ---")  
    # 初始化 LLM,并让其输出结构化 Critique 对象  
    critic_llm = llm.with_structured_output(Critique)  

    code_to_critique = state['draft']['code']  

    prompt = f"""You are an expert code reviewer and senior Python developer. Your task is to perform a thorough critique of the following code.  

    Analyze the code for:  
    1.  **Bugs and Errors:** Are there any potential runtime errors, logical flaws, or edge cases that are not handled?  
    2.  **Efficiency and Best Practices:** Is this the most efficient way to solve the problem? Does it follow standard Python conventions (PEP 8)?  

    Provide a structured critique with specific, actionable suggestions.  

    Code to Review:  
{code_to_critique}  
``` 
"""  

critique = critic_llm.invoke(prompt)  
return {"critique": critique.model_dump()}

精炼节点负责把批评意见落地成实际的代码改动:

def refiner_node(state):

"""Refines the code based on the critique."""  
console.print("--- 3. Refining Code ---")  
# 初始化 LLM,并让其输出结构化 RefinedCode 对象  
refiner_llm = llm.with_structured_output(RefinedCode)  

draft_code = state['draft']['code']  
critique_suggestions = json.dumps(state['critique'], indent=2)  

prompt = f"""You are an expert Python programmer tasked with refining a piece of code based on a critique.  

Your goal is to rewrite the original code, implementing all the suggested improvements from the critique.  

**Original Code:**  
```python  
{draft_code}  
``` 

**Critique and Suggestions:**  
{critique_suggestions}  

Please provide the final, refined code and a summary of the changes you made.  
"""  

refined_code = refiner_llm.invoke(prompt)  
return {"refined_code": refined_code.model_dump()}

用LangGraph把这三个节点串成工作流:

class ReflectionState(TypedDict):

"""Represents the state of our reflection graph."""  
user_request: str  
draft: Optional[dict]  
critique: Optional[dict]  
refined_code: Optional[dict]  

初始化 state graph

graph_builder = StateGraph(ReflectionState)

添加节点

graph_builder.add_node("generator", generator_node)
graph_builder.add_node("critic", critic_node)
graph_builder.add_node("refiner", refiner_node)

线性流程

graph_builder.set_entry_point("generator")
graph_builder.add_edge("generator", "critic")
graph_builder.add_edge("critic", "refiner")
graph_builder.add_edge("refiner", END)

编译为可运行应用

reflection_app = graph_builder.compile()

![](http://images.overfit.cn/upload/20250926/43d4a0b89dc446208570eabe2b38ec62.webp)

拿经典的斐波那契数列来测试效果。这个问题很适合:因为简单递归容易写但效率极差,改进空间明显。

user_request = "Write a Python function to find the nth Fibonacci number."
initial_input = {"user_request": user_request}

console.print(f"[bold cyan]🚀 Kicking off Reflection workflow for request:[/bold cyan] '{user_request}'\n")

流式运行并获取最终状态

final_state = None
for state_update in reflection_app.stream(initial_input, stream_mode="values"):

final_state = state_update  

console.print("\n[bold green]✅ Reflection workflow complete![/bold green]")


结果对比:

--- ### Initial Draft ---
Explanation: This function uses a recursive approach to calculate the nth Fibonacci number... This approach is not efficient for large values of n due to the repeated calculations...

1 def fibonacci(n):
2 if n <= 0:
3 return 0
4 elif n == 1:
5 return 1
6 else:
7 return fibonacci(n-1) + fibonacci(n-2)

--- ### Critique ---
Summary: The function has potential bugs and inefficiencies. It should be revised to handle negative inputs and improve its time complexity.
Improvements Suggested:

  • The function does not handle negative numbers correctly.
  • The function has a high time complexity due to the repeated calculations. Consider using dynamic programming or memoization.
  • The function does not follow PEP 8 conventions...

--- ### Final Refined Code ---
Refinement Summary: The original code has been revised to handle negative inputs, improve its time complexity, and follow PEP 8 conventions.

1 def fibonacci(n):
2 """Calculates the nth Fibonacci number."""
3 if n < 0:
4 raise ValueError("n must be a non-negative integer")
5 elif n == 0:
6 return 0
7 elif n == 1:
8 return 1
9 else:
10 fib = [0, 1]
11 for i in range(2, n + 1):
12 fib.append(fib[i-1] + fib[i-2])
13 return fib[n]


初版是标准的递归实现,复杂度O(2^n),完全不实用。改进版改成动态规划,复杂度降到O(n),还加了输入验证和文档字符串。

为了量化改进效果,可以再加个评分机制:

class CodeEvaluation(BaseModel):

"""Schema for evaluating a piece of code."""  
correctness_score: int = Field(description="Score from 1-10 on whether the code is logically correct.")  
efficiency_score: int = Field(description="Score from 1-10 on the code's algorithmic efficiency.")  
style_score: int = Field(description="Score from 1-10 on code style and readability (PEP 8). ")  
justification: str = Field(description="A brief justification for the scores.")  

def evaluate_code(code_to_evaluate: str):

prompt = f"""You are an expert judge of Python code. Evaluate the following function on a scale of 1-10 for correctness, efficiency, and style. Provide a brief justification.  

Code:  
```python  
{code_to_evaluate}  
```  
"""  
return judge_llm.invoke(prompt)

评分结果一目了然:

--- Evaluating Initial Draft ---
{

'correctness_score': 2,  
'efficiency_score': 4,  
'style_score': 2,  
'justification': 'The function has a time complexity of O(2^n)...'  

}

--- Evaluating Refined Code ---
{

'correctness_score': 8,  
'efficiency_score': 6,  
'style_score': 9,  
'justification': 'The code is correct... it has a time complexity of O(n)...'  

}


数据说明反思机制确实有效果,不是简单的表面修改。

## 工具调用架构(Tool Using)

没有外部工具的话,LLM只能依赖训练时的知识,无法访问实时数据或执行具体操作。工具调用架构就是为了解决这个问题。在生产级AI系统中,这几乎是必需的——不管是客服查询订单、金融系统获取股价,还是代码生成器执行测试都离不开工具集成。
![](http://images.overfit.cn/upload/20250926/c359bc999453445ba55b5c358fd75fcd.webp)

流程如下:

接收查询:智能体收到用户请求  

决策分析:判断是否需要调用工具来获取信息

工具调用:按正确格式执行工具调用

结果处理:获取工具输出并整合

生成回复:结合工具结果给出最终答案

我们需要先配置一个搜索工具。这里用TavilySearch,并且工具描述要写清楚,LLM会根据这个自然语言描述来判断什么时候该用:

初始化工具。设置返回结果数量以保持上下文简洁。

search_tool = TavilySearchResults(max_results=2)

为工具提供明确的名称与描述,便于智能体理解

search_tool.name = "web_search"
search_tool.description = "A tool that can be used to search the internet for up-to-date information on any topic, including news, events, and current affairs."

tools = [search_tool]


智能体的状态结构比反思架构简单,就是维护消息历史:

class AgentState(TypedDict):

messages: Annotated[list[AnyMessage], add_messages]

不过最关键步骤是让LLM"感知"到工具的存在。用bind_tools把工具信息注入到系统提示中:

llm = ChatNebius(model="meta-llama/Meta-Llama-3.1-8B-Instruct", temperature=0)

绑定工具,使 LLM 具备"工具感知"

llm_with_tools = llm.bind_tools(tools)


LangGraph工作流需要两个核心节点:agent_node负责推理决策,tool_node负责执行工具:

def agent_node(state: AgentState):

"""主要节点:调用 LLM 决定下一步动作。"""  
console.print("--- AGENT: Thinking... ---")  
response = llm_with_tools.invoke(state["messages"])  
return {"messages": [response]}

ToolNode 是 LangGraph 的预置节点,用于执行工具

tool_node = ToolNode(tools)


路由逻辑比较简单:检查最后一条消息是否包含tool_calls,有的话就去执行工具,否则就结束:
if hasattr(last_message, 'tool_calls') and last_message.tool_calls:  
    console.print("--- ROUTER: 检测到工具调用请求 ---")  
    return "call_tool"  
else:  
    console.print("--- ROUTER: 准备输出最终答案 ---")  
    return "__end__"

整个的工作流就是把这些节点用LangGraph串起来。条件边的设计可以让智能体能在推理和工具调用间形成循环:

graph_builder = StateGraph(AgentState)

Logo

更多推荐