新版langchain1.x+RAG+多智能体协作 从入门到实战

bash# 创建虚拟环境
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate

# 一键安装核心栈
pip install -U langchain langgraph langchain-community langchain-openai chromadb python-dotenv

env# .env 配置
OPENAI_API_KEY=sk-xxxxx
# 或
DEEPSEEK_API_KEY=sk-xxxxx
TAVILY_API_KEY=tvly-xxxxx  # 多智能体搜索场景用

📂 离线准备（图书馆上架新书）
文档加载 → 切块 → 向量化 → 存入向量数据库

💬 在线问答（读者来借书）
用户提问 → 向量化 → 库里找相似内容 → 打包给大模型 → 输出答案

pythonimport os
from langchain_openai import ChatOpenAI
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# ── 1. 文档加载 ──
loader = PyPDFLoader("企业知识库.pdf")
docs = loader.load()  # 返回 List[Document]

# ── 2. 文本切分 ──
# 最佳实践: 256-512 token, overlap 20-50
splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=50)
splits = splitter.split_documents(docs)

# ── 3. 向量化 + 入库 ──
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=OpenAIEmbeddings(),
    persist_directory="./chroma_db"
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# ── 4. 构建 RAG 链 ──
llm = ChatOpenAI(
    model="qwen3-max",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
prompt = ChatPromptTemplate.from_template("""
你是一个企业知识助手。请根据以下参考资料回答问题。
如果资料中没有答案,请明确说"不知道"。

参考资料: {context}
用户问题: {question}
""")

rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

pythonfrom langchain.agents import create_agent
from langchain_openai import ChatOpenAI

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    return f"It's always sunny in {city}!"

llm = ChatOpenAI(
    model="qwen3-max",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key=os.environ.get("CHAT_OPENAI_API_KEY"),
    streaming=True,
)

agent = create_agent(
    model=llm,
    tools=[get_weather],
    system_prompt="You are a helpful assistant",
)

result = agent.invoke({"messages": [{"role": "user", "content": "what is the weather in sf"}]})
print(result["messages"][-1].content)
# 输出: It's always sunny in San Francisco!

思考(LLM Node) → 行动(可选) → 观察结果 → 再思考 → ... → 最终回答

pythonfrom langchain.agents import create_agent
from langchain_openai import ChatOpenAI

agent = create_agent(
    model=ChatOpenAI(model="gpt-4o", api_key="..."),
    tools=[weather_tool, cal_tool],
    middleware=[
        SummarizationMiddleware(
            model="gpt-4o-mini",
            trigger=("tokens", 4000),  # 4000 token 触发摘要
            keep=("messages", 20)       # 保留最后 20 条原始消息
        )
    ]
)

pythonfrom langchain.agents import create_agent
from langchain_openai import ChatOpenAI

agent = create_agent(
    model=ChatOpenAI(model="gpt-4o", api_key="..."),
    tools=[write_file_tool, execute_sql_tool, read_data_tool],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "writefile": True,   # 文件写入需要人工审批
                "executesql": True,  # SQL 执行需要审批
                "read_data": False   # 只读无需审批
            }
        )
    ]
)

pythonfrom langchain.messages import AIMessage

msg = AIMessage(...)  # 统一结构化字段，无视底层模型来源
blocks = msg.content_blocks  # 可直接遍历 reasoning、text 字段

pythonfrom pydantic import BaseModel, Field

class MeetingAction(BaseModel):
    action: str = Field(description="提取的会议行动项")
    priority: str = Field(description="优先级: high/medium/low")
    assignee: str = Field(description="负责人")

# 三种返回策略：
# ProviderStrategy: 利用模型厂商原生 JSON 格式能力（GPT 系列）
# ToolStrategy: 通过工具调用实现结构化，兼容全品类大模型
# 自动策略: 传入 Pydantic 实体，框架自动择优

pythonfrom typing import Annotated, TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_community.client.tavily import TavilyClient

# 定义状态结构
class SearchState(TypedDict):
    messages: Annotated[list, add_messages]
    user_query: str
    search_query: str
    search_results: str
    final_answer: str
    step: str

# 初始化
tavily_client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
llm = ChatOpenAI(model="gpt-4o-mini", api_key="...", temperature=0.7)

# 构建 LangGraph 工作流
workflow = StateGraph(SearchState)

# 节点1: 理解查询，生成搜索关键词
def understand_query_node(state: SearchState) -> SearchState:
    user_message = ""
    for msg in reversed(state["messages"]):
        if isinstance(msg, HumanMessage):
            user_message = msg.content
            break
    
    prompt = f"""分析用户的查询:"{user_message}"。
请完成: 1.总结用户需求 2.生成搜索关键词
格式: 理解:[总结] 搜索词:[关键词]"""
    
    response = llm.invoke([SystemMessage(content=prompt)])
    return {**state, "search_query": response.content, "step": "search"}

# 节点2: Tavily 搜索
def search_node(state: SearchState) -> SearchState:
    results = tavily_client.search(state["search_query"], max_results=5)
    return {**state, "search_results": results, "step": "answer"}

# 节点3: 生成最终答案
def answer_node(state: SearchState) -> SearchState:
    prompt = f"""根据以下搜索结果回答用户问题:
    查询: {state['user_query']}
    结果: {state['search_results']}"""
    response = llm.invoke([SystemMessage(content=prompt)])
    return {**state, "final_answer": response.content, "step": "done"}

# 组装图
workflow.add_node("understand", understand_query_node)
workflow.add_node("search", search_node)
workflow.add_node("answer", answer_node)
workflow.set_entry_point("understand")
workflow.add_edge("understand", "search")
workflow.add_edge("search", "answer")
workflow.add_edge("answer", END)

app = workflow.compile()