垦荒学园
  • 中文

      • 实战案例

      三个由浅入深的案例，覆盖 RAG、多工具 Agent、LangGraph 编排。

      案例 1：文档问答 Bot（RAG Agent）

      目标： 对公司 docs/ 目录问答。

      from pathlib import Path
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.tools import tool
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

# 1. 索引
loader = DirectoryLoader("docs", glob="**/*.md", loader_cls=TextLoader, loader_kwargs={"encoding": "utf-8"})
chunks = RecursiveCharacterTextSplitter(chunk_size=600, chunk_overlap=80).split_documents(loader.load())
vs = Chroma.from_documents(chunks, OpenAIEmbeddings(model="text-embedding-3-small"))
retriever = vs.as_retriever(k=5)

@tool
def search_knowledge_base(query: str) -> str:
    """Search internal markdown docs."""
    docs = retriever.invoke(query)
    return "\n\n".join(f"[{d.metadata.get('source','')}] {d.page_content[:800]}" for d in docs)

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[search_knowledge_base],
    system_prompt="Answer from search results. Cite sources. Say if unknown.",
    checkpointer=InMemorySaver(),
)

      验证： 问文档中独有的事实；观察 LangSmith 中 tool 入参/出参。

      案例 2：研究助手（多工具 + thread）

      工具： 搜索（模拟）、计算器、当前时间。

      from datetime import datetime, timezone
from langchain.tools import tool
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

@tool
def web_search(query: str) -> str:
    """Search the web (demo stub)."""
    return f"[demo] Top result for '{query}': LangChain 1.0 uses create_agent on LangGraph."

@tool
def calculate(expression: str) -> str:
    """Evaluate a math expression, e.g. (2+3)*4."""
    return str(eval(expression, {"__builtins__": {}}, {}))

@tool
def now_utc() -> str:
    """Current UTC time."""
    return datetime.now(timezone.utc).isoformat()

agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[web_search, calculate, now_utc],
    system_prompt="You are a research assistant. Use tools; show reasoning briefly.",
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "research-1"}}
agent.invoke({"messages": [{"role": "user", "content": "What is (12+8)*3? Also search LangChain 1.0."}]}, config=config)

      生产环境勿对不可信输入使用 eval；改用 numexpr 或专用数学库。

      案例 3：LangGraph 质检工作流

      流程： 生成 → 自动检查长度 → 通过则结束，否则重写。

      from typing_extensions import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4.1-mini", temperature=0.7)

class State(TypedDict):
    topic: str
    article: str
    ok: bool

def generate(state: State) -> dict:
    msg = llm.invoke(f"Write a 2-sentence article about {state['topic']}.")
    return {"article": msg.content, "ok": len(msg.content) >= 80}

def rewrite(state: State) -> dict:
    msg = llm.invoke(f"Expand this to at least 80 chars:\n{state['article']}")
    return {"article": msg.content, "ok": len(msg.content) >= 80}

def check(state: State) -> Literal["end", "rewrite"]:
    return "end" if state["ok"] else "rewrite"

graph = StateGraph(State)
graph.add_node("generate", generate)
graph.add_node("rewrite", rewrite)
graph.add_edge(START, "generate")
graph.add_conditional_edges("generate", check, {"end": END, "rewrite": "rewrite"})
graph.add_conditional_edges("rewrite", check, {"end": END, "rewrite": "rewrite"})
app = graph.compile()

print(app.invoke({"topic": "LangGraph checkpoints"})["article"])

      反模式

      反模式改进
      无 eval 上线 RAGLangSmith 数据集 + 抽样人工
      单 thread 服务所有用户每用户独立 thread_id
      仅 InMemory checkpoint 上生产Postgres + 备份

      下一步