LangChain RunnableSequence — input schema mismatch between steps

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "You answer briefly."),
    ("human", "{question}"),
])

# ❌ WRONG — bare string flows into prompt that expects a dict
# chain = prompt | model | StrOutputParser()
# chain.invoke("what is the capital of France")  # KeyError: 'question'

# ✓ RIGHT — shape the input first
chain = (
    RunnableLambda(lambda x: {"question": x} if isinstance(x, str) else x)
    | prompt
    | model
    | StrOutputParser()
)
print(chain.invoke("what is the capital of France"))
# > Paris

# ✓ ALTERNATIVE — call invoke with a dict directly
chain2 = prompt | model | StrOutputParser()
print(chain2.invoke({"question": "what is the capital of France"}))

# ✓ BEST for retrieval workflows — RunnablePassthrough.assign
retrieval_chain = (
    RunnablePassthrough.assign(
        # add 'context' key derived from 'question' key, keeping 'question'
        context=lambda x: retrieve_docs(x["question"])
    )
    | prompt_with_context
    | model
    | StrOutputParser()
)

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_openai import ChatOpenAI

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

# --- Multi-stage chain: research → refine ---

research_prompt = ChatPromptTemplate.from_template(
    "Research: {topic}. List 5 key facts."
)
refine_prompt = ChatPromptTemplate.from_template(
    "Given these research notes, write a 2-paragraph brief:\n\n{notes}"
)

# ❌ WRONG — research emits AIMessage, refine_prompt.format wants a string
# chain = research_prompt | model | (lambda ai: {"notes": ai}) | refine_prompt | model
# TypeError: expected str, got AIMessage

# ✓ RIGHT — StrOutputParser between the two model calls
chain = (
    research_prompt
    | model
    | StrOutputParser()                            # AIMessage -> str
    | (lambda notes: {"notes": notes})             # str -> dict for next prompt
    | refine_prompt
    | model
    | StrOutputParser()                            # final AIMessage -> str
)

result = chain.invoke({"topic": "carbon capture"})
print(result)

# --- For structured output, use JsonOutputParser or with_structured_output ---
structured_chain = (
    ChatPromptTemplate.from_template("Extract as JSON: {text}")
    | model
    | JsonOutputParser()                           # AIMessage -> dict
)

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template("Explain {concept} to a {audience}")
model = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | model | StrOutputParser()

# 1) What does the WHOLE chain expect / emit?
print(chain.input_schema.model_json_schema())
# {'title': 'PromptInput', 'type': 'object',
#  'properties': {'concept': {'type': 'string'}, 'audience': {'type': 'string'}},
#  'required': ['concept', 'audience']}

print(chain.output_schema.model_json_schema())
# {'type': 'string', 'title': 'StrOutputParserOutput'}

# 2) What does each STEP in the chain expect / emit?
graph = chain.get_graph()
print(graph.draw_ascii())          # visual pipeline
print(graph.draw_mermaid())        # mermaid syntax for docs

# 3) Try invoking with the schema in mind
chain.invoke({"concept": "gradient descent", "audience": "10-year-old"})

# 4) When a chain fails, isolate the failing hop
# Invoke each prefix separately:
step1 = prompt.invoke({"concept": "x", "audience": "y"})
print(type(step1))  # ChatPromptValue

step2 = model.invoke(step1)
print(type(step2))  # AIMessage

step3 = StrOutputParser().invoke(step2)
print(type(step3))  # str