计划执行与检查节点
executing-plans
按既定实施计划完成工作,保持过程检查和反馈。
langchain-core-workflow-a【效率与协作】Build LangChain chains and prompts for structured LLM workflows. Use when creating prompt templates, building LCEL chains, or implementing sequential processing pipelines. Trigger with phrases like "langchain chains", "langchain prompts", "LCEL workflow", "langchain pipeline", "prompt template".
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
Build production-ready chains using LangChain Expression Language (LCEL) with prompt templates, output parsers, and composition patterns.
langchain-install-auth setupfrom langchain_core.prompts import (
ChatPromptTemplate,
SystemMessagePromptTemplate,
HumanMessagePromptTemplate,
MessagesPlaceholder
)
# Simple template
simple_prompt = ChatPromptTemplate.from_template(
"Translate '{text}' to {language}"
)
# Chat-style template
chat_prompt = ChatPromptTemplate.from_messages([
SystemMessagePromptTemplate.from_template(
"You are a {role}. Respond in {style} style."
),
MessagesPlaceholder(variable_name="history", optional=True),
HumanMessagePromptTemplate.from_template("{input}")
])
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
llm = ChatOpenAI(model="gpt-4o-mini")
# Basic chain: prompt -> llm -> parser
basic_chain = simple_prompt | llm | StrOutputParser()
# Invoke the chain
result = basic_chain.invoke({
"text": "Hello, world!",
"language": "Spanish"
})
print(result) # "Hola, mundo!"
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
# Sequential chain
chain1 = prompt1 | llm | StrOutputParser()
chain2 = prompt2 | llm | StrOutputParser()
sequential = chain1 | (lambda x: {"summary": x}) | chain2
# Parallel execution
parallel = RunnableParallel(
summary=prompt1 | llm | StrOutputParser(),
keywords=prompt2 | llm | StrOutputParser(),
sentiment=prompt3 | llm | StrOutputParser()
)
results = parallel.invoke({"text": "Your input text"})
# Returns: {"summary": "...", "keywords": "...", "sentiment": "..."}
from langchain_core.runnables import RunnableBranch
# Conditional branching
branch = RunnableBranch(
(lambda x: x["type"] == "question", question_chain),
(lambda x: x["type"] == "command", command_chain),
default_chain # Fallback
)
result = branch.invoke({"type": "question", "input": "What is AI?"})
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
llm = ChatOpenAI(model="gpt-4o-mini")
# Step 1: Extract key points
extract_prompt = ChatPromptTemplate.from_template(
"Extract 3 key points from: {text}"
)
# Step 2: Summarize
summarize_prompt = ChatPromptTemplate.from_template(
"Create a one-sentence summary from these points: {points}"
)
# Compose the chain
chain = (
{"points": extract_prompt | llm | StrOutputParser()}
| summarize_prompt
| llm
| StrOutputParser()
)
summary = chain.invoke({"text": "Long article text here..."})
from langchain_core.runnables import RunnablePassthrough
def get_context(input_dict):
"""Fetch relevant context from database."""
return f"Context for: {input_dict['query']}"
chain = (
RunnablePassthrough.assign(context=get_context)
| prompt
| llm
| StrOutputParser()
)
result = chain.invoke({"query": "user question"})
| Error | Cause | Solution |
|---|---|---|
| Missing Variable | Template variable not provided | Check input dict keys match template |
| Type Error | Wrong input type | Ensure inputs match expected schema |
| Parse Error | Output doesn't match parser | Use more specific prompts or fallback |
Proceed to langchain-core-workflow-b for agents and tools workflow.