LangChain 1.0 & LangGraph 1.0 代码示例手册

本手册包含可直接运行的代码示例，涵盖 LangChain 和 LangGraph 的核心功能。

基础示例

1. 模型初始化与简单调用

"""
基础示例：模型初始化与调用
"""
import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

# 加载环境变量
load_dotenv()

# 初始化模型（Groq - 免费）
model = init_chat_model(
    "groq:llama-3.3-70b-versatile",
    api_key=os.getenv("GROQ_API_KEY"),
    temperature=0.7
)

# 方式 1：纯字符串
response = model.invoke("什么是机器学习？")
print(response.content)

# 方式 2：字典列表（推荐）
messages = [
    {"role": "system", "content": "你是一个专业的 Python 导师"},
    {"role": "user", "content": "什么是装饰器？"}
]
response = model.invoke(messages)
print(response.content)

# 方式 3：消息对象
from langchain_core.messages import SystemMessage, HumanMessage

messages = [
    SystemMessage(content="你是一个专业的 Python 导师"),
    HumanMessage(content="什么是装饰器？")
]
response = model.invoke(messages)
print(response.content)

2. 提示词模板

"""
提示词模板示例
"""
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate
from langchain.chat_models import init_chat_model

model = init_chat_model("groq:llama-3.3-70b-versatile")

# 简单模板
template = PromptTemplate.from_template(
    "将以下文本翻译成{language}：\n{text}"
)
prompt = template.format(language="法语", text="Hello, how are you?")
print(prompt)

# 聊天模板
chat_template = ChatPromptTemplate.from_messages([
    ("system", "你是一个{role}，擅长{skill}"),
    ("user", "{question}")
])

messages = chat_template.format_messages(
    role="编程导师",
    skill="用简单语言解释复杂概念",
    question="什么是递归？"
)

response = model.invoke(messages)
print(response.content)

# LCEL 链式调用
chain = chat_template | model
response = chain.invoke({
    "role": "Python 导师",
    "skill="用简单语言解释",
    "question": "什么是装饰器？"
})
print(response.content)

3. 对话历史管理

"""
对话历史管理示例
关键：每次调用必须传递完整的对话历史
"""
from langchain.chat_models import init_chat_model

model = init_chat_model("groq:llama-3.3-70b-versatile")

# 初始化对话历史
conversation = []

# 第一轮
conversation.append({"role": "user", "content": "我叫张三"})
response = model.invoke(conversation)
conversation.append({"role": "assistant", "content": response.content})
print(f"AI: {response.content}")

# 第二轮（传递完整历史）
conversation.append({"role": "user", "content": "我叫什么？"})
response = model.invoke(conversation)
conversation.append({"role": "assistant", "content": response.content})
print(f"AI: {response.content}")

# 查看完整历史
print("\n完整对话历史：")
for msg in conversation:
    print(f"{msg['role']}: {msg['content']}")

工具调用示例

4. 自定义工具

"""
自定义工具示例
"""
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """
    获取指定城市的天气信息

    参数:
        city: 城市名称，如"北京"、"上海"

    返回:
        天气信息字符串
    """
    # 模拟天气数据
    weather_data = {
        "北京": "晴天，温度 15°C",
        "上海": "多云，温度 18°C",
        "广州": "小雨，温度 22°C"
    }
    return weather_data.get(city, f"未找到 {city} 的天气信息")

@tool
def calculator(operation: str, a: float, b: float) -> str:
    """
    执行数学计算

    参数:
        operation: "add", "subtract", "multiply", "divide"
        a: 第一个数字
        b: 第二个数字
    """
    operations = {
        "add": lambda x, y: x + y,
        "subtract": lambda x, y: x - y,
        "multiply": lambda x, y: x * y,
        "divide": lambda x, y: x / y if y != 0 else "错误：除数不能为零"
    }
    result = operations.get(operation, lambda x, y: "未知操作")(a, b)
    return f"{a} {operation} {b} = {result}"

# 测试工具
print(get_weather.invoke({"city": "北京"}))
print(calculator.invoke({"operation": "add", "a": 10, "b": 5}))

5. 工具绑定到模型

"""
工具绑定示例
"""
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

model = init_chat_model("groq:llama-3.3-70b-versatile")

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息"""
    weather_data = {"北京": "晴天，15°C", "上海": "多云，18°C"}
    return weather_data.get(city, f"未找到 {city} 的天气")

# 绑定工具到模型
model_with_tools = model.bind_tools([get_weather])

# AI 决定是否调用工具
response = model_with_tools.invoke("北京天气如何？")

# 检查工具调用
if response.tool_calls:
    print("AI 想调用工具：")
    for tool_call in response.tool_calls:
        print(f"  工具：{tool_call['name']}")
        print(f"  参数：{tool_call['args']}")
        
        # 执行工具
        if tool_call['name'] == "get_weather":
            result = get_weather.invoke(tool_call['args'])
            print(f"  结果：{result}")
else:
    print(f"AI 直接回答：{response.content}")

Agent 示例

6. 创建简单 Agent

"""
简单 Agent 示例
"""
import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool

load_dotenv()

model = init_chat_model("groq:llama-3.3-70b-versatile", api_key=os.getenv("GROQ_API_KEY"))

@tool
def calculator(expression: str) -> str:
    """计算数学表达式，如 '2+3' 或 '10*5'"""
    try:
        result = eval(expression)
        return f"{expression} = {result}"
    except:
        return "计算错误"

# 创建 Agent
agent = create_agent(
    model=model,
    tools=[calculator],
    system_prompt="你是一个有帮助的助手，可以使用计算器工具。"
)

# 执行
response = agent.invoke({
    "messages": [{"role": "user", "content": "25 乘以 8 等于多少？"}]
})

# 获取最终回答
final_answer = response['messages'][-1].content
print(f"最终回答：{final_answer}")

# 查看执行过程
print("\n执行过程：")
for i, msg in enumerate(response['messages'], 1):
    print(f"{i}. {msg.__class__.__name__}: {msg.content[:100]}...")

7. 多工具 Agent

"""
多工具 Agent 示例
"""
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool

model = init_chat_model("groq:llama-3.3-70b-versatile")

@tool
def search_web(query: str) -> str:
    """搜索网络获取信息"""
    # 模拟搜索
    return f"关于 '{query}' 的搜索结果..."

@tool
def get_weather(city: str) -> str:
    """获取城市天气"""
    return f"{city}：晴天，15°C"

@tool
def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        return f"{expression} = {eval(expression)}"
    except:
        return "计算错误"

# 创建多工具 Agent
agent = create_agent(
    model=model,
    tools=[search_web, get_weather, calculator],
    system_prompt="""你是一个多功能助手，可以：
1. 搜索网络获取信息
2. 查询天气
3. 进行数学计算

根据用户问题选择合适的工具。"""
)

# 测试
questions = [
    "北京天气如何？",
    "25 * 8 等于多少？",
    "搜索一下 Python 装饰器"
]

for q in questions:
    print(f"\n问题：{q}")
    response = agent.invoke({"messages": [{"role": "user", "content": q}]})
    print(f"回答：{response['messages'][-1].content}")

内存管理示例

8. 带内存的 Agent

"""
带内存的 Agent 示例
关键：使用 InMemorySaver 和 thread_id
"""
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

model = init_chat_model("groq:llama-3.3-70b-versatile")

# 创建带内存的 Agent
agent = create_agent(
    model=model,
    tools=[],
    system_prompt="你是一个有帮助的助手。",
    checkpointer=InMemorySaver()  # 添加内存
)

# 指定会话 ID
config = {"configurable": {"thread_id": "conversation-1"}}

# 第一轮
print("第一轮：")
response1 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫张三"}]},
    config=config
)
print(f"AI: {response1['messages'][-1].content}")

# 第二轮（记得第一轮）
print("\n第二轮：")
response2 = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么？"}]},
    config=config
)
print(f"AI: {response2['messages'][-1].content}")

# 第三轮
print("\n第三轮：")
response3 = agent.invoke(
    {"messages": [{"role": "user", "content": "我喜欢编程"}]},
    config=config
)
print(f"AI: {response3['messages'][-1].content}")

9. 多会话管理

"""
多会话管理示例
"""
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

model = init_chat_model("groq:llama-3.3-70b-versatile")

agent = create_agent(
    model=model,
    tools=[],
    checkpointer=InMemorySaver()
)

# Alice 的会话
config_alice = {"configurable": {"thread_id": "user_alice"}}
agent.invoke(
    {"messages": [{"role": "user", "content": "我叫 Alice"}]},
    config=config_alice
)

# Bob 的会话
config_bob = {"configurable": {"thread_id": "user_bob"}}
agent.invoke(
    {"messages": [{"role": "user", "content": "我叫 Bob"}]},
    config=config_bob
)

# 回到 Alice
response = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么？"}]},
    config=config_alice
)
print(f"Alice 的会话：{response['messages'][-1].content}")  # 应该说 Alice

# 回到 Bob
response = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么？"}]},
    config=config_bob
)
print(f"Bob 的会话：{response['messages'][-1].content}")  # 应该说 Bob

RAG 示例

10. 完整 RAG 流程

"""
RAG 完整流程示例
"""
import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain.agents import create_agent
from langchain_core.tools import tool

load_dotenv()

# 1. 文档加载
# 假设有一个知识文档
doc_content = """
LangChain 是一个用于构建 LLM 应用的框架。

核心组件：
1. Models - 语言模型接口
2. Prompts - 提示词模板
3. Chains - 链式调用
4. Agents - 智能代理
5. Memory - 记忆管理

RAG (Retrieval-Augmented Generation) 是核心应用场景之一。
"""

# 保存到临时文件
with open("temp_doc.txt", "w", encoding="utf-8") as f:
    f.write(doc_content)

loader = TextLoader("temp_doc.txt", encoding="utf-8")
documents = loader.load()

# 2. 文本分割
splitter = RecursiveCharacterTextSplitter(
    chunk_size=200,
    chunk_overlap=50
)
chunks = splitter.split_documents(documents)
print(f"分割成 {len(chunks)} 个文档块")

# 3. 向量嵌入
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)

# 4. 存入 Pinecone（需要 API Key）
# vectorstore = PineconeVectorStore.from_documents(
#     documents=chunks,
#     embedding=embeddings,
#     index_name="my-index"
# )

# 5. 创建检索工具
# @tool
# def search_kb(query: str) -> str:
#     """搜索知识库"""
#     docs = vectorstore.similarity_search(query, k=3)
#     return "\n\n".join([doc.page_content for doc in docs])

# 6. 创建 RAG Agent
# agent = create_agent(
#     model=init_chat_model("groq:llama-3.3-70b-versatile"),
#     tools=[search_kb],
#     system_prompt="使用 search_kb 检索信息，然后回答问题。"
# )

# 7. 问答
# response = agent.invoke({"messages": [{"role": "user", "content": "LangChain 有哪些核心组件？"}]})
# print(response['messages'][-1].content)

LangGraph 示例

11. 简单状态图

"""
LangGraph 简单状态图示例
"""
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 1. 定义状态
class State(TypedDict):
    messages: Annotated[list, add_messages]
    step: str

# 2. 定义节点
def node_a(state: State) -> dict:
    print("执行节点 A")
    return {"step": "A 完成"}

def node_b(state: State) -> dict:
    print("执行节点 B")
    return {"step": "B 完成"}

# 3. 创建图
graph = StateGraph(State)
graph.add_node("node_a", node_a)
graph.add_node("node_b", node_b)

# 4. 添加边
graph.add_edge(START, "node_a")
graph.add_edge("node_a", "node_b")
graph.add_edge("node_b", END)

# 5. 编译并运行
app = graph.compile()
result = app.invoke({"messages": [], "step": "开始"})
print(f"最终状态：{result['step']}")

12. 条件分支图

"""
LangGraph 条件分支示例
"""
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    value: int
    result: str

def check_value(state: State) -> dict:
    """检查值"""
    print(f"检查值：{state['value']}")
    return {}

def process_positive(state: State) -> dict:
    """处理正数"""
    return {"result": f"{state['value']} 是正数"}

def process_negative(state: State) -> dict:
    """处理负数"""
    return {"result": f"{state['value']} 是负数"}

def process_zero(state: State) -> dict:
    """处理零"""
    return {"result": "这是零"}

def route(state: State) -> Literal["positive", "negative", "zero"]:
    if state["value"] > 0:
        return "positive"
    elif state["value"] < 0:
        return "negative"
    else:
        return "zero"

# 构建图
graph = StateGraph(State)
graph.add_node("check", check_value)
graph.add_node("positive", process_positive)
graph.add_node("negative", process_negative)
graph.add_node("zero", process_zero)

graph.add_edge(START, "check")
graph.add_conditional_edges(
    "check",
    route,
    {
        "positive": "positive",
        "negative": "negative",
        "zero": "zero"
    }
)
graph.add_edge("positive", END)
graph.add_edge("negative", END)
graph.add_edge("zero", END)

app = graph.compile()

# 测试
for value in [5, -3, 0]:
    result = app.invoke({"value": value, "result": ""})
    print(f"输入 {value}：{result['result']}")

多 Agent 示例

13. 监督者模式

"""
多 Agent 监督者模式示例
"""
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

model = init_chat_model("groq:llama-3.3-70b-versatile")

# 定义状态
class TeamState(TypedDict):
    task: str
    messages: Annotated[list, add_messages]
    research_result: str
    draft: str
    final_content: str
    next_agent: str

# 监督者
def supervisor(state: TeamState) -> dict:
    print("  [监督者] 分析任务状态...")
    
    if not state.get("research_result"):
        next_agent = "researcher"
        print("  [监督者] 分配给研究员")
    elif not state.get("draft"):
        next_agent = "writer"
        print("  [监督者] 分配给作家")
    elif not state.get("final_content"):
        next_agent = "editor"
        print("  [监督者] 分配给编辑")
    else:
        next_agent = "complete"
        print("  [监督者] 任务完成")
    
    return {"next_agent": next_agent}

# 研究员
def researcher(state: TeamState) -> dict:
    print("  [研究员] 开始研究...")
    response = model.invoke([
        SystemMessage(content="你是研究员，整理关键信息。"),
        HumanMessage(content=state["task"])
    ])
    print(f"  [研究员] 完成，整理了 {len(response.content)} 字")
    return {"research_result": response.content}

# 作家
def writer(state: TeamState) -> dict:
    print("  [作家] 开始写作...")
    response = model.invoke([
        SystemMessage(content="你是作家，根据资料撰写文章。"),
        HumanMessage(content=f"主题：{state['task']}\n资料：{state['research_result']}")
    ])
    print(f"  [作家] 完成初稿，{len(response.content)} 字")
    return {"draft": response.content}

# 编辑
def editor(state: TeamState) -> dict:
    print("  [编辑] 开始审核...")
    response = model.invoke([
        SystemMessage(content="你是编辑，优化文章。"),
        HumanMessage(content=f"初稿：{state['draft']}")
    ])
    print(f"  [编辑] 完成，最终版本 {len(response.content)} 字")
    return {"final_content": response.content}

# 路由
def route_to_agent(state: TeamState) -> Literal["researcher", "writer", "editor", "complete"]:
    return state["next_agent"]

# 构建图
graph = StateGraph(TeamState)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.add_node("editor", editor)

graph.add_edge(START, "supervisor")
graph.add_conditional_edges(
    "supervisor",
    route_to_agent,
    {
        "researcher": "researcher",
        "writer": "writer",
        "editor": "editor",
        "complete": END
    }
)
graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", "supervisor")
graph.add_edge("editor", "supervisor")

app = graph.compile()

# 运行
result = app.invoke({
    "task": "写一篇关于人工智能的简短介绍",
    "messages": []
})

print("\n" + "="*50)
print("最终内容：")
print("="*50)
print(result["final_content"])

14. 协作链模式

"""
多 Agent 协作链模式示例
"""
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

model = init_chat_model("groq:llama-3.3-70b-versatile")

class ReviewState(TypedDict):
    code: str
    messages: Annotated[list, add_messages]
    security_review: str
    performance_review: str
    style_review: str
    final_report: str

def security_reviewer(state: ReviewState) -> dict:
    print("  [安全审查] 检查代码安全性...")
    response = model.invoke([
        SystemMessage(content="你是安全专家，审查代码安全性。"),
        HumanMessage(content=f"代码：\n{state['code']}")
    ])
    return {"security_review": response.content}

def performance_reviewer(state: ReviewState) -> dict:
    print("  [性能审查] 分析代码性能...")
    response = model.invoke([
        SystemMessage(content="你是性能专家，分析性能问题。"),
        HumanMessage(content=f"代码：\n{state['code']}")
    ])
    return {"performance_review": response.content}

def style_reviewer(state: ReviewState) -> dict:
    print("  [风格审查] 检查代码风格...")
    response = model.invoke([
        SystemMessage(content="你是代码风格专家，检查最佳实践。"),
        HumanMessage(content=f"代码：\n{state['code']}")
    ])
    return {"style_review": response.content}

def report_generator(state: ReviewState) -> dict:
    print("  [报告生成] 汇总审查结果...")
    response = model.invoke([
        SystemMessage(content="你是技术报告撰写者，汇总审查结果。"),
        HumanMessage(content=f"""
安全审查：{state['security_review']}
性能审查：{state['performance_review']}
风格审查：{state['style_review']}
""")
    ])
    return {"final_report": response.content}

# 构建顺序执行图
graph = StateGraph(ReviewState)
graph.add_node("security", security_reviewer)
graph.add_node("performance", performance_reviewer)
graph.add_node("style", style_reviewer)
graph.add_node("report", report_generator)

graph.add_edge(START, "security")
graph.add_edge("security", "performance")
graph.add_edge("performance", "style")
graph.add_edge("style", "report")
graph.add_edge("report", END)

app = graph.compile()

# 测试代码
test_code = '''
def get_user_data(user_id):
    query = f"SELECT * FROM users WHERE id = {user_id}"
    result = db.execute(query)
    return result
'''

result = app.invoke({"code": test_code, "messages": []})

print("\n" + "="*50)
print("代码审查报告：")
print("="*50)
print(result["final_report"])

环境配置

.env 文件

# Groq API（免费）
GROQ_API_KEY=your_groq_api_key_here

# OpenAI API（可选）
OPENAI_API_KEY=your_openai_api_key_here

# Pinecone API（RAG 需要）
PINECONE_API_KEY=your_pinecone_api_key_here

# LangSmith（监控，可选）
LANGSMITH_API_KEY=your_langsmith_api_key_here
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=langchain-study

安装依赖

# 核心依赖
pip install langchain langchain-groq langgraph python-dotenv

# 可选依赖
pip install langchain-openai langchain-huggingface langchain-pinecone
pip install pinecone-client langchain-text-splitters langchain-community
pip install langsmith pydantic

本手册整理自 LangChain1.0-Langgraph1.0-Learning 学习资料

LangChain 1.0 & LangGraph 1.0 代码示例手册

目录

基础示例

1. 模型初始化与简单调用

2. 提示词模板

3. 对话历史管理

工具调用示例

4. 自定义工具

5. 工具绑定到模型

Agent 示例

6. 创建简单 Agent

7. 多工具 Agent

内存管理示例

8. 带内存的 Agent

9. 多会话管理

RAG 示例

10. 完整 RAG 流程

LangGraph 示例

11. 简单状态图

12. 条件分支图

多 Agent 示例

13. 监督者模式

14. 协作链模式

环境配置

.env 文件

安装依赖