how to write a dynamic json output parser chain

[shaojun]

I'm creating a service, besides the content and prompt, that allows input a json sample str which for constrait the output, and output the final expecting json, the sample code:

from langchain.output_parsers import PydanticOutputParser
from langchain_core.output_parsers import JsonOutputParser
class OneContentToOneRequest(CustomUserType):
    content: str
    prompt: str
    output_format_sample_json_str: str = None

model = ChatOpenAI()
def create_prompt(request: OneContentToOneRequest):
    from pydantic import BaseModel, create_model
    import json
    output_format_sample_json = json.loads(
        request.output_format_sample_json_str)
    output_format_pydantic_model = create_model(
        'output_format_pydantic_model', **output_format_sample_json)
    output_format_pydantic_model = output_format_pydantic_model
    # Set up a parser + inject instructions into the prompt template.
    parser = PydanticOutputParser(
        pydantic_object=output_format_pydantic_model)
    return {"system_prompt": request.prompt, "format_instructions": parser.get_format_instructions(), "user_content": request.content}

prompt = ChatPromptTemplate.from_messages([
            ("system", '{system_prompt}\n{format_instructions}'),
            ("human", '{user_content}')
        ])
add_routes(
            app,
            RunnableLambda(create_prompt).with_types(
                input_type=OneContentToOneRequest) | prompt | model | how_to_write_this_dynamic_parser ,
            path="/MyService1",
        )
if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=9000)

this code obvious can't work as the last part of chain: how_to_write_this_dynamic_parser has not existed and defined yet.
could you help?

[eyurtsev]

is how_to_write_this_dynamic_parser = the parser that's instantiated inside the Runnable Lambda?

If so, you can instantiate the entire chain inside the runnable lambda and call .invoke()

def runnable_lambda(request):
   some stuff
   chain = prompt | model | parser
   response = chain.invoke(...)
   return response

[shaojun]

Then does this still support stream interface?

[eyurtsev]

No, that won't support streaming, but I think a pydantic parser will break streaming regardless.

Could you describe what the purpose of the API is (why is streaming needed/which step of the chain should be streamed?) and perhaps give an example of the payload? I can try to write an example for that use case. (I can't quite tell from the code what the goal is.)

[shaojun]

The purpose is let the caller application can customize the output (a json) structure for better process it.
like the caller input json sample: {'name':'bar', 'age':15}, and content: can u write me a person's description? The response will be a expected json

[shaojun]

seems the last part of https://python.langchain.com/docs/modules/model_io/output_parsers/quick_start#lcel has a demo of a streaming json parser

[eyurtsev]

You can do something like this for streaming support. Things might not be rendered nicely in the playground (not sure).

Alternatively, you can use APIHandler directly (see those examples) to create a runnable and then stream from it.

I don't believe that this will stream well because it uses a pydantic parser which doesn't operate on input stream. See details here: https://python.langchain.com/docs/expression_language/streaming.

Your options are to do something with stream_events endpoint instead which will yield chunks from the model.

Another option is to try to use JSONParser and then follow up with a custom parser that uses the pydantic model to parse the json once its complete.

import json

from fastapi import FastAPI
from langchain.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI
from pydantic.v1 import create_model

from langserve import CustomUserType, add_routes


class OneContentToOneRequest(CustomUserType):
    content: str
    prompt: str
    output_format_sample_json_str: str = None


model = ChatOpenAI()


async def run_extraction(request: OneContentToOneRequest):
    """Run an extraction on the given content.

    This function uses an async generator so that streaming is supposed.
    """
    output_format_sample_json = json.loads(
        request.output_format_sample_json_str)
    output_format_pydantic_model = create_model(
        'output_format_pydantic_model', **output_format_sample_json)
    output_format_pydantic_model = output_format_pydantic_model
    # Set up a parser + inject instructions into the prompt template.
    parser = PydanticOutputParser(
        pydantic_object=output_format_pydantic_model)

    input = {"system_prompt": request.prompt,
             "format_instructions": parser.get_format_instructions(),
             "user_content": request.content}
    chain = prompt | model | parser

    async for chunk in chain.astream(input):
        yield chunk


prompt = ChatPromptTemplate.from_messages([
    ("system", '{system_prompt}\n{format_instructions}'),
    ("human", '{user_content}')
])

app = FastAPI()

add_routes(
    app,
    RunnableLambda(run_extraction),
    path="/MyService1",
)
if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=9000)