LangChain 编程范式

TRANSLATE_TEMPLATE = ChatPromptTemplate.from_template(
    """Translate the text delimited by triple backticks into a style that is {style}.
    text: ```{text}```"""
)

def translate(text: str, style: str, llm: ChatDeepSeek) -> str:
    messages = TRANSLATE_TEMPLATE.format_messages(style=style, text=text)
    return llm.invoke(messages).content

class ReviewInfo(BaseModel):
    gift: bool
    delivery_days: int
    price_value: list[str]

# 注意：with_structured_output 走 tool_choice，关闭 deepseek 的 thinking 模式
struct_llm = ChatDeepSeek(
    model="deepseek-v4-flash",
    extra_body={"thinking": {"type": "disabled"}},
).with_structured_output(ReviewInfo)
gift_response = struct_llm.invoke(messages)   # 直接返回 ReviewInfo 对象

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Search for AI news ..."}]},
    stream_mode="values",
):
    latest = chunk["messages"][-1]
    if latest.content:
        print(f"{'User' if isinstance(latest, HumanMessage) else 'Agent'}: {latest.content}")
    elif latest.tool_calls:
        print(f"Calling tools: {[tc['name'] for tc in latest.tool_calls]}")

chain = (
    prompt1
    | llm
    | RunnableLambda(extract_company_name)   # 从 LLM 输出里抽公司名
    | prompt2
    | llm
)
result = chain.invoke({"product": "colorful socks"})

class Answer(BaseModel):
    summary: str
    confidence: float

agent = create_agent(
    model="deepseek-chat",
    tools=[get_weather],
    response_format=Answer,                   # LLM 自动产出结构化 Answer
    checkpointer=InMemorySaver(),             # 跨调用保留会话历史
    middleware=[
        ModelRetryMiddleware(max_retries=3),
        ToolRetryMiddleware(max_retries=2),
    ],
)

# 同一 thread_id 才能复用历史；context 会透传到 tools / middleware
result = agent.invoke(
    {"messages": [{"role": "user", "content": "What's the weather in San Francisco?"}]},
    config={"configurable": {"thread_id": str(uuid7())}},
    context=Context(user_id="user-123"),
)

# physics / math / history / computer science 四个 prompt 模板
prompt_infos = [{"name": ..., "description": ..., "prompt_template": ...}, ...]

router_chain = LLMRouterChain.from_llm(llm, router_prompt)
chain = MultiPromptChain(
    router_chain=router_chain,
    destination_chains=destination_chains,   # 4 个专家链
    default_chain=default_chain,             # 不相关问题退回 default
    verbose=True,
)
chain.invoke({"input": "什么是黑洞？"})        # → 路由到 physics

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    MessagesPlaceholder("history"),           # 历史占位
    ("human", "{input}"),
])

store = {}
def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

def chat(session_id: str, user_input: str):
    history = get_session_history(session_id)
    prompt_value = prompt.invoke({"history": history.messages, "input": user_input})
    ai_message = llm.invoke(prompt_value)
    history.add_user_message(user_input)
    history.add_ai_message(ai_message.content)
    return ai_message

memory = ConversationBufferWindowMemory(k=1)   # 只记最近 1 轮
memory.save_context({"input": "Hi"}, {"output": "what's up?"})
memory.save_context({"input": "Not much, just hanging"}, {"output": "cool"})

conversation = ConversationChain(llm=llm, memory=memory)
conversation.invoke({"input": "What is my name?"})   # 因 k=1 已丢失，LLM 不知道

@tool
def time(text: str) -> str:
    """Returns todays date, use this for any questions related to knowing todays date."""
    return str(date.today())

class WeatherInput(BaseModel):
    location: str
    units: Literal["celsius", "fahrenheit"] = "celsius"
    include_forecast: bool = False

@tool("get_weather_tool", args_schema=WeatherInput)
def get_weather(location: str, units: str = "celsius", include_forecast: bool = False) -> str:
    ...

@dataclass
class UserContext:
    user_id: str

@tool
def get_account_info(runtime: ToolRuntime[UserContext]) -> str:
    user_id = runtime.context.user_id          # 从 runtime 拿 user_id
    ...

agent = create_agent(
    model="deepseek-chat",
    tools=[get_account_info],
    context_schema=UserContext,               # 声明 context 类型
)
agent.invoke(..., context=UserContext(user_id="user123"))

def calculate(expression: str) -> str:
    return str(eval(expression))

@tool
def time(text: str) -> str:
    """Returns todays date ..."""
    return str(date.today())

TOOLS = [
    Tool(name="Calculator", func=calculate,
         description="Useful for math calculations. Input should be a valid Python expression."),
    Tool(name="Wikipedia", func=wikipedia_search,
         description="Useful for querying Wikipedia."),
    time,
]

REACT_PROMPT = PromptTemplate.from_template("""Answer the following questions as best you can. You have access to the following tools:

{tools}

Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {input}
Thought:{agent_scratchpad}""")

agent = create_react_agent(llm, TOOLS, REACT_PROMPT)
executor = AgentExecutor(
    agent=agent, tools=TOOLS, verbose=True,
    handle_parsing_errors=True,               # LLM 跳出 ReAct 格式时不直接报错
)

from langchain_experimental.agents.agent_toolkits import create_python_agent
from langchain_experimental.tools.python.tool import PythonREPLTool

agent = create_python_agent(
    llm,
    tool=PythonREPLTool(),                    # LLM 自己写 Python 并真跑
    verbose=True,
    agent_executor_kwargs={"handle_parsing_errors": True},
)
agent.invoke({"input": "Sort these customers by last name..."})

@tool
def fetch_text_from_url(url: str) -> str:
    """Fetch the document from a URL."""
    req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 ..."})
    with urllib.request.urlopen(req, timeout=120) as resp:
        return resp.read().decode("utf-8", errors="replace")

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

# thread_id 控制会话记忆；同样任务跑两次对比两者表现
agent.invoke({...}, config={"configurable": {"thread_id": "great-gatsby-lc"}})
deep_agent.invoke({...}, config={"configurable": {"thread_id": "great-gatsby-da"}})

def csv_to_documents(path: str) -> list[Document]:
    df = pd.read_csv(path)
    return [Document(page_content=row.to_string(),
                     metadata={"row": int(idx), "source": path})
            for idx, row in df.iterrows()]

index = VectorstoreIndexCreator(
    vectorstore_cls=Chroma,
    embedding=qwen_embeddings,
    vectorstore_kwargs={"client": chromadb.EphemeralClient(),
                        "collection_name": "outdoor_catalog"},
).from_documents(csv_to_documents(file))
response = index.query("List all shirts with sun protection ...", llm=llm)

db: InMemoryVectorStore = InMemoryVectorStore.from_documents(docs, embedding=qwen_embeddings)
similar_docs = db.similarity_search("Please suggest a shirt with sunblocking")

# 直接把召回片段拼进 prompt
qdocs = "".join(d.page_content for d in similar_docs)
llm.invoke(f"{qdocs} Question: {QUERY}").content

# RetrievalQA 包装版
qa = RetrievalQA.from_chain_type(llm=llm, chain_type="stuff", retriever=db.as_retriever())

gen_chain = QAGenerateChain.from_llm(llm)
new_examples = gen_chain.apply_and_parse(
    [{"doc": t} for t in data[:5]]           # 从前 5 条文档各造一道 QA
)
examples += new_examples

import langchain
langchain.debug = True                        # 打印检索和 prompt 细节
result = qa.run(examples[0]["query"])         # 跑第一条 QA 看效果

loader = PyMuPDF4LLMLoader("./ArangoDB-GraphCourse_Beginners.pdf")
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1500, chunk_overlap=150)
splits = text_splitter.split_documents(loader.load())      # → 1305 个 chunk

Chroma.from_documents(
    documents=splits,
    persist_directory="./store/chroma",
    embedding=QwenEmbeddings(),
    collection_name="arangodb_pdf",
)

class ChromaStructuredQueryTranslator(Visitor):
    """把 StructuredQuery 翻译成 Chroma 的 where 语法。
    langchain-classic 1.0.x 仍依赖 langchain_community 提供 ChromaTranslator,
    但 langchain_community 与 langchain>=1.0.0 存在死锁,自实现一份。
    """
    def visit_comparison(self, comparison):
        return {comparison.attribute:
                {self._map_comparator(comparison.comparator): comparison.value}}
    def visit_structured_query(self, query):
        return query.query, {"filter": query.filter.accept(self) if query.filter else None}

retriever = SelfQueryRetriever.from_llm(
    llm=deepseek_llm,
    vectorstore=chroma_db,
    document_contents="AQL 查询语言、图遍历、Graph 概念...",
    metadata_field_info=[AttributeInfo(name="page", type="integer", ...), ...],
    structured_query_translator=ChromaStructuredQueryTranslator(),
    verbose=True,
)

zh_extract_prompt = PromptTemplate(
    template="根据下面的问题，从上下文中**原样**提取与问题相关的内容片段。"
             "如果没有相关内容，返回 NO_OUTPUT。... 问题: {question} 上下文: {context}",
    output_parser=NoOutputParser(),
)
compressor = LLMChainExtractor.from_llm(llm=deepseek_llm, prompt=zh_extract_prompt)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=base_retriever,
)

qa_chain = RetrievalQA.from_chain_type(
    ds_model,
    retriever=vectordb.as_retriever(search_kwargs={"k": 8}),
    return_source_documents=True,
    chain_type_kwargs={
        "question_prompt": question_prompt,    # map 阶段：{question}+{context} → 摘要
        "combine_prompt": combine_prompt,      # reduce 阶段：{summaries}+{question} → 最终答案
    },
    chain_type="map_reduce",
)

memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
qa = ConversationalRetrievalChain.from_llm(ds_model, retriever=vectordb.as_retriever(), memory=memory)
qa.invoke({"question": "How many path finding methods in AQL?"})
qa.invoke({"question": "How to write full path AQL?"})      # 能复用历史
qa.invoke({"question": "what's my first question in the conversation?"})
# 已知问题：该版本不会回头看历史来回答"我第一个问题是什么"

class QwenEmbeddings(Embeddings):
    def __init__(self, model: str = "text-embedding-v3"):
        self._client = OpenAI(
            api_key=os.getenv("QWEN_API_KEY"),
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
        )
        self.model = model

    def embed_query(self, text):     return self._embed(text)
    def embed_documents(self, texts):
        BATCH_SIZE = 10                              # Qwen 限制
        return [e for i in range(0, len(texts), BATCH_SIZE)
                  for e in self._embed_batch(texts[i:i+BATCH_SIZE])]

# 0.90 vs 0.47 —— "i like dogs" 与 "i love canines" 远高于与 "the weather is ugly" 的相似度
print(np.dot(embeddings1, embeddings2))