### 如何解析 JSON 输出  

#### 前提条件  
在阅读本教程之前，你需要对以下概念有一定的了解：  
- 聊天模型（Chat Models）  
- 输出解析器（Output Parsers）  
- 提示模板（Prompt Templates）  
- 结构化输出（Structured Output）  
- 将可运行对象链接在一起（Chaining Runnables Together）  

#### 为什么需要解析 JSON 输出？  
虽然一些模型提供商支持内置的方式返回结构化输出，但并非所有提供商都支持。我们可以使用输出解析器来帮助用户通过提示指定任意的 JSON 模式，查询模型以生成符合该模式的输出，并最终将该模式解析为 JSON。  

**注意**：大语言模型是“有漏洞的抽象”！你需要使用具有足够能力的 LLM 来生成格式良好的 JSON。  

`JsonOutputParser` 是一个内置选项，用于提示并解析 JSON 输出。虽然它在功能上与 `PydanticOutputParser` 类似，但它还支持流式返回部分 JSON 对象。  

以下是一个示例，展示如何与 Pydantic 结合使用，方便地声明预期的模式：  

---

#### 安装依赖  
```bash
%pip install -qU langchain langchain-openai
```

---

#### 示例代码  
```python
import os
from getpass import getpass

# 设置 OpenAI API 密钥
if "OPENAI_API_KEY" not in os.environ:
    os.environ["OPENAI_API_KEY"] = getpass()

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

# 初始化模型
model = ChatOpenAI(temperature=0)

# 定义期望的数据结构
class Joke(BaseModel):
    setup: str = Field(description="笑话的开头问题")
    punchline: str = Field(description="笑话的结尾答案")

# 定义查询
joke_query = "Tell me a joke."

# 设置解析器并将指令注入提示模板
parser = JsonOutputParser(pydantic_object=Joke)

prompt = PromptTemplate(
    template="Answer the user query.\n{format_instructions}\n{query}\n",
    input_variables=["query"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

# 构建链
chain = prompt | model | parser

# 调用链
result = chain.invoke({"query": joke_query})
print(result)
```

---

#### 输出示例  
```json
{
    "setup": "Why couldn't the bicycle stand up by itself?",
    "punchline": "Because it was two tired!"
}
```

---

#### 解析器的格式指令  
`parser.get_format_instructions()` 返回的指令如下：  
```plaintext
The output should be formatted as a JSON instance that conforms to the JSON schema below.

As an example, for the schema {"properties": {"foo": {"title": "Foo", "description": "a list of strings", "type": "array", "items": {"type": "string"}}}, "required": ["foo"]}
the object {"foo": ["bar", "baz"]} is a well-formatted instance of the schema. The object {"properties": {"foo": ["bar", "baz"]}} is not well-formatted.

Here is the output schema:
```json
{
    "properties": {
        "setup": {"title": "Setup", "description": "question to set up a joke", "type": "string"},
        "punchline": {"title": "Punchline", "description": "answer to resolve the joke", "type": "string"}
    },
    "required": ["setup", "punchline"]
}
```

你可以尝试在提示的其他部分添加自己的格式化提示，以增强或替换默认指令。  

---

#### 流式输出  
`JsonOutputParser` 支持流式返回部分 JSON 对象。以下是一个流式输出的示例：  

```python
for s in chain.stream({"query": joke_query}):
    print(s)
```

输出示例：  
```json
{}
{'setup': ''}
{'setup': 'Why'}
{'setup': 'Why couldn'}
{'setup': "Why couldn't"}
{'setup': "Why couldn't the"}
{'setup': "Why couldn't the bicycle"}
{'setup': "Why couldn't the bicycle stand"}
{'setup': "Why couldn't the bicycle stand up"}
{'setup': "Why couldn't the bicycle stand up by"}
{'setup': "Why couldn't the bicycle stand up by itself"}
{'setup': "Why couldn't the bicycle stand up by itself?"}
{'setup': "Why couldn't the bicycle stand up by itself?", 'punchline': ''}
{'setup': "Why couldn't the bicycle stand up by itself?", 'punchline': 'Because'}
{'setup': "Why couldn't the bicycle stand up by itself?", 'punchline': 'Because it'}
{'setup': "Why couldn't the bicycle stand up by itself?", 'punchline': 'Because it was'}
{'setup': "Why couldn't the bicycle stand up by itself?", 'punchline': 'Because it was two'}
{'setup': "Why couldn't the bicycle stand up by itself?", 'punchline': 'Because it was two tired'}
{'setup': "Why couldn't the bicycle stand up by itself?", 'punchline': 'Because it was two tired!'}
```

---

#### 不使用 Pydantic  
你也可以在不使用 Pydantic 的情况下使用 `JsonOutputParser`。这会提示模型返回 JSON，但不提供关于模式的具体信息。  

```python
joke_query = "Tell me a joke."

parser = JsonOutputParser()

prompt = PromptTemplate(
    template="Answer the user query.\n{format_instructions}\n{query}\n",
    input_variables=["query"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

chain = prompt | model | parser

result = chain.invoke({"query": joke_query})
print(result)
```

输出示例：  
```json
{
    "response": "Sure! Here's a joke for you: Why couldn't the bicycle stand up by itself? Because it was two tired!"
}
```

---

#### 下一步  
现在你已经学会了如何解析 JSON 输出。接下来，你可以查看以下指南：  
- [如何让模型返回结构化输出](https://example.com)  
- [如何创建自定义聊天模型](https://example.com)  

希望这篇教程对你有帮助！如果有任何问题，欢迎留言讨论~ 🚀
```

LLM 格式化输出Json

### 如何从模型中返回结构化数据

在实际应用中，我们经常需要让模型返回符合特定结构的数据。比如，从文本中提取信息并插入到数据库中，或者用于其他下游系统。本文将介绍几种从模型中获取结构化输出的策略。

#### 使用 `.with_structured_output()` 方法

**支持的模型**  
你可以在这里找到支持该方法的模型列表。

这是获取结构化输出最简单且最可靠的方法。`with_structured_output()` 方法利用了模型的原生 API（如工具/函数调用或 JSON 模式），并在此基础上实现结构化输出。

该方法接受一个模式（schema）作为输入，该模式指定了输出属性的名称、类型和描述。方法返回一个类似模型的 `Runnable`，但它输出的不是字符串或消息，而是与给定模式对应的对象。模式可以是一个 `TypedDict` 类、JSON Schema 或 Pydantic 类。如果使用 `TypedDict` 或 JSON Schema，`Runnable` 将返回一个字典；如果使用 Pydantic 类，则返回一个 Pydantic 对象。

**示例：生成笑话并分离笑点和笑料**

首先，我们选择一个聊天模型并安装必要的库：

```bash
pip install -qU "langchain[groq]"
```

然后，设置 API 密钥并初始化模型：

```python
import getpass
import os

if not os.environ.get("GROQ_API_KEY"):
  os.environ["GROQ_API_KEY"] = getpass.getpass("Enter API key for Groq: ")

from langchain.chat_models import init_chat_model

llm = init_chat_model("llama3-8b-8192", model_provider="groq")
```

**使用 Pydantic 类**  
如果我们希望模型返回一个 Pydantic 对象，只需传入所需的 Pydantic 类。使用 Pydantic 的主要优势是，模型生成的输出将被验证。如果缺少任何必填字段或字段类型错误，Pydantic 将抛出错误。

```python
from typing import Optional
from pydantic import BaseModel, Field

class Joke(BaseModel):
    """笑话内容"""
    setup: str = Field(description="笑话的开头")
    punchline: str = Field(description="笑话的笑点")
    rating: Optional[int] = Field(default=None, description="笑话的有趣程度，1到10分")

structured_llm = llm.with_structured_output(Joke)
structured_llm.invoke("讲一个关于猫的笑话")
```

输出示例：

```python
Joke(setup='为什么猫坐在电脑上？', punchline='因为它想盯着鼠标！', rating=7)
```

**使用 TypedDict 或 JSON Schema**  
如果你不想使用 Pydantic，或者不想验证参数，或者希望能够流式传输模型输出，可以使用 `TypedDict` 类来定义模式。我们还可以使用 LangChain 支持的 `Annotated` 语法来指定字段的默认值和描述。

```python
from typing import Optional
from typing_extensions import Annotated, TypedDict

class Joke(TypedDict):
    """笑话内容"""
    setup: Annotated[str, ..., "笑话的开头"]
    punchline: Annotated[str, ..., "笑话的笑点"]
    rating: Annotated[Optional[int], None, "笑话的有趣程度，1到10分"]

structured_llm = llm.with_structured_output(Joke)
structured_llm.invoke("讲一个关于猫的笑话")
```

输出示例：

```python
{'setup': '为什么猫坐在电脑上？', 'punchline': '因为它想盯着鼠标！', 'rating': 7}
```

**使用 JSON Schema**  
我们也可以直接传入一个 JSON Schema 字典。这种方式不需要导入任何类，且能清晰地展示每个参数的文档，但会稍微冗长一些。

```python
json_schema = {
    "title": "joke",
    "description": "笑话内容",
    "type": "object",
    "properties": {
        "setup": {"type": "string", "description": "笑话的开头"},
        "punchline": {"type": "string", "description": "笑话的笑点"},
        "rating": {"type": "integer", "description": "笑话的有趣程度，1到10分", "default": None},
    },
    "required": ["setup", "punchline"],
}

structured_llm = llm.with_structured_output(json_schema)
structured_llm.invoke("讲一个关于猫的笑话")
```

输出示例：

```python
{'setup': '为什么猫坐在电脑上？', 'punchline': '因为它想盯着鼠标！', 'rating': 7}
```

**在多个模式之间选择**  
最简单的让模型在多个模式之间选择的方法是创建一个包含 `Union` 类型属性的父模式。

**使用 Pydantic**  
```python
from typing import Union

class Joke(BaseModel):
    """笑话内容"""
    setup: str = Field(description="笑话的开头")
    punchline: str = Field(description="笑话的笑点")
    rating: Optional[int] = Field(default=None, description="笑话的有趣程度，1到10分")

class ConversationalResponse(BaseModel):
    """以对话方式回应。保持友好和乐于助人。"""
    response: str = Field(description="对用户查询的对话式回应")

class FinalResponse(BaseModel):
    final_output: Union[Joke, ConversationalResponse]

structured_llm = llm.with_structured_output(FinalResponse)
structured_llm.invoke("讲一个关于猫的笑话")
```

输出示例：

```python
FinalResponse(final_output=Joke(setup='为什么猫坐在电脑上？', punchline='因为它想盯着鼠标！', rating=7))
```

**使用 TypedDict**  
```python
from typing import Optional, Union
from typing_extensions import Annotated, TypedDict

class Joke(TypedDict):
    """笑话内容"""
    setup: Annotated[str, ..., "笑话的开头"]
    punchline: Annotated[str, ..., "笑话的笑点"]
    rating: Annotated[Optional[int], None, "笑话的有趣程度，1到10分"]

class ConversationalResponse(TypedDict):
    """以对话方式回应。保持友好和乐于助人。"""
    response: Annotated[str, ..., "对用户查询的对话式回应"]

class FinalResponse(TypedDict):
    final_output: Union[Joke, ConversationalResponse]

structured_llm = llm.with_structured_output(FinalResponse)
structured_llm.invoke("讲一个关于猫的笑话")
```

输出示例：

```python
{'final_output': {'setup': '为什么猫坐在电脑上？', 'punchline': '因为它想盯着鼠标！', 'rating': 7}}
```

**流式传输**  
当输出类型为字典时（即模式为 `TypedDict` 类或 JSON Schema 字典），我们可以流式传输结构化模型的输出。

```python
from typing_extensions import Annotated, TypedDict

class Joke(TypedDict):
    """笑话内容"""
    setup: Annotated[str, ..., "笑话的开头"]
    punchline: Annotated[str, ..., "笑话的笑点"]

structured_llm = llm.with_structured_output(Joke)
for chunk in structured_llm.stream("讲一个关于猫的笑话"):
    print(chunk)
```

输出示例：

```python
{'setup': '为什么猫坐在电脑上？', 'punchline': '因为它想盯着鼠标！'}
```

**（高级）原始输出**  
LLM 在生成结构化输出时并不完美，尤其是当模式变得复杂时。你可以通过传递 `include_raw=True` 来避免抛出异常，并自行处理原始输出。这将更改输出格式，包含原始消息输出、解析后的值（如果成功）以及任何错误。

```python
structured_llm = llm.with_structured_output(Joke, include_raw=True)
structured_llm.invoke("讲一个关于猫的笑话")
```

输出示例：

```python
{'raw': AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_f25ZRmh8u5vHlOWfTUw8sJFZ', 'function': {'arguments': '{"setup":"为什么猫坐在电脑上？","punchline":"因为它想盯着鼠标！","rating":7}', 'name': 'Joke'}, 'type': 'function'}]}, response_metadata={'token_usage': {'completion_tokens': 33, 'prompt_tokens': 93, 'total_tokens': 126}, 'model_name': 'gpt-4o-2024-05-13', 'system_fingerprint': 'fp_4e2b2da518', 'finish_reason': 'stop', 'logprobs': None}, id='run-d880d7e2-df08-4e9e-ad92-dfc29f2fd52f-0', tool_calls=[{'name': 'Joke', 'args': {'setup': '为什么猫坐在电脑上？', 'punchline': '因为它想盯着鼠标！', 'rating': 7}, 'id': 'call_f25ZRmh8u5vHlOWfTUw8sJFZ', 'type': 'tool_call'}], usage_metadata={'input_tokens': 93, 'output_tokens': 33, 'total_tokens': 126}),
 'parsed': {'setup': '为什么猫坐在电脑上？', 'punchline': '因为它想盯着鼠标！', 'rating': 7},
 'parsing_error': None}
```

**直接提示和解析模型输出**  
并非所有模型都支持 `.with_structured_output()`，因为并非所有模型都支持工具调用或 JSON 模式。对于这些模型，你需要直接提示模型使用特定格式，并使用输出解析器从原始模型输出中提取结构化响应。

**使用 PydanticOutputParser**  
以下示例使用内置的 `PydanticOutputParser` 来解析聊天模型的输出，并提示其匹配给定的 Pydantic 模式。

```python
from typing import List
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field

class Person(BaseModel):
    """个人信息"""
    name: str = Field(..., description="姓名")
    height_in_meters: float = Field(..., description="身高，以米为单位")

class People(BaseModel):
    """文本中所有人的信息"""
    people: List[Person]

# 设置解析器
parser = PydanticOutputParser(pydantic_object=People)

# 提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "回答用户查询。将输出包装在 `json` 标签中\n{format_instructions}"),
    ("human", "{query}"),
]).partial(format_instructions=parser.get_format_instructions())

# 查看发送给模型的信息
query = "安娜23岁，身高6英尺"
print(prompt.invoke({"query": query}).to_string())

# 调用模型
chain = prompt | llm | parser
chain.invoke({"query": query})
```

输出示例：

```python
People(people=[Person(name='安娜', height_in_meters=1.8288)])
```

**自定义解析**  
你还可以使用 LangChain 表达式语言（LCEL）创建自定义提示和解析器，使用普通函数解析模型输出。

```python
import json
import re
from typing import List
from langchain_core.messages import AIMessage
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field

class Person(BaseModel):
    """个人信息"""
    name: str = Field(..., description="姓名")
    height_in_meters: float = Field(..., description="身高，以米为单位")

class People(BaseModel):
    """文本中所有人的信息"""
    people: List[Person]

# 提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "回答用户查询。输出你的答案作为 JSON，并匹配以下模式：\`\`\`json\n{schema}\n\`\`\`。确保将答案包装在 \`\`\`json 和 \`\`\` 标签中"),
    ("human", "{query}"),
]).partial(schema=People.schema())

# 自定义解析器
def extract_json(message: AIMessage) -> List[dict]:
    """从包含 JSON 的字符串中提取 JSON 内容。"""
    text = message.content
    pattern = r"\`\`\`json(.*?)\`\`\`"
    matches = re.findall(pattern, text, re.DOTALL)
    try:
        return [json.loads(match.strip()) for match in matches]
    except Exception:
        raise ValueError(f"解析失败: {message}")

# 查看发送给模型的信息
query = "安娜23岁，身高6英尺"
print(prompt.format_prompt(query=query).to_string())

# 调用模型
chain = prompt | llm | extract_json
chain.invoke({"query": query})
```

输出示例：

```python
[{'people': [{'name': '安娜', 'height_in_meters': 1.8288}]}]
```

通过以上方法，你可以灵活地从模型中获取结构化数据，并根据需求进行进一步处理。

格式化输出

# 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! 🚀