LangChain框架深度解析：AI项目开发的技术选型与最佳实践指南

import os
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# 1. 定义工具
@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息"""
    return f"{city}今天天气晴朗，温度23°C"

# 2. 配置模型 (使用 DeepSeek)
llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    temperature=0
)

# 3. 创建智能体
agent = create_agent(
    model=llm,
    tools=[get_weather],
    system_prompt="你是一个专业的天气助手"
)

# 4. 执行查询
result = agent.invoke({
    "messages": [{"role": "user", "content": "北京今天天气怎么样？"}]
})

# 输出结果
print(result["messages"][-1].content)
# 输出：北京今天天气晴朗，温度23°C

from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI
import os

@tool
def check_order_status(order_id: str) -> str:
    """查询订单状态"""
    # 实际应用中这里会调用数据库或API
    return f"订单{order_id}已发货，预计明天送达"

@tool
def process_return(order_id: str, reason: str) -> str:
    """处理退货申请"""
    return f"订单{order_id}的退货申请已受理，原因是：{reason}"

# 配置 DeepSeek 模型
llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    temperature=0
)

service_agent = create_agent(
    model=llm,
    tools=[check_order_status, process_return],
    system_prompt="""你是一个专业的客户服务助手。
    你可以帮助客户查询订单状态和处理退货申请。
    始终保持礼貌和专业的态度。"""
)

# 客户咨询
response = service_agent.invoke({
    "messages": [{
        "role": "user",
        "content": "我的订单ORD12345发了吗？"
    }]
})
print(response["messages"][-1].content)

from typing import Dict, Any, List
import os
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from langchain_openai import ChatOpenAI

class LoggingHandler(BaseCallbackHandler):
    """
    一个简单的中间件/回调，用于记录 LLM 的交互过程
    """
    def on_llm_start(
        self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
    ) -> None:
        """LLM 开始处理请求时的回调"""
        print("\n[Middleware] LLM 开始处理请求...")

    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
        """LLM 处理完成时的回调"""
        print(f"\n[Middleware] LLM 处理完成。消耗 Token: {response.llm_output.get('token_usage', 'N/A')}")

# 使用 Callbacks 配置 DeepSeek
llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    callbacks=[LoggingHandler()]  # 注入中间件
)
# 当 Agent 使用这个 LLM 时，所有操作都会被 LoggingHandler 捕获

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware

agent = create_agent(
    model="gpt-4o-mini",
    tools=[email_tool],
    middleware=[
        # 自动屏蔽邮箱地址
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        # 完全阻止电话号码传递给模型
        PIIMiddleware(
            "phone_number",
            detector=r"(?:\+?\d{1,3}[\s.-]?)?\(?\d{2,4}\)?[\s.-]?\d{3,4}[\s.-]?\d{4}",
            strategy="block"
        ),
        # 屏蔽身份证号
        PIIMiddleware(
            "id_card",
            detector=r"\d{17}[\dXx]",
            strategy="redact"
        )
    ]
)

# 即使用户输入包含敏感信息，也会被自动处理
result = agent.invoke({
    "messages": [{"role": "user", "content": "我的邮箱是test@example.com，电话13812345678"}]
})
# 实际传递给模型的内容：我的邮箱是***，电话***

from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model="gpt-4o-mini",
    tools=[search_tool],
    middleware=[
        SummarizationMiddleware(
            model="gpt-4o-mini",
            max_tokens_before_summary=3000,  # 超过3000 tokens自动摘要
            summary_prompt="将以下对话内容摘要为关键要点"  # 自定义摘要提示词
        )
    ]
)

# 长对话自动管理
for i in range(20):
    result = agent.invoke({
        "messages": [{"role": "user", "content": f"问题{i+1}"}]
    })
    # 当对话历史过长时，会自动进行摘要压缩

from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

agent = create_agent(
    model="gpt-4o",
    tools=[delete_file, send_email, transfer_money],
    checkpointer=InMemorySaver(),  # 需要记忆管理支持
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "delete_file": {
                    "allowed_decisions": ["approve", "edit", "reject"],
                    "require_reason": True  # 拒绝时必须提供原因
                },
                "send_email": {
                    "allowed_decisions": ["approve", "reject"]
                },
                "transfer_money": {
                    "allowed_decisions": ["approve", "reject"],
                    "require_reason": True
                }
            }
        )
    ]
)

config = {"configurable": {"thread_id": "session_1"}}

result = agent.invoke(
    {"messages": [{"role": "user", "content": "删除重要文件.doc"}]},
    config
)

if "__interrupt__" in result:
    # 程序暂停，等待人工决策
    decision = input("批准删除文件？(approve/reject): ")

    if decision == "approve":
        # 继续执行
        result = agent.invoke(
            Command(resume={"decisions": [{"type": "approve"}]}),
            config
        )
    else:
        # 拒绝执行
        result = agent.invoke(
            Command(resume={
                "decisions": [{
                    "type": "reject",
                    "message": "用户取消操作"
                }]
            }),
            config
        )

from dataclasses import dataclass
from typing import Callable
from langchain.chat_models import init_chat_model
from langchain.agents.middleware import AgentMiddleware
from langchain.agents.middleware.types import ModelRequest, ModelResponse

@dataclass
class Context:
    user_id: str
    user_tier: str = "free"  # free, pro, enterprise
    request_count: int = 0

class TierBasedRoutingMiddleware(AgentMiddleware):
    """根据用户等级路由到不同的模型"""

    def __init__(self):
        super().__init__()
        # 定义不同等级使用的模型
        self.tier_models = {
            "free": "gpt-4o-mini",
            "pro": "gpt-4o",
            "enterprise": "gpt-4o-turbo"
        }

    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse]
    ) -> ModelResponse:
        # 获取用户等级
        user_tier = request.runtime.context.user_tier

        # 动态选择模型
        model_name = self.tier_models.get(user_tier, "gpt-4o-mini")
        request.model = init_chat_model(model=model_name)

        # 记录请求次数
        request.runtime.context.request_count += 1

        print(f"[中间件] 用户等级: {user_tier}, 使用模型: {model_name}")

        return handler(request)

# 应用中间件
agent = create_agent(
    model="gpt-4o-mini",  # 默认模型，会被中间件覆盖
    tools=[search_tool],
    middleware=[TierBasedRoutingMiddleware()],
    context_schema=Context
)

# 不同用户使用不同模型
result_free = agent.invoke(
    {"messages": [{"role": "user", "content": "搜索新闻"}]},
    context=Context(user_id="user_001", user_tier="free")
)

result_enterprise = agent.invoke(
    {"messages": [{"role": "user", "content": "搜索新闻"}]},
    context=Context(user_id="user_002", user_tier="enterprise")
)

from langchain.agents import create_agent
from langchain.agents.structured_output import AutoStrategy
from pydantic import BaseModel

class ProductComparison(BaseModel):
    """产品比较结果"""
    product_name: str
    price: float
    rating: float
    pros: list[str]
    cons: list[str]
    verdict: str  # 购买建议

agent = create_agent(
    model="gpt-4o",
    tools=[web_search],
    response_format=AutoStrategy(ProductComparison),
    system_prompt="你是一个产品比较专家"
)

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "比较iPhone 15和Samsung S24的优缺点"
    }]
})

# 直接获得结构化对象
comparison: ProductComparison = result["structured_response"]
print(f"产品: {comparison.product_name}")
print(f"价格: ${comparison.price}")
print(f"评分: {comparison.rating}/5")
print(f"优点: {', '.join(comparison.pros)}")
print(f"缺点: {', '.join(comparison.cons)}")
print(f"建议: {comparison.verdict}")

from langchain.agents.structured_output import ToolStrategy

class WeatherForecast(BaseModel):
    city: str
    temperature: int
    condition: str  # sunny, cloudy, rainy
    humidity: int
    wind_speed: int

agent = create_agent(
    model="gpt-4o-mini",
    tools=[get_weather_data],
    response_format=ToolStrategy(WeatherForecast)
)

# 适用于任何支持工具调用的模型
# 但依赖模型自身的结构化推理能力

from langchain.agents.structured_output import ProviderStrategy
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")

agent = create_agent(
    model=model,
    tools=[],
    response_format=ProviderStrategy(WeatherForecast)
)

# 直接使用OpenAI的结构化输出API
# 更稳定可靠，但仅限于支持此功能的提供商

import os
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    temperature=0
)

# 创建带记忆的智能体
agent = create_agent(
    model=llm,
    tools=[],
    checkpointer=InMemorySaver()  # 启用内存检查点
)

# 使用thread_id区分不同会话
config = {"configurable": {"thread_id": "user_session_123"}}

# 第一轮对话
agent.invoke(
    {"messages": [{"role": "user", "content": "我叫张三，今年25岁"}]},
    config
)

# 第二轮对话（智能体记得之前的信息）
result = agent.invoke(
    {"messages": [{"role": "user", "content": "我今年多大？"}]},
    config
)
print(result["messages"][-1].content)
# 输出：你今年25岁

from langgraph.checkpoint.postgres import PostgresSaver

# 配置PostgreSQL检查点
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:password@localhost:5432/langchain"
)

agent = create_agent(
    model="gpt-4o-mini",
    tools=[search_tool],
    checkpointer=checkpointer
)

# 即使程序重启，对话历史仍然保留
config = {"configurable": {"thread_id": "user_persistent_001"}}

# 第一次运行（程序A）
result1 = agent.invoke(
    {"messages": [{"role": "user", "content": "我喜欢编程"}]},
    config
)

# 第二次运行（程序B，重启后）
result2 = agent.invoke(
    {"messages": [{"role": "user", "content": "我有什么爱好？"}]},
    config
)
# 输出：你之前提到喜欢编程

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    """智能体状态定义"""
    # 消息历史（使用add_messages注解实现自动追加而非覆盖）
    messages: Annotated[list, add_messages]

    # 自定义字段
    user_input: str
    search_results: list[dict]
    analysis_result: str
    current_step: str  # 追踪当前执行步骤

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

def search_node(state: AgentState, config):
    """搜索节点：执行网络搜索"""
    query = state["user_input"]

    # 调用搜索工具
    results = search_web(query)

    print(f"[搜索节点] 找到 {len(results)} 条结果")

    return {
        "search_results": results,
        "current_step": "search_completed"
    }

def analysis_node(state: AgentState, config):
    """分析节点：分析搜索结果"""
    results = state["search_results"]

    # 构造分析提示词
    prompt = f"分析以下搜索结果：\n{results}"

    # 调用LLM分析
    response = llm.invoke([
        {"role": "system", "content": "你是一个信息分析专家"},
        {"role": "user", "content": prompt}
    ])

    print(f"[分析节点] 分析完成")

    return {
        "analysis_result": response.content,
        "current_step": "analysis_completed"
    }

def report_node(state: AgentState, config):
    """报告节点：生成最终报告"""
    analysis = state["analysis_result"]

    report = f"""
    分析报告
    =========
    {analysis}
    """

    print(f"[报告节点] 报告生成完毕")

    return {
        "messages": [{"role": "assistant", "content": report}],
        "current_step": "report_generated"
    }

from langgraph.graph import StateGraph, END

def should_search(state: AgentState) -> str:
    """条件边：决定是否需要搜索"""
    user_input = state["user_input"]

    # 如果用户问的是事实性问题，需要搜索
    question_words = ["什么", "哪里", "谁", "何时", "如何"]
    if any(word in user_input for word in question_words):
        return "search"

    # 否则直接分析
    return "analyze"

def has_results(state: AgentState) -> str:
    """条件边：检查搜索是否有结果"""
    if len(state["search_results"]) > 0:
        return "analyze"
    else:
        return "no_results"

from langgraph.graph import StateGraph, END

# 创建状态图
builder = StateGraph(AgentState)

# 添加节点
builder.add_node("search", search_node)
builder.add_node("analyze", analysis_node)
builder.add_node("report", report_node)
builder.add_node("no_results", lambda state: {
    "messages": [{"role": "assistant", "content": "未找到相关信息"}]
})

# 设置入口点
builder.set_entry_point("search")

# 添加边
# 从search节点出发，根据是否有结果决定下一步
builder.add_conditional_edges(
    "search",
    has_results,
    {
        "analyze": "analyze",
        "no_results": "no_results"
    }
)

# analyze完成后总是到report
builder.add_edge("analyze", "report")

# report和no_results都结束
builder.add_edge("report", END)
builder.add_edge("no_results", END)

# 编译图
graph = builder.compile()

# 执行
result = graph.invoke({
    "user_input": "什么是LangChain？",
    "messages": [],
    "search_results": [],
    "analysis_result": "",
    "current_step": ""
})

print(result["messages"][-1]["content"])

from typing import Literal

def route_question(state: AgentState) -> Literal["web_search", "database", "direct_answer"]:
    """根据问题类型路由到不同处理节点"""
    question = state["user_input"].lower()

    # 技术问题 -> 网络搜索
    if any(word in question for word in ["如何", "怎么", "教程", "技术"]):
        return "web_search"

    # 数据查询问题 -> 数据库查询
    if any(word in question for word in ["查询", "数据", "记录", "统计"]):
        return "database"

    # 简单问题 -> 直接回答
    return "direct_answer"

builder.add_conditional_edges(
    "router",
    route_question,
    {
        "web_search": "web_search_node",
        "database": "database_query_node",
        "direct_answer": "direct_answer_node"
    }
)

class IterationState(TypedDict):
    messages: Annotated[list, add_messages]
    iteration: int
    max_iterations: int
    result: str
    converged: bool

def process_step(state: IterationState):
    """处理步骤"""
    iteration = state["iteration"]
    print(f"[迭代] 第 {iteration} 次处理")

    # 模拟处理逻辑
    result = f"处理结果 {iteration}"

    # 检查是否收敛（假设5次后收敛）
    converged = iteration >= 5

    return {
        "result": result,
        "converged": converged,
        "iteration": iteration + 1
    }

def should_continue(state: IterationState) -> Literal["continue", "end"]:
    """决定是否继续迭代"""
    if state["converged"]:
        return "end"
    if state["iteration"] >= state["max_iterations"]:
        return "end"
    return "continue"

builder = StateGraph(IterationState)
builder.add_node("process", process_step)
builder.set_entry_point("process")

# 添加循环边
builder.add_conditional_edges(
    "process",
    should_continue,
    {
        "continue": "process",  # 循环回process节点
        "end": END
    }
)

graph = builder.compile()

# 执行（会循环5次直到收敛）
result = graph.invoke({
    "messages": [],
    "iteration": 1,
    "max_iterations": 10,
    "result": "",
    "converged": False
})

from langgraph.checkpoint.sqlite import SqliteSaver

# 创建SQLite检查点
checkpointer = SqliteSaver.from_conn_string("agent_state.db")

builder = StateGraph(AgentState)
# ... 添加节点和边 ...

graph = builder.compile(
    checkpointer=checkpointer  # 启用检查点
)

# 使用thread_id追踪会话
config = {"configurable": {"thread_id": "user_session_001"}}

# 第一次执行
result1 = graph.invoke({"user_input": "搜索AI新闻"}, config)

# 第二次执行（从上次停止的状态继续）
result2 = graph.invoke({"user_input": "继续搜索"}, config)

from langgraph.checkpoint.postgres import PostgresSaver
import psycopg

# 配置PostgreSQL连接
connection = psycopg.connect(
    host="localhost",
    port=5432,
    database="langchain",
    user="postgres",
    password="password"
)

# 创建检查点
checkpointer = PostgresSaver(connection)

# 初始化表（首次运行）
checkpointer.setup()

graph = builder.compile(checkpointer=checkpointer)

# 获取指定会话的所有历史状态
history = graph.get_state_history(config)

# 查看历史状态记录
for state in history:
    print(f"步骤: {state.step}, 时间: {state.timestamp}")

# 回溯到特定的历史状态
past_state = history[5]  # 回到第5步的状态
restored_result = graph.invoke(None, config)

# 从某个历史状态开始重新执行（分支）
graph.update_state(config, past_state.values)
new_result = graph.invoke({"user_input": "基于历史状态的新指令"}, config)

from langgraph.types import interrupt, Command

def human_review_node(state: AgentState):
    """需要人工审核的节点"""
    analysis = state["analysis_result"]

    # 暂停执行，等待人工输入
    feedback = interrupt({
        "type": "human_review",
        "content": analysis,
        "question": "请审核以上分析结果"
    })

    # 处理人工反馈
    if feedback["approved"]:
        return {"messages": [{"role": "assistant", "content": analysis}]}
    else:
        # 根据反馈意见修改分析结果
        revised = revise_analysis(analysis, feedback["comments"])
        return {"messages": [{"role": "assistant", "content": revised}]}

builder.add_node("human_review", human_review_node)

# 第一次执行，会在human_review节点暂停
result = graph.invoke({"user_input": "分析市场趋势"}, config)

if "__interrupt__" in result:
    # 获取需要人工审核的内容
    review_data = result["__interrupt__"]

    print("需要审核的内容：")
    print(review_data["content"])

    # 人工做出决策
    approved = input("是否批准？(y/n): ") == "y"
    comments = "" if approved else input("请提供修改意见: ")

    # 携带人工决策结果继续执行
    result = graph.invoke(
        Command(resume={
            "approved": approved,
            "comments": comments
        }),
        config
    )

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph import add_messages, StateGraph


class MultiAgentState(TypedDict):
    messages: Annotated[list, add_messages]
    current_agent: str
    research_data: dict
    code: str
    test_results: str

def researcher_agent(state: MultiAgentState):
    """研究Agent：负责收集信息"""
    print("[研究Agent] 开始研究...")

    research_data = {
        "topic": state["messages"][-1].content,
        "sources": ["source1", "source2", "source3"],
        "summary": "研究摘要..."
    }

    return {
        "research_data": research_data,
        "current_agent": "researcher"
    }

def coder_agent(state: MultiAgentState):
    """编程Agent：负责编写代码"""
    print("[编程Agent] 开始编码...")

    research = state["research_data"]
    code = f"# 基于 {research['topic']} 生成的代码\nprint('Hello World')"

    return {
        "code": code,
        "current_agent": "coder"
    }

def tester_agent(state: MultiAgentState):
    """测试Agent：负责测试代码"""
    print("[测试Agent] 开始测试...")

    code = state["code"]
    test_results = "测试通过！所有测试用例均通过。"

    return {
        "test_results": test_results,
        "current_agent": "tester"
    }

def route_to_next_agent(state: MultiAgentState) -> str:
    """根据当前Agent路由到下一个Agent"""
    current = state["current_agent"]

    if current == "researcher":
        return "coder"
    elif current == "coder":
        return "tester"
    else:
        return "end"

builder = StateGraph(MultiAgentState)
builder.add_node("researcher", researcher_agent)
builder.add_node("coder", coder_agent)
builder.add_node("tester", tester_agent)

builder.set_entry_point("researcher")

# 添加路由边
builder.add_conditional_edges(
    "researcher",
    route_to_next_agent,
    {"coder": "coder"}
)

builder.add_conditional_edges(
    "coder",
    route_to_next_agent,
    {"tester": "tester"}
)

builder.add_conditional_edges(
    "tester",
    route_to_next_agent,
    {"end": END}
)

graph = builder.compile()

# 执行多Agent协同任务
result = graph.invoke({
    "messages": [{"role": "user", "content": "开发一个待办事项应用"}],
    "current_agent": "researcher",
    "research_data": {},
    "code": "",
    "test_results": ""
})

print("\n最终结果：")
print(f"研究数据: {result['research_data']}")
print(f"代码: {result['code']}")
print(f"测试结果: {result['test_results']}")

# 定义子图
sub_builder = StateGraph(SubState)
# ... 构建子图逻辑 ...
sub_graph = sub_builder.compile()

# 在父图中将子图作为一个节点使用
parent_builder = StateGraph(ParentState)
parent_builder.add_node("sub_task", sub_graph)  # 子图直接作为父图的一个节点