# LangGraph子图架构：构建模块化AI工作流的完整指南

在构建复杂的AI工作流时，如何有效地管理和组织代码是一个常见的挑战。LangGraph的子图（Subgraph）架构提供了一种强大的解决方案，通过将复杂的工作流分解为更小、更易管理的模块，开发者可以实现更高的代码可重用性、可维护性和扩展性。本文将详细介绍LangGraph子图的基本概念、实现方法、实际应用以及最佳实践，帮助你掌握这一强大的工具。



## 一、子图架构概述

### 1.1 子图的基本概念

子图是LangGraph中的一个核心特性，它本质上是一个完整的图结构，可以作为更大图结构中的一个节点使用。子图允许我们将复杂的工作流分解为独立的模块，每个模块负责特定的功能。通过这种方式，我们可以实现模块化设计，使代码更易于理解和维护。

以下是一个简单的子图示例：

```python
from langgraph.graph import SubGraph, Graph

# 创建子图
class MarketingSubGraph(SubGraph):
    def __init__(self):
        super().__init__()
        
    def build(self) -> Graph:
        graph = Graph()
        # 定义子图内部结构
        return graph
```

在这个示例中，`MarketingSubGraph`是一个子图类，它继承自`SubGraph`，并通过`build`方法定义其内部结构。

### 1.2 子图的优势

子图架构带来了以下显著优势：

- **模块化**：将复杂逻辑封装在独立的子图中，使代码结构更清晰。
- **可重用性**：子图可以在不同的主图中重复使用，减少代码冗余。
- **可维护性**：独立的子图更容易进行测试和调试。
- **扩展性**：通过添加新的子图或修改现有子图，可以轻松扩展系统功能。

---

## 二、子图的实现方法

### 2.1 创建基本子图

以下是一个内容生成子图的示例，展示了如何定义子图的内部节点和边：

```python
class ContentGenerationSubGraph(SubGraph):
    def build(self) -> Graph:
        graph = Graph()
        
        # 添加内容生成节点
        graph.add_node("generate_content", self.generate_content)
        graph.add_node("review_content", self.review_content)
        
        # 添加边
        graph.add_edge("generate_content", "review_content")
        
        return graph
    
    def generate_content(self, state):
        # 内容生成逻辑
        return state
    
    def review_content(self, state):
        # 内容审查逻辑
        return state
```

在这个示例中，`ContentGenerationSubGraph`包含两个节点：`generate_content`和`review_content`，它们通过边连接，形成一个简单的工作流。

### 2.2 子图的状态管理

子图的状态管理是确保数据在不同节点间正确传递的关键。以下是一个分析子图的示例，展示了如何处理状态：

```python
class AnalyticsSubGraph(SubGraph):
    def build(self) -> Graph:
        graph = Graph()
        
        def process_analytics(state):
            # 确保状态包含必要的键
            if 'metrics' not in state:
                state['metrics'] = {}
            
            # 处理分析数据
            state['metrics']['engagement'] = calculate_engagement(state)
            return state
            
        graph.add_node("analytics", process_analytics)
        return graph
```

在这个示例中，`process_analytics`函数确保状态中包含`metrics`键，并计算参与度指标。

---

## 三、子图的组合与交互

### 3.1 在主图中使用子图

子图可以作为主图中的一个节点使用。以下示例展示了如何将内容生成子图和分析子图组合成一个主图：

```python
def create_marketing_workflow():
    main_graph = Graph()
    
    # 实例化子图
    content_graph = ContentGenerationSubGraph()
    analytics_graph = AnalyticsSubGraph()
    
    # 将子图添加到主图
    main_graph.add_node("content", content_graph)
    main_graph.add_node("analytics", analytics_graph)
    
    # 连接子图
    main_graph.add_edge("content", "analytics")
    
    return main_graph
```

在这个示例中，`create_marketing_workflow`函数创建了一个主图，并将两个子图连接起来，形成一个完整的工作流。

### 3.2 子图间的数据传递

子图间的数据传递是确保工作流正常运行的关键。以下示例展示了如何在子图中准备数据，供其他子图使用：

```python
class DataProcessingSubGraph(SubGraph):
    def build(self) -> Graph:
        graph = Graph()
        
        def prepare_data(state):
            # 准备数据供其他子图使用
            state['processed_data'] = {
                'content_type': state['raw_data']['type'],
                'metrics': state['raw_data']['metrics'],
                'timestamp': datetime.now()
            }
            return state
            
        graph.add_node("prepare", prepare_data)
        return graph
```

在这个示例中，`prepare_data`函数将原始数据处理为其他子图可用的格式。

---

## 四、实际案例：营销智能体实现

### 4.1 内容生成子图

以下是一个内容生成子图的完整实现，展示了如何使用LLM生成内容并优化：

```python
class ContentCreationSubGraph(SubGraph):
    def build(self) -> Graph:
        graph = Graph()
        
        def generate_content(state):
            prompt = f"""
            Target Audience: {state['audience']}
            Platform: {state['platform']}
            Campaign Goal: {state['goal']}
            """
            # 使用LLM生成内容
            content = generate_with_llm(prompt)
            state['generated_content'] = content
            return state
            
        def optimize_content(state):
            # 根据平台特点优化内容
            optimized = optimize_for_platform(
                state['generated_content'],
                state['platform']
            )
            state['final_content'] = optimized
            return state
            
        graph.add_node("generate", generate_content)
        graph.add_node("optimize", optimize_content)
        graph.add_edge("generate", "optimize")
        
        return graph
```

### 4.2 分析子图

以下是一个分析子图的实现，展示了如何计算内容表现并生成改进建议：

```python
class AnalyticsSubGraph(SubGraph):
    def build(self) -> Graph:
        graph = Graph()
        
        def analyze_performance(state):
            metrics = calculate_metrics(state['final_content'])
            state['analytics'] = {
                'engagement_score': metrics['engagement'],
                'reach_prediction': metrics['reach'],
                'conversion_estimate': metrics['conversion']
            }
            return state
            
        def generate_recommendations(state):
            recommendations = generate_improvements(
                state['analytics'],
                state['goal']
            )
            state['recommendations'] = recommendations
            return state
            
        graph.add_node("analyze", analyze_performance)
        graph.add_node("recommend", generate_recommendations)
        graph.add_edge("analyze", "recommend")
        
        return graph
```

### 4.3 主工作流程

以下是一个完整的营销智能体主工作流程的实现：

```python
def create_marketing_agent():
    main_graph = Graph()
    
    # 实例化子图
    content_graph = ContentCreationSubGraph()
    analytics_graph = AnalyticsSubGraph()
    
    # 添加配置节点
    def setup_campaign(state):
        # 初始化营销活动配置
        if 'config' not in state:
            state['config'] = {
                'audience': state.get('audience', 'general'),
                'platform': state.get('platform', 'twitter'),
                'goal': state.get('goal', 'engagement')
            }
        return state
    
    main_graph.add_node("setup", setup_campaign)
    main_graph.add_node("content", content_graph)
    main_graph.add_node("analytics", analytics_graph)
    
    # 构建工作流
    main_graph.add_edge("setup", "content")
    main_graph.add_edge("content", "analytics")
    
    return main_graph
```

---

## 五、最佳实践与注意事项

### 子图设计原则

1. **功能单一**：每个子图应专注于一个特定的功能，避免功能过于复杂。
2. **清晰的接口**：确保子图的输入输出接口清晰明确，便于与其他子图交互。
3. **状态管理**：合理设计状态传递机制，确保数据在不同子图间正确流动。

### 性能考虑

1. **减少数据传输**：避免子图间频繁的大数据传输，优化数据传递效率。
2. **异步处理**：对于耗时操作，考虑使用异步处理以提高性能。

### 错误处理

1. **内部错误处理**：在子图内部实现错误处理机制，确保系统稳定性。
2. **状态一致性**：在发生错误时，确保状态的一致性，避免数据丢失或损坏。

---



LangGraph的子图架构为构建复杂AI工作流提供了强大而灵活的解决方案。通过合理使用子图，我们可以将复杂的任务分解为可管理的模块，提高代码的可维护性和可重用性。本文通过详细的示例和最佳实践，帮助你掌握子图的设计与实现方法。希望你能在实际项目中运用这些知识，构建出功能强大、结构清晰的AI应用。

子图架构



# LangGraph实战指南：构建智能代理工作流的全解析

![LangGraph架构示意图](https://via.placeholder.com/800x400?text=LangGraph+核心组件交互流程)

## 一、LangGraph核心特性解析

### 1. 三大核心优势
- **循环工作流**：支持动态路径选择与迭代执行（如多轮对话场景）
```mermaid
graph LR
    A[用户输入] --> B{意图识别}
    B -->|咨询类| C[知识库查询]
    B -->|操作类| D[API调用]
    C --> E[生成回复]
    D --> E
    E --> F{是否完成}
    F -->|否| B
```

- **细粒度控制**：精确控制状态流转与执行流程
- **持久化机制**：内置Checkpointer实现状态快照与恢复

### 二、环境搭建指南

#### 1. 基础安装
```bash
# 安装核心组件
pip install -U langgraph langchain_anthropic

# 可选组件（监控与调试）
pip install langsmith
```

#### 2. 密钥配置
```python
import os
from getpass import getpass

def secure_config(keys):
    """安全配置环境密钥"""
    for key in keys:
        if not os.environ.get(key):
            os.environ[key] = getpass(f"请输入{key}: ")

required_keys = ["ANTHROPIC_API_KEY", "LANGSMITH_API_KEY"]
secure_config(required_keys)

# 启用运行监控
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "智能客服系统"
```

### 三、核心组件实战应用

#### 1. 状态容器设计
```python
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    """对话状态容器"""
    message_history: Annotated[list, add_messages]  # 消息历史
    session_context: dict  # 会话上下文
    pending_actions: list  # 待处理操作
```

#### 2. 节点与边配置
```python
from langgraph.graph import StateGraph

# 初始化状态图
workflow = StateGraph(AgentState)

# 添加功能节点
workflow.add_node("intent_parser", parse_intent)
workflow.add_node("knowledge_retriever", retrieve_knowledge)
workflow.add_node("response_generator", generate_response)

# 配置条件边
def route_intent(state: AgentState):
    if state["intent"] == "FAQ":
        return "knowledge_retriever"
    else:
        return "response_generator"

workflow.add_conditional_edges(
    "intent_parser",
    route_intent,
    {"knowledge_retriever", "response_generator"}
)
```

### 四、典型应用场景实现

#### 场景1：智能翻译工作流
```python
def translation_flow(state: AgentState):
    """多阶段翻译流程"""
    # 初始化翻译节点
    workflow.add_node("translate_phase1", initial_translation)
    workflow.add_node("quality_check", evaluate_quality)
    workflow.add_node("finalize_translation", final_output)

    # 配置循环校验
    workflow.add_conditional_edges(
        "quality_check",
        lambda s: "translate_phase1" if s["score"] < 95 else "finalize_translation"
    )
```

#### 场景2：个性化聊天机器人
```python
class UserProfile(TypedDict):
    """用户画像存储结构"""
    preferences: dict
    interaction_history: list
    last_session: str

# 持久化配置
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store import InMemoryStore

store = InMemoryStore()
checkpointer = MemorySaver()

# 构建个性化对话流程
def personal_chatbot(state):
    """带用户画像的聊天流程"""
    # 加载用户数据
    user_data = store.get(("profiles", state["user_id"]))
    # 生成个性化回复
    return format_response(user_data)
```

### 五、高级特性应用

#### 1. 状态持久化实战
```python
# 配置检查点
from langgraph.checkpoint import MemorySaver

checkpointer = MemorySaver()
compiled_graph = workflow.compile(checkpointer=checkpointer)

# 保存会话状态
config = {"configurable": {"thread_id": "user123"}}
compiled_graph.invoke(
    {"messages": [("user", "推荐北京的美食")]},
    config=config
)

# 恢复会话
saved_state = checkpointer.get(config)
```

#### 2. 多代理协作架构
```python
def multi_agent_system():
    """多代理协作系统"""
    # 初始化代理节点
    workflow.add_node("reception_agent", handle_greeting)
    workflow.add_node("tech_agent", solve_technical)
    workflow.add_node("sales_agent", process_order)

    # 配置智能路由
    def route_to_specialist(state):
        if "技术问题" in state["tags"]:
            return "tech_agent"
        elif "订单" in state["tags"]:
            return "sales_agent"
        return "reception_agent"
```

### 六、调试与优化技巧

#### 1. 可视化监控
```python
# 生成流程拓扑图
from IPython.display import Image

try:
    Image(workflow.get_graph().draw_mermaid_png())
except Exception as e:
    print(f"可视化错误: {str(e)}")
```

#### 2. 性能优化策略
- **节点缓存**：对稳定节点结果进行缓存
- **异步执行**：对独立节点启用并行处理
- **资源限制**：配置执行超时和重试机制

```python
# 异步节点配置示例
async def async_knowledge_query(state):
    """异步知识查询"""
    result = await query_knowledge_base(state["query"])
    return {"result": result}
```

### 七、生产环境最佳实践

1. **安全防护**：
   - 输入内容过滤
   - 输出内容审核
   - 权限分级控制

2. **性能保障**：
   ```python
   # 限流配置示例
   from langgraph import RateLimiter

   limiter = RateLimiter(requests=100, per_minute=5)
   workflow.add_middleware(limiter)
   ```

3. **监控体系**：
   - 成功率监控
   - 响应时间监控
   - 资源消耗监控

4. **CI/CD流程**：
   ```mermaid
   graph TD
       A[代码提交] --> B[单元测试]
       B --> C[流程验证]
       C --> D[性能压测]
       D --> E[灰度发布]
       E --> F[全量部署]
   ```

### 八、扩展阅读推荐

1. 官方文档精读：
   - 状态管理高级技巧
   - 分布式执行模式
   - 自定义序列化方案

2. 进阶应用场景：
   - 金融风控决策流程
   - 医疗诊断辅助系统
   - 智能制造排程系统

3. 性能调优手册：
   ```python
   # 性能分析示例
   from cProfile import Profile
   profiler = Profile()
   profiler.runcall(workflow.invoke, input_state)
   profiler.print_stats()
   ```

本指南通过理论解析与实战案例的结合，系统性地展示了LangGraph的核心能力与应用技巧。建议读者：
1. 从简单流程入手逐步验证概念
2. 善用可视化工具分析流程瓶颈
3. 定期参与开发者社区交流
4. 关注官方更新日志获取新特性

期待您用LangGraph构建出更智能、更可靠的AI应用！

LangGraph实战指南：构建智能代理工作流的全解析

### 🌟 多智能体监督员：让AI团队协作更高效！🌟

大家好！今天我们要聊的是一个超级酷炫的话题——**多智能体监督员**！想象一下，你有一支AI团队，每个成员都有自己擅长的领域，比如有的擅长做研究，有的擅长写代码。那么，如何让这些AI成员高效协作，完成复杂的任务呢？这就是我们今天要探讨的内容！

#### 🛠️ 准备工作

首先，我们需要安装一些必要的工具包，并设置API密钥。别担心，跟着我一步步来，你也能轻松搞定！

```python
%%capture --no-stell
%pip install -U langgraph langchain_community langchain_anthropic langchain_experimental

import getpass
import os

def _set_if_undefined(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"请提供你的 {var}")

_set_if_undefined("ANTHROPIC_API_KEY")
_set_if_undefined("TAVILY_API_KEY")
```

#### 🛠️ 创建工具

接下来，我们需要为我们的AI团队创建一些工具。比如，一个用于网络搜索的工具，另一个用于生成图表的工具。

```python
from typing import Annotated
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.tools import tool
from langchain_experimental.utilities import PythonREPL

tavily_tool = TavilySearchResults(max_results=5)

# 这个工具可以执行Python代码，但要注意安全哦！
repl = PythonREPL()

@tool
def python_repl_tool(
    code: Annotated[str, "要执行的Python代码，用于生成图表。"],
):
    """使用这个工具来执行Python代码并进行数学计算。如果你想看到某个值的输出，记得用`print(...)`打印出来哦！"""
    try:
        result = repl.run(code)
    except BaseException as e:
        return f"执行失败。错误：{repr(e)}"
    result_str = f"成功执行：\n\`\`\`python\n{code}\n\`\`\`\n标准输出：{result}"
    return result_str
```

#### 🕵️‍♂️ 创建智能体监督员

现在，我们需要一个“监督员”来协调这些AI成员的工作。这个监督员会使用LLM（大语言模型）来决定下一个该由谁来处理任务，或者任务是否已经完成。

```python
from typing import Literal
from typing_extensions import TypedDict
from langchain_anthropic import ChatAnthropic
from langgraph.graph import MessagesState, END
from langgraph.types import Command

members = ["researcher", "coder"]
options = members + ["FINISH"]

system_prompt = (
    "你是一个监督员，负责管理以下成员之间的对话：{members}。根据用户请求，决定下一个该由谁来处理任务。每个成员会执行任务并返回结果和状态。当任务完成时，请返回FINISH。"
)

class Router(TypedDict):
    """决定下一个该由谁来处理任务。如果不需要其他成员处理，返回FINISH。"""
    next: Literal[*options]

llm = ChatAnthropic(model="claude-3-5-sonnet-latest")

class State(MessagesState):
    next: str

def supervisor_node(state: State) -> Command[Literal[*members, "__end__"]]:
    messages = [
        {"role": "system", "content": system_prompt},
    ] + state["messages"]
    response = llm.with_structured_output(Router).invoke(messages)
    goto = response["next"]
    if goto == "FINISH":
        goto = END

    return Command(goto=goto, update={"next": goto})
```

#### 🏗️ 构建图

接下来，我们开始构建图。定义状态和各个工作节点。

```python
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import create_react_agent

research_agent = create_react_agent(
    llm, tools=[tavily_tool], prompt="你是一个研究员。不要做任何数学计算。"
)

def research_node(state: State) -> Command[Literal["supervisor"]]:
    result = research_agent.invoke(state)
    return Command(
        update={
            "messages": [
                HumanMessage(content=result["messages"][-1].content, name="researcher")
            ]
        },
        goto="supervisor",
    )

# 注意：这个工具可以执行任意代码，使用时要注意安全！
code_agent = create_react_agent(llm, tools=[python_repl_tool])

def code_node(state: State) -> Command[Literal["supervisor"]]:
    result = code_agent.invoke(state)
    return Command(
        update={
            "messages": [
                HumanMessage(content=result["messages"][-1].content, name="coder")
            ]
        },
        goto="supervisor",
    )

builder = StateGraph(State)
builder.add_edge(START, "supervisor")
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", research_node)
builder.add_node("coder", code_node)
graph = builder.compile()
```

#### 🚀 启动团队

现在，我们的AI团队已经准备就绪，可以开始执行任务了！

```python
for s in graph.stream(
    {"messages": [("user", "42的平方根是多少？")]}, subgraphs=True
):
    print(s)
    print("----")
```

#### 🎉 总结

通过这个教程，我们学会了如何创建一个多智能体监督员，让AI团队成员高效协作，完成复杂的任务。希望这个教程对你有所帮助，快去试试吧！如果你有任何问题或想法，欢迎在评论区留言哦！😄

---

**小贴士**：在使用Python REPL工具时，一定要注意代码的安全性，避免执行恶意代码哦！🔒

Happy coding! 🚀

智能体监督者


单个智能体通常可以在单个领域内使用少量工具有效地操作，但即使使用像GPT-4这样强大的模型，它在使用许多工具时也可能效果不佳。

处理复杂任务的一种方法是通过“分而治之”的方式：为每个任务或领域创建一个专门的智能体，并将任务路由到正确的“专家”。这是多智能体网络架构的一个示例。

本笔记本（灵感来自Wu等人的论文《AutoGen: 通过多智能体对话实现下一代LLM应用》）展示了如何使用LangGraph实现这一点。

生成的图将类似于以下图表：

多智能体图

在我们开始之前，快速说明一下：本笔记本和其他多智能体笔记本旨在展示如何在LangGraph中实现某些设计模式。如果该模式符合您的需求，我们建议将其与文档中描述的其他基本模式结合使用，以获得最佳性能。


首先，让我们安装所需的包并设置API密钥：

```python
%%capture --no-stderr
%pip install -U langchain_community langchain_anthropic langchain_experimental matplotlib langgraph

import getpass
import os

def _set_if_undefined(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"Please provide your {var}")

_set_if_undefined("ANTHROPIC_API_KEY")
_set_if_undefined("TAVILY_API_KEY")
```

为LangGraph开发设置LangSmith

注册LangSmith以快速发现问题并提高LangGraph项目的性能。LangSmith允许您使用跟踪数据来调试、测试和监控使用LangGraph构建的LLM应用程序——阅读更多关于如何开始的信息。


我们还将定义一些工具，供我们的智能体将来使用：

```python
from typing import Annotated
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.tools import tool
from langchain_experimental.utilities import PythonREPL

tavily_tool = TavilySearchResults(max_results=5)

# 警告：此代码在本地执行，未沙盒化时可能不安全
repl = PythonREPL()

@tool
def python_repl_tool(
    code: Annotated[str, "The python code to execute to generate your chart."],
):
    """使用此工具执行Python代码。如果你想查看某个值的输出，你应该使用`print(...)`打印出来。用户可以看到此输出。"""
    try:
        result = repl.run(code)
    except BaseException as e:
        return f"Failed to execute. Error: {repr(e)}"
    result_str = f"Successfully executed:\n\`\`\`python\n{code}\n\`\`\`\nStdout: {result}"
    return (
        result_str + "\n\nIf you have completed all tasks, respond with FINAL ANSWER."
    )
```


现在我们已经定义了工具并创建了一些辅助函数，我们将创建各个智能体，并使用LangGraph告诉它们如何相互通信。


我们现在需要定义节点。

首先，我们将创建一个实用程序来为每个智能体生成系统提示。

```python
def make_system_prompt(suffix: str) -> str:
    return (
        "You are a helpful AI assistant, collaborating with other assistants."
        " Use the provided tools to progress towards answering the question."
        " If you are unable to fully answer, that's OK, another assistant with different tools "
        " will help where you left off. Execute what you can to make progress."
        " If you or any of the other assistants have the final answer or deliverable,"
        " prefix your response with FINAL ANSWER so the team knows to stop."
        f"\n{suffix}"
    )

from typing import Literal
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_anthropic import ChatAnthropic
from langgraph.prebuilt import create_react_agent
from langgraph.graph import MessagesState, END
from langgraph.types import Command

llm = ChatAnthropic(model="claude-3-5-sonnet-latest")

def get_next_node(last_message: BaseMessage, goto: str):
    if "FINAL ANSWER" in last_message.content:
        # 任何智能体决定工作已完成
        return END
    return goto

# 研究智能体和节点
research_agent = create_react_agent(
    llm,
    tools=[tavily_tool],
    prompt=make_system_prompt(
        "You can only do research. You are working with a chart generator colleague."
    ),
)

def research_node(
    state: MessagesState,
) -> Command[Literal["chart_generator", END]]:
    result = research_agent.invoke(state)
    goto = get_next_node(result["messages"][-1], "chart_generator")
    # 包装在人类消息中，因为并非所有提供程序都允许
    # 在输入消息列表的最后位置使用AI消息
    result["messages"][-1] = HumanMessage(
        content=result["messages"][-1].content, name="researcher"
    )
    return Command(
        update={
            # 与其他智能体共享研究智能体的内部消息历史
            "messages": result["messages"],
        },
        goto=goto,
    )

# 图表生成智能体和节点
# 注意：此操作执行任意代码执行，未沙盒化时可能不安全
chart_agent = create_react_agent(
    llm,
    [python_repl_tool],
    prompt=make_system_prompt(
        "You can only generate charts. You are working with a researcher colleague."
    ),
)

def chart_node(state: MessagesState) -> Command[Literal["researcher", END]]:
    result = chart_agent.invoke(state)
    goto = get_next_node(result["messages"][-1], "researcher")
    # 包装在人类消息中，因为并非所有提供程序都允许
    # 在输入消息列表的最后位置使用AI消息
    result["messages"][-1] = HumanMessage(
        content=result["messages"][-1].content, name="chart_generator"
    )
    return Command(
        update={
            # 与其他智能体共享图表生成智能体的内部消息历史
            "messages": result["messages"],
        },
        goto=goto,
    )
```


我们现在可以将所有内容放在一起并定义图！

```python
from langgraph.graph import StateGraph, START

workflow = StateGraph(MessagesState)
workflow.add_node("researcher", research_node)
workflow.add_node("chart_generator", chart_node)

workflow.add_edge(START, "researcher")
graph = workflow.compile()
```


图创建完成后，您可以调用它！让我们用它来绘制一些统计数据。

```python
events = graph.stream(
    {
        "messages": [
            (
                "user",
                "First, get the UK's GDP over the past 5 years, then make a line chart of it. "
                "Once you make the chart, finish.",
            )
        ],
    },
    # 图中执行的最大步骤数
    {"recursion_limit": 150},
)
for s in events:
    print(s)
    print("----")
```

输出结果将显示智能体之间的交互过程，最终生成一个包含英国过去5年GDP数据的折线图。

多Agent

### 使用本地LLM实现Self-RAG

Self-RAG是一种结合了自我反思/自我评分的RAG（检索增强生成）策略。在论文中，Self-RAG通过以下几个决策点来优化检索和生成过程：

1. **是否从检索器中检索（R）**  
   - 输入：问题（x）或问题（x）和生成结果（y）  
   - 决定何时从检索器R中检索D个文档块  
   - 输出：是、否、继续

2. **检索到的文档块D是否与问题x相关**  
   - 输入：问题（x）和文档块（d）  
   - 判断d是否为解决x提供了有用的信息  
   - 输出：相关、不相关

3. **LLM生成的每个文档块D的生成结果是否与文档块相关（是否存在幻觉等）**  
   - 输入：问题（x）、文档块（d）、生成结果（y）  
   - 判断y中的所有可验证陈述是否由d支持  
   - 输出：完全支持、部分支持、不支持

4. **LLM生成的每个文档块D的生成结果是否对问题x有用**  
   - 输入：问题（x）、生成结果（y）  
   - 判断y是否对解决x有用  
   - 输出：5、4、3、2、1

我们将使用LangGraph从头实现这些思想。

---

### 设置

首先，安装所需的包并设置API密钥。

```python
%%capture --no-stderr
%pip install -U langchain-nomic langchain_community tiktoken langchainhub chromadb langchain langgraph nomic[local]

import getpass
import os

def _set_env(key: str):
    if key not in os.environ:
        os.environ[key] = getpass.getpass(f"{key}:")

_set_env("NOMIC_API_KEY")
```

---

### 本地LLM和嵌入

1. **本地嵌入**  
   使用Nomic的`GPT4AllEmbeddings()`，支持Nomic发布的v1和v1.5嵌入模型。  
   参考[文档](https://docs.nomic.ai/)。

2. **本地LLM**  
   - 下载Ollama应用。  
   - 从[Mistral版本](https://mistral.ai/models/)或[Mixtral版本](https://mixtral.ai/models/)下载模型。  
   - 使用以下命令拉取模型：  
     ```bash
     ollama pull mistral
     ```  
   - 设置本地LLM名称：  
     ```python
     local_llm = "mistral"
     ```

---

### 创建索引

索引3篇博客文章。

```python
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores import Chroma
from langchain_nomic.embeddings import NomicEmbeddings

urls = [
    "https://lilianweng.github.io/posts/2023-06-23-agent/",
    "https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/",
    "https://lilianweng.github.io/posts/2023-10-25-adv-attack-llm/",
]

docs = [WebBaseLoader(url).load() for url in urls]
docs_list = [item for sublist in docs for item in sublist]

text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=250, chunk_overlap=0
)
doc_splits = text_splitter.split_documents(docs_list)

# 添加到向量数据库
vectorstore = Chroma.from_documents(
    documents=doc_splits,
    collection_name="rag-chroma",
    embedding=NomicEmbeddings(model="nomic-embed-text-v1.5", inference_mode="local"),
)
retriever = vectorstore.as_retriever()
```

---

### 检索评分器

```python
from langchain.prompts import PromptTemplate
from langchain_community.chat_models import ChatOllama
from langchain_core.output_parsers import JsonOutputParser

# LLM
llm = ChatOllama(model=local_llm, format="json", temperature=0)

prompt = PromptTemplate(
    template="""You are a grader assessing relevance of a retrieved document to a user question. \n 
    Here is the retrieved document: \n\n {document} \n\n
    Here is the user question: {question} \n
    If the document contains keywords related to the user question, grade it as relevant. \n
    It does not need to be a stringent test. The goal is to filter out erroneous retrievals. \n
    Give a binary score 'yes' or 'no' score to indicate whether the document is relevant to the question. \n
    Provide the binary score as a JSON with a single key 'score' and no premable or explanation.""",
    input_variables=["question", "document"],
)

retrieval_grader = prompt | llm | JsonOutputParser()
question = "agent memory"
docs = retriever.invoke(question)
doc_txt = docs[1].page_content
print(retrieval_grader.invoke({"question": question, "document": doc_txt}))
```

---

### 生成

```python
from langchain import hub
from langchain_core.output_parsers import StrOutputParser

# Prompt
prompt = hub.pull("rlm/rag-prompt")

# LLM
llm = ChatOllama(model=local_llm, temperature=0)

# 后处理
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# Chain
rag_chain = prompt | llm | StrOutputParser()

# 运行
generation = rag_chain.invoke({"context": docs, "question": question})
print(generation)
```

---

### 幻觉评分器

```python
# LLM
llm = ChatOllama(model=local_llm, format="json", temperature=0)

# Prompt
prompt = PromptTemplate(
    template="""You are a grader assessing whether an answer is grounded in / supported by a set of facts. \n 
    Here are the facts:
    \n ------- \n
    {documents} 
    \n ------- \n
    Here is the answer: {generation}
    Give a binary score 'yes' or 'no' score to indicate whether the answer is grounded in / supported by a set of facts. \n
    Provide the binary score as a JSON with a single key 'score' and no preamble or explanation.""",
    input_variables=["generation", "documents"],
)

hallucination_grader = prompt | llm | JsonOutputParser()
hallucination_grader.invoke({"documents": docs, "generation": generation})
```

---

### 答案评分器

```python
# LLM
llm = ChatOllama(model=local_llm, format="json", temperature=0)

# Prompt
prompt = PromptTemplate(
    template="""You are a grader assessing whether an answer is useful to resolve a question. \n 
    Here is the answer:
    \n ------- \n
    {generation} 
    \n ------- \n
    Here is the question: {question}
    Give a binary score 'yes' or 'no' to indicate whether the answer is useful to resolve a question. \n
    Provide the binary score as a JSON with a single key 'score' and no preamble or explanation.""",
    input_variables=["generation", "question"],
)

answer_grader = prompt | llm | JsonOutputParser()
answer_grader.invoke({"question": question, "generation": generation})
```

---

### 问题重写器

```python
# LLM
llm = ChatOllama(model=local_llm, temperature=0)

# Prompt
re_write_prompt = PromptTemplate(
    template="""You a question re-writer that converts an input question to a better version that is optimized \n 
     for vectorstore retrieval. Look at the initial and formulate an improved question. \n
     Here is the initial question: \n\n {question}. Improved question with no preamble: \n """,
    input_variables=["generation", "question"],
)

question_rewriter = re_write_prompt | llm | StrOutputParser()
question_rewriter.invoke({"question": question})
```

---

### 构建图

将上述流程封装为图。

```python
from langgraph.graph import END, StateGraph, START

workflow = StateGraph(GraphState)

# 定义节点
workflow.add_node("retrieve", retrieve)  # 检索
workflow.add_node("grade_documents", grade_documents)  # 评分文档
workflow.add_node("generate", generate)  # 生成
workflow.add_node("transform_query", transform_query)  # 重写问题

# 构建图
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {
        "transform_query": "transform_query",
        "generate": "generate",
    },
)
workflow.add_edge("transform_query", "retrieve")
workflow.add_conditional_edges(
    "generate",
    grade_generation_v_documents_and_question,
    {
        "not supported": "generate",
        "useful": END,
        "not useful": "transform_query",
    },
)

# 编译
app = workflow.compile()
```

---

### 运行

```python
from pprint import pprint

# 运行
inputs = {"question": "Explain how the different types of agent memory work?"}
for output in app.stream(inputs):
    for key, value in output.items():
        # 节点
        pprint(f"Node '{key}':")
        # 可选：打印每个节点的完整状态
        # pprint.pprint(value["keys"], indent=2, width=80, depth=None)
    pprint("\n---\n")

# 最终生成结果
pprint(value["generation"])
```

---

### 总结

通过本教程，我们使用本地LLM和LangGraph实现了Self-RAG策略，优化了检索和生成过程。Self-RAG通过自我反思和评分机制，显著提升了生成结果的质量和相关性。