### 教程：使用LangGraph实现Agentic RAG（检索增强生成）

在本教程中，我们将使用LangGraph框架实现一个Agentic RAG（检索增强生成）系统。该系统能够根据用户的问题决定是否从索引中检索信息，并生成相应的回答。以下是详细的实现步骤。

---

### 1. 环境设置
首先，安装所需的Python包并设置API密钥。

```python
%%capture --no-stderr
%pip install -U --quiet langchain-community tiktoken langchain-openai langchainhub chromadb langchain langgraph langchain-text-splitters beautifulsoup4

import getpass
import os

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

_set_env("OPENAI_API_KEY")
```

---

### 2. 创建检索器
我们首先加载3篇博客文章，并将其索引到向量数据库中。

```python
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

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=100, chunk_overlap=50
)
doc_splits = text_splitter.split_documents(docs_list)

# 添加到向量数据库
vectorstore = Chroma.from_documents(
    documents=doc_splits,
    collection_name="rag-chroma",
    embedding=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever()
```

---

### 3. 创建检索工具
我们将检索器封装为一个工具，供Agent调用。

```python
from langchain.tools.retriever import create_retriever_tool

retriever_tool = create_retriever_tool(
    retriever,
    "retrieve_blog_posts",
    "Search and return information about Lilian Weng blog posts on LLM agents, prompt engineering, and adversarial attacks on LLMs.",
)

tools = [retriever_tool]
```

---

### 4. 定义Agent状态
我们定义一个状态对象，用于在图中传递消息。

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

from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
```

---

### 5. 定义节点和边
我们定义图中的节点和边，包括Agent、检索、重写问题和生成回答等功能。

#### 5.1 评估文档相关性
```python
from typing import Literal
from pydantic import BaseModel, Field

def grade_documents(state) -> Literal["generate", "rewrite"]:
    class grade(BaseModel):
        binary_score: str = Field(description="Relevance score 'yes' or 'no'")

    model = ChatOpenAI(temperature=0, model="gpt-4o", streaming=True)
    llm_with_tool = model.with_structured_output(grade)

    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 {context} \n\n
        Here is the user question: {question} \n
        If the document contains keyword(s) or semantic meaning related to the user question, grade it as relevant. \n
        Give a binary score 'yes' or 'no' score to indicate whether the document is relevant to the question.""",
        input_variables=["context", "question"],
    )

    chain = prompt | llm_with_tool
    messages = state["messages"]
    last_message = messages[-1]
    question = messages[0].content
    docs = last_message.content

    scored_result = chain.invoke({"question": question, "context": docs})
    score = scored_result.binary_score

    if score == "yes":
        return "generate"
    else:
        return "rewrite"
```

#### 5.2 调用Agent
```python
def agent(state):
    messages = state["messages"]
    model = ChatOpenAI(temperature=0, streaming=True, model="gpt-4-turbo")
    model = model.bind_tools(tools)
    response = model.invoke(messages)
    return {"messages": [response]}
```

#### 5.3 重写问题
```python
def rewrite(state):
    messages = state["messages"]
    question = messages[0].content

    msg = [
        HumanMessage(
            content=f""" \n 
    Look at the input and try to reason about the underlying semantic intent / meaning. \n 
    Here is the initial question:
    \n ------- \n
    {question} 
    \n ------- \n
    Formulate an improved question: """,
        )
    ]

    model = ChatOpenAI(temperature=0, model="gpt-4-0125-preview", streaming=True)
    response = model.invoke(msg)
    return {"messages": [response]}
```

#### 5.4 生成回答
```python
def generate(state):
    messages = state["messages"]
    question = messages[0].content
    last_message = messages[-1]
    docs = last_message.content

    prompt = hub.pull("rlm/rag-prompt")
    llm = ChatOpenAI(model_name="gpt-4o-mini", temperature=0, streaming=True)

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

    rag_chain = prompt | llm | StrOutputParser()
    response = rag_chain.invoke({"context": docs, "question": question})
    return {"messages": [response]}
```

---

### 6. 构建图
我们将节点和边组合成一个完整的图。

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

workflow = StateGraph(AgentState)

workflow.add_node("agent", agent)
retrieve = ToolNode([retriever_tool])
workflow.add_node("retrieve", retrieve)
workflow.add_node("rewrite", rewrite)
workflow.add_node("generate", generate)

workflow.add_edge(START, "agent")
workflow.add_conditional_edges(
    "agent",
    tools_condition,
    {
        "tools": "retrieve",
        END: END,
    },
)

workflow.add_conditional_edges(
    "retrieve",
    grade_documents,
)
workflow.add_edge("generate", END)
workflow.add_edge("rewrite", "agent")

graph = workflow.compile()
```

---

### 7. 运行图
最后，我们运行图并查看输出。

```python
import pprint

inputs = {
    "messages": [
        ("user", "What does Lilian Weng say about the types of agent memory?"),
    ]
}
for output in graph.stream(inputs):
    for key, value in output.items():
        pprint.pprint(f"Output from node '{key}':")
        pprint.pprint("---")
        pprint.pprint(value, indent=2, width=80, depth=None)
    pprint.pprint("\n---\n")
```

---

### 完整代码
以下是完整的代码实现：

```python
# 环境设置
%%capture --no-stderr
%pip install -U --quiet langchain-community tiktoken langchain-openai langchainhub chromadb langchain langgraph langchain-text-splitters beautifulsoup4

import getpass
import os

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

_set_env("OPENAI_API_KEY")

# 创建检索器
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

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=100, chunk_overlap=50
)
doc_splits = text_splitter.split_documents(docs_list)

vectorstore = Chroma.from_documents(
    documents=doc_splits,
    collection_name="rag-chroma",
    embedding=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever()

# 创建检索工具
from langchain.tools.retriever import create_retriever_tool

retriever_tool = create_retriever_tool(
    retriever,
    "retrieve_blog_posts",
    "Search and return information about Lilian Weng blog posts on LLM agents, prompt engineering, and adversarial attacks on LLMs.",
)

tools = [retriever_tool]

# 定义Agent状态
from typing import Annotated, Sequence
from typing_extensions import TypedDict

from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]

# 定义节点和边
from typing import Literal
from pydantic import BaseModel, Field

def grade_documents(state) -> Literal["generate", "rewrite"]:
    class grade(BaseModel):
        binary_score: str = Field(description="Relevance score 'yes' or 'no'")

    model = ChatOpenAI(temperature=0, model="gpt-4o", streaming=True)
    llm_with_tool = model.with_structured_output(grade)

    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 {context} \n\n
        Here is the user question: {question} \n
        If the document contains keyword(s) or semantic meaning related to the user question, grade it as relevant. \n
        Give a binary score 'yes' or 'no' score to indicate whether the document is relevant to the question.""",
        input_variables=["context", "question"],
    )

    chain = prompt | llm_with_tool
    messages = state["messages"]
    last_message = messages[-1]
    question = messages[0].content
    docs = last_message.content

    scored_result = chain.invoke({"question": question, "context": docs})
    score = scored_result.binary_score

    if score == "yes":
        return "generate"
    else:
        return "rewrite"

def agent(state):
    messages = state["messages"]
    model = ChatOpenAI(temperature=0, streaming=True, model="gpt-4-turbo")
    model = model.bind_tools(tools)
    response = model.invoke(messages)
    return {"messages": [response]}

def rewrite(state):
    messages = state["messages"]
    question = messages[0].content

    msg = [
        HumanMessage(
            content=f""" \n 
    Look at the input and try to reason about the underlying semantic intent / meaning. \n 
    Here is the initial question:
    \n ------- \n
    {question} 
    \n ------- \n
    Formulate an improved question: """,
        )
    ]

    model = ChatOpenAI(temperature=0, model="gpt-4-0125-preview", streaming=True)
    response = model.invoke(msg)
    return {"messages": [response]}

def generate(state):
    messages = state["messages"]
    question = messages[0].content
    last_message = messages[-1]
    docs = last_message.content

    prompt = hub.pull("rlm/rag-prompt")
    llm = ChatOpenAI(model_name="gpt-4o-mini", temperature=0, streaming=True)

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

    rag_chain = prompt | llm | StrOutputParser()
    response = rag_chain.invoke({"context": docs, "question": question})
    return {"messages": [response]}

# 构建图
from langgraph.graph import END, StateGraph, START
from langgraph.prebuilt import ToolNode

workflow = StateGraph(AgentState)

workflow.add_node("agent", agent)
retrieve = ToolNode([retriever_tool])
workflow.add_node("retrieve", retrieve)
workflow.add_node("rewrite", rewrite)
workflow.add_node("generate", generate)

workflow.add_edge(START, "agent")
workflow.add_conditional_edges(
    "agent",
    tools_condition,
    {
        "tools": "retrieve",
        END: END,
    },
)

workflow.add_conditional_edges(
    "retrieve",
    grade_documents,
)
workflow.add_edge("generate", END)
workflow.add_edge("rewrite", "agent")

graph = workflow.compile()

# 运行图
import pprint

inputs = {
    "messages": [
        ("user", "What does Lilian Weng say about the types of agent memory?"),
    ]
}
for output in graph.stream(inputs):
    for key, value in output.items():
        pprint.pprint(f"Output from node '{key}':")
        pprint.pprint("---")
        pprint.pprint(value, indent=2, width=80, depth=None)
    pprint.pprint("\n---\n")
```

---

通过本教程，你可以掌握如何使用LangGraph实现一个Agentic RAG系统，并根据用户问题动态决定是否检索信息并生成回答。希望这对你的学习和开发有所帮助！

检索增强

### 代码生成与 RAG 和自我修正
AlphaCodium 提出了一种使用控制流的代码生成方法。

**核心思想**：迭代式地构建编码问题的答案。

AlphaCodium 通过迭代测试和改进答案，使用公开和 AI 生成的测试来验证特定问题的解决方案。

我们将使用 LangGraph 从头实现其中的一些想法：

1. 从用户指定的文档集开始。
2. 使用长上下文 LLM 来消化这些文档，并通过 RAG 回答问题。
3. 调用工具生成结构化输出。
4. 在将解决方案返回给用户之前，执行两个单元测试（检查导入和代码执行）。

---

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

```python
! pip install -U langchain_community langchain-openai langchain-anthropic langchain langgraph bs4

import getpass
import os

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

_set_env("OPENAI_API_KEY")
_set_env("ANTHROPIC_API_KEY")
```

---

### 文档加载
加载 LangChain Expression Language (LCEL) 文档作为示例。

```python
from bs4 import BeautifulSoup as Soup
from langchain_community.document_loaders.recursive_url_loader import RecursiveUrlLoader

# LCEL 文档
url = "https://python.langchain.com/docs/concepts/lcel/"
loader = RecursiveUrlLoader(
    url=url, max_depth=20, extractor=lambda x: Soup(x, "html.parser").text
)
docs = loader.load()

# 按 URL 排序并获取文本
d_sorted = sorted(docs, key=lambda x: x.metadata["source"])
d_reversed = list(reversed(d_sorted))
concatenated_content = "\n\n\n --- \n\n\n".join(
    [doc.page_content for doc in d_reversed]
)
```

---

### LLM 配置
我们将尝试使用 OpenAI 和 Claude3 进行代码生成。

```python
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

# OpenAI 配置
code_gen_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            """你是精通 LCEL（LangChain 表达式语言）的编码助手。\n
            以下是完整的 LCEL 文档：\n ------- \n {context} \n ------- \n
            根据提供的文档回答用户问题。确保提供的代码可以执行，并包含所有必需的导入和变量定义。
            结构化你的答案：1) 代码解决方案的描述，2) 导入语句，3) 可执行的代码块。用户问题是：""",
        ),
        ("placeholder", "{messages}"),
    ]
)

# 数据模型
class code(BaseModel):
    """LCEL 问题代码解决方案的 Schema"""
    prefix: str = Field(description="问题描述和解决方案方法")
    imports: str = Field(description="导入语句代码块")
    code: str = Field(description="不包括导入语句的代码块")

expt_llm = "gpt-4o-mini"
llm = ChatOpenAI(temperature=0, model=expt_llm)
code_gen_chain_oai = code_gen_prompt | llm.with_structured_output(code)
```

---

### 代码生成与测试
生成代码并执行单元测试。

```python
from langchain_anthropic import ChatAnthropic

# Anthropic 配置
code_gen_prompt_claude = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            """<instructions> 你是精通 LCEL 的编码助手。\n
            以下是 LCEL 文档：\n ------- \n {context} \n ------- \n
            根据提供的文档回答用户问题。确保提供的代码可以执行，并包含所有必需的导入和变量定义。
            结构化你的答案：1) 代码解决方案的描述，2) 导入语句，3) 可执行的代码块。
            调用代码工具以正确结构化输出。</instructions> \n 用户问题是：""",
        ),
        ("placeholder", "{messages}"),
    ]
)

expt_llm = "claude-3-opus-20240229"
llm = ChatAnthropic(
    model=expt_llm,
    default_headers={"anthropic-beta": "tools-2024-04-04"},
)

structured_llm_claude = llm.with_structured_output(code, include_raw=True)
```

---

### 代码检查与修正
定义代码检查和修正的逻辑。

```python
def code_check(state: GraphState):
    """检查代码"""
    print("---CHECKING CODE---")
    messages = state["messages"]
    code_solution = state["generation"]
    iterations = state["iterations"]

    # 检查导入
    try:
        exec(code_solution.imports)
    except Exception as e:
        print("---CODE IMPORT CHECK: FAILED---")
        error_message = [("user", f"导入检查失败: {e}")]
        messages += error_message
        return {"error": "yes"}

    # 检查代码执行
    try:
        exec(code_solution.imports + "\n" + code_solution.code)
    except Exception as e:
        print("---CODE BLOCK CHECK: FAILED---")
        error_message = [("user", f"代码执行检查失败: {e}")]
        messages += error_message
        return {"error": "yes"}

    print("---NO CODE TEST FAILURES---")
    return {"error": "no"}
```

---

### 完整代码
以下是完整的代码实现，包括 LangGraph 的工作流定义和测试逻辑。

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

workflow = StateGraph(GraphState)

# 定义节点
workflow.add_node("generate", generate)  # 生成解决方案
workflow.add_node("check_code", code_check)  # 检查代码
workflow.add_node("reflect", reflect)  # 反思错误

# 构建图
workflow.add_edge(START, "generate")
workflow.add_edge("generate", "check_code")
workflow.add_conditional_edges(
    "check_code",
    decide_to_finish,
    {
        "end": END,
        "reflect": "reflect",
        "generate": "generate",
    },
)
workflow.add_edge("reflect", "generate")
app = workflow.compile()
```

---

### 测试与评估
使用 LangSmith 进行测试和评估。

```python
from langsmith import Client

client = Client()

# 克隆公共数据集
try:
    public_dataset = "https://smith.langchain.com/public/326674a6-62bd-462d-88ae-eea49d503f9d/d"
    client.clone_public_dataset(public_dataset)
except:
    print("请设置 LangSmith")

# 自定义评估
def check_import(run: Run, example: Example) -> dict:
    imports = run.outputs.get("imports")
    try:
        exec(imports)
        return {"key": "import_check", "score": 1}
    except Exception:
        return {"key": "import_check", "score": 0}

def check_execution(run: Run, example: Example) -> dict:
    imports = run.outputs.get("imports")
    code = run.outputs.get("code")
    try:
        exec(imports + "\n" + code)
        return {"key": "code_execution_check", "score": 1}
    except Exception:
        return {"key": "code_execution_check", "score": 0}
```

---

### 结果
LangGraph 表现优于基础案例：添加重试循环提高了性能。  
反思并未帮助：在重试之前进行反思反而导致性能下降。  
GPT-4 优于 Claude3：Claude3 在某些运行中因工具使用错误而失败。

---

**总结**：通过 LangGraph 和 RAG，我们能够迭代生成和优化代码解决方案，确保其正确性和可执行性。

代码助手



#### 提示生成从用户需求
在这个例子中，我们将创建一个帮助用户生成提示的聊天机器人。它首先从用户那里收集需求，然后生成提示（并根据用户输入进行优化）。这些步骤被分为两个独立的状态，LLM 决定何时在它们之间进行转换。

#### 设置
首先，让我们安装所需的包并设置 OpenAI API 密钥（我们将使用的 LLM）。

```python
%%capture --no-stderr
% pip install -U langgraph langchain_openai

import getpass
import os

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

_set_env("OPENAI_API_KEY")
```

#### 设置 LangSmith 用于 LangGraph 开发
注册 LangSmith 以快速发现问题并提高 LangGraph 项目的性能。LangSmith 允许您使用跟踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用程序——阅读更多关于如何开始的信息 [这里](https://docs.smith.langchain.com/)。

#### 收集信息
首先，让我们定义图的一部分，该部分将收集用户需求。这将是一个带有特定系统消息的 LLM 调用。它将访问一个工具，当它准备好生成提示时可以调用该工具。

```python
from typing import List
from langchain_core.messages import SystemMessage
from langchain_openai import ChatOpenAI
from pydantic import BaseModel

template = """你的工作是从用户那里获取有关他们想要创建的提示模板类型的信息。

你应该从他们那里获取以下信息：

- 提示的目标是什么
- 哪些变量将传递到提示模板中
- 输出不应做什么的任何约束
- 输出必须遵守的任何要求

如果你无法辨别这些信息，请要求他们澄清！不要试图胡乱猜测。

在你能够辨别所有信息后，调用相关工具。"""

def get_messages_info(messages):
    return [SystemMessage(content=template)] + messages

class PromptInstructions(BaseModel):
    """关于如何提示 LLM 的说明。"""
    objective: str
    variables: List[str]
    constraints: List[str]
    requirements: List[str]

llm = ChatOpenAI(temperature=0)
llm_with_tool = llm.bind_tools([PromptInstructions])

def info_chain(state):
    messages = get_messages_info(state["messages"])
    response = llm_with_tool.invoke(messages)
    return {"messages": [response]}
```

#### 生成提示
我们现在设置将生成提示的状态。这将需要一个单独的系统消息，以及一个过滤掉工具调用之前所有消息的函数（因为这是前一个状态决定生成提示的时候）。

```python
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage

# 新的系统提示
prompt_system = """根据以下要求，编写一个好的提示模板：

{reqs}"""

# 获取提示消息的函数
# 只会获取工具调用之后的消息
def get_prompt_messages(messages: list):
    tool_call = None
    other_msgs = []
    for m in messages:
        if isinstance(m, AIMessage) and m.tool_calls:
            tool_call = m.tool_calls[0]["args"]
        elif isinstance(m, ToolMessage):
            continue
        elif tool_call is not None:
            other_msgs.append(m)
    return [SystemMessage(content=prompt_system.format(reqs=tool_call))] + other_msgs

def prompt_gen_chain(state):
    messages = get_prompt_messages(state["messages"])
    response = llm.invoke(messages)
    return {"messages": [response]}
```

#### 定义状态逻辑
这是聊天机器人所处状态的逻辑。如果最后一条消息是工具调用，那么我们处于“提示创建者”（prompt）应该响应的状态。否则，如果最后一条消息不是 HumanMessage，那么我们知道人类应该接下来响应，因此我们处于 END 状态。如果最后一条消息是 HumanMessage，那么如果之前有工具调用，我们处于提示状态。否则，我们处于“信息收集”（info）状态。

```python
from typing import Literal
from langgraph.graph import END

def get_state(state):
    messages = state["messages"]
    if isinstance(messages[-1], AIMessage) and messages[-1].tool_calls:
        return "add_tool_message"
    elif not isinstance(messages[-1], HumanMessage):
        return END
    return "info"
```

#### 创建图
我们现在可以创建图。我们将使用 SqliteSaver 来持久化对话历史。

```python
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from typing import Annotated
from typing_extensions import TypedDict

class State(TypedDict):
    messages: Annotated[list, add_messages]

memory = MemorySaver()
workflow = StateGraph(State)
workflow.add_node("info", info_chain)
workflow.add_node("prompt", prompt_gen_chain)

@workflow.add_node
def add_tool_message(state: State):
    return {
        "messages": [
            ToolMessage(
                content="Prompt generated!",
                tool_call_id=state["messages"][-1].tool_calls[0]["id"],
            )
        ]
    }

workflow.add_conditional_edges("info", get_state, ["add_tool_message", "info", END])
workflow.add_edge("add_tool_message", "prompt")
workflow.add_edge("prompt", END)
workflow.add_edge(START, "info")
graph = workflow.compile(checkpointer=memory)
```

#### 使用图
我们现在可以使用创建的聊天机器人。

```python
import uuid

cached_human_responses = ["hi!", "rag prompt", "1 rag, 2 none, 3 no, 4 no", "red", "q"]
cached_response_index = 0
config = {"configurable": {"thread_id": str(uuid.uuid4())}}
while True:
    try:
        user = input("User (q/Q to quit): ")
    except:
        user = cached_human_responses[cached_response_index]
        cached_response_index += 1
    print(f"User (q/Q to quit): {user}")
    if user in {"q", "Q"}:
        print("AI: Byebye")
        break
    output = None
    for output in graph.stream(
        {"messages": [HumanMessage(content=user)]}, config=config, stream_mode="updates"
    ):
        last_message = next(iter(output.values()))["messages"][-1]
        last_message.pretty_print()

    if output and "prompt" in output:
        print("Done!")
```

### 完整代码

```python
%%capture --no-stderr
% pip install -U langgraph langchain_openai

import getpass
import os

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

_set_env("OPENAI_API_KEY")

from typing import List
from langchain_core.messages import SystemMessage
from langchain_openai import ChatOpenAI
from pydantic import BaseModel

template = """你的工作是从用户那里获取有关他们想要创建的提示模板类型的信息。

你应该从他们那里获取以下信息：

- 提示的目标是什么
- 哪些变量将传递到提示模板中
- 输出不应做什么的任何约束
- 输出必须遵守的任何要求

如果你无法辨别这些信息，请要求他们澄清！不要试图胡乱猜测。

在你能够辨别所有信息后，调用相关工具。"""

def get_messages_info(messages):
    return [SystemMessage(content=template)] + messages

class PromptInstructions(BaseModel):
    """关于如何提示 LLM 的说明。"""
    objective: str
    variables: List[str]
    constraints: List[str]
    requirements: List[str]

llm = ChatOpenAI(temperature=0)
llm_with_tool = llm.bind_tools([PromptInstructions])

def info_chain(state):
    messages = get_messages_info(state["messages"])
    response = llm_with_tool.invoke(messages)
    return {"messages": [response]}

from langchain_core.messages import AIMessage, HumanMessage, ToolMessage

# 新的系统提示
prompt_system = """根据以下要求，编写一个好的提示模板：

{reqs}"""

# 获取提示消息的函数
# 只会获取工具调用之后的消息
def get_prompt_messages(messages: list):
    tool_call = None
    other_msgs = []
    for m in messages:
        if isinstance(m, AIMessage) and m.tool_calls:
            tool_call = m.tool_calls[0]["args"]
        elif isinstance(m, ToolMessage):
            continue
        elif tool_call is not None:
            other_msgs.append(m)
    return [SystemMessage(content=prompt_system.format(reqs=tool_call))] + other_msgs

def prompt_gen_chain(state):
    messages = get_prompt_messages(state["messages"])
    response = llm.invoke(messages)
    return {"messages": [response]}

from typing import Literal
from langgraph.graph import END

def get_state(state):
    messages = state["messages"]
    if isinstance(messages[-1], AIMessage) and messages[-1].tool_calls:
        return "add_tool_message"
    elif not isinstance(messages[-1], HumanMessage):
        return END
    return "info"

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from typing import Annotated
from typing_extensions import TypedDict

class State(TypedDict):
    messages: Annotated[list, add_messages]

memory = MemorySaver()
workflow = StateGraph(State)
workflow.add_node("info", info_chain)
workflow.add_node("prompt", prompt_gen_chain)

@workflow.add_node
def add_tool_message(state: State):
    return {
        "messages": [
            ToolMessage(
                content="Prompt generated!",
                tool_call_id=state["messages"][-1].tool_calls[0]["id"],
            )
        ]
    }

workflow.add_conditional_edges("info", get_state, ["add_tool_message", "info", END])
workflow.add_edge("add_tool_message", "prompt")
workflow.add_edge("prompt", END)
workflow.add_edge(START, "info")
graph = workflow.compile(checkpointer=memory)

import uuid

cached_human_responses = ["hi!", "rag prompt", "1 rag, 2 none, 3 no, 4 no", "red", "q"]
cached_response_index = 0
config = {"configurable": {"thread_id": str(uuid.uuid4())}}
while True:
    try:
        user = input("User (q/Q to quit): ")
    except:
        user = cached_human_responses[cached_response_index]
        cached_response_index += 1
    print(f"User (q/Q to quit): {user}")
    if user in {"q", "Q"}:
        print("AI: Byebye")
        break
    output = None
    for output in graph.stream(
        {"messages": [HumanMessage(content=user)]}, config=config, stream_mode="updates"
    ):
        last_message = next(iter(output.values()))["messages"][-1]
        last_message.pretty_print()

    if output and "prompt" in output:
        print("Done!")
```

这个代码实现了一个帮助用户生成提示的聊天机器人，它首先收集用户的需求，然后根据这些需求生成提示，并根据用户的反馈进行优化。

提示词生成助手

客户支持

### **模板应用**

模板是开源的参考应用，旨在帮助您快速开始使用 LangGraph 构建应用。它们提供了常见的代理工作流程的示例，可以根据您的需求进行自定义。

您可以使用 LangGraph CLI 从模板创建应用。

---

#### **环境要求**
- **Python** >= 3.11
- **LangGraph CLI**：需要 `langchain-cli[inmem]` >= 0.1.58

---

#### **安装 LangGraph CLI**
```bash
pip install "langgraph-cli[inmem]" --upgrade
```

---

#### **可用模板**
| 模板名称               | 描述                                                                 | Python 仓库 | JS/TS 仓库 |
|------------------------|----------------------------------------------------------------------|-------------|------------|
| **New LangGraph Project** | 一个简单、最小化的聊天机器人，带有记忆功能。                           | [Repo](https://github.com/langgraph/template-new-project-python) | [Repo](https://github.com/langgraph/template-new-project-js) |
| **ReAct Agent**         | 一个简单的代理，可以灵活扩展以支持多种工具。                           | [Repo](https://github.com/langgraph/template-react-agent-python) | [Repo](https://github.com/langgraph/template-react-agent-js) |
| **Memory Agent**         | 一个 ReAct 风格的代理，带有额外的工具，用于跨线程存储记忆。             | [Repo](https://github.com/langgraph/template-memory-agent-python) | [Repo](https://github.com/langgraph/template-memory-agent-js) |
| **Retrieval Agent**     | 一个包含基于检索的问答系统的代理。                                     | [Repo](https://github.com/langgraph/template-retrieval-agent-python) | [Repo](https://github.com/langgraph/template-retrieval-agent-js) |
| **Data-Enrichment Agent** | 一个执行网络搜索并将其发现组织成结构化格式的代理。                     | [Repo](https://github.com/langgraph/template-data-enrichment-agent-python) | [Repo](https://github.com/langgraph/template-data-enrichment-agent-js) |

---

#### **🌱 创建 LangGraph 应用**
要从模板创建新应用，请使用 `langgraph new` 命令：
```bash
langgraph new
```

---

#### **下一步**
1. **查看 README.md 文件**  
   在您的新 LangGraph 应用的根目录中，查看 `README.md` 文件，了解模板的详细信息以及如何自定义它。

2. **启动应用**  
   在正确配置应用并添加 API 密钥后，可以使用以下命令启动应用：
   ```bash
   langgraph dev
   ```

3. **部署应用**  
   参考以下指南部署您的应用：
   - [启动本地 LangGraph 服务器](https://langgraph.ai/quickstart)：本快速入门指南展示了如何为 ReAct Agent 模板启动本地 LangGraph 服务器。其他模板的步骤类似。
   - [部署到 LangGraph Cloud](https://langgraph.ai/deploy)：使用 LangGraph Cloud 部署您的应用。

---

#### **LangGraph 框架**
- [LangGraph 概念](https://langgraph.ai/concepts)：学习 LangGraph 的基础概念。
- [LangGraph 操作指南](https://langgraph.ai/guides)：了解使用 LangGraph 完成常见任务的指南。

---

#### **📚 了解更多关于 LangGraph 平台**
通过以下资源扩展您的知识：
- [LangGraph 平台概念](https://langgraph.ai/concepts)：了解 LangGraph 平台的基础概念。
- [LangGraph 平台操作指南](https://langgraph.ai/guides)：探索构建和部署应用的逐步指南。

---

**开始构建您的智能系统吧！🚀**