langgraph.errors.InvalidUpdateError: Invalid update for channel medical with values

[xtu-xiaoc]

Checked other resources

• I added a very descriptive title to this issue.
• I searched the LangGraph/LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.
• I am sure that this is a bug in LangGraph/LangChain rather than my code.
• I am sure this is better as an issue rather than a GitHub discussion, since this is a LangGraph bug and not a design question.

Example Code

Here is mine workflow:
def workflow(one:dict):
    workflow = StateGraph(GraphState)
    # Define the nodes
    workflow.add_node("generate_new_extraction", generate_new_extraction)  

    workflow.add_node("medical_mentor", medical_mentor)  # check imports

    workflow.add_node("medical_mentor_pass", medical_mentor_pass)  # 用来过滤事件抽取的agent


    # Build graph
    workflow.set_entry_point("generate_new_extraction")

    workflow.add_edge("generate_new_extraction", "medical_mentor")
    workflow.add_edge("medical_mentor", "medical_mentor_pass")
    workflow.add_edge("medical_mentor_pass", END)

    workflow.add_conditional_edges(
        "medical_mentor",
        decide_to_finish,
        {
            "medical_mentor_pass": "medical_mentor_pass",
            "generate_new_extraction": "generate_new_extraction",
        },
    )
    app = workflow.compile()
    sss = app.invoke({"medical":{'dialogue':dialogue,'extrction':extrction,'medical_record':medical_record,'medical_iterations':0}})
  
generate_new_extraction return :
return {
        "medical": {"new_generation": medical_solution,'dialogue':dialogue,'extrction':extrction,'medical_record':medical_record,'medical_iterations':medical_iter_num}
    }

medical_mentor return:
return {
        "medical": {
            "feedback": mentor['feedback'],
            "correct": mentor['correct'],
            "new_generation": generate_new_extraction_1,
            "dialogue": dialogue,
            "extrction": extrction,
            "medical_record": medical_record,
            "medical_iterations": medical_iter_num,
        }
    }

medical_mentor_pass return :
return {
        "medical": {
            "feedback":'完成过滤',
            "correct": True,
            "new_generation": str(mentor.content),
            "dialogue": dialogue,
            "extrction": extrction,
            "medical_record": medical_record,
            "medical_iterations": medical_iter_num,
        }
    }


Where the error is reported:
for chan, vals in pending_writes_by_channel.items():
        if chan in channels:
            try:
                
                # if len(vals) > 1:
                #     vals = [vals[-1]]
                # pdb.set_trace()
                # if chan== 'medical' and len(vals)>1:
                #     vals=  [vals[-1]]
                channels[chan].update(vals)

            except InvalidUpdateError as e:
                
                pdb.set_trace()
                raise InvalidUpdateError(
                    f"Invalid update for channel {chan} with values {vals}"
                ) from e
            checkpoint["channel_versions"][chan] = max_version + 1
            updated_channels.add(chan)


Cause of error:langgraph.errors.InvalidUpdateError: Invalid update for channel medical with values

Error Message and Stack Trace (if applicable)

langgraph.errors.InvalidUpdateError: Invalid update for channel medical with values
 File "/Users/wushiqi/anaconda3/envs/01bot/lib/python3.9/site-packages/langgraph/pregel/__init__.py", line 978, in stream
    _apply_writes(checkpoint, channels, pending_writes)
  File "/Users/wushiqi/anaconda3/envs/01bot/lib/python3.9/site-packages/langgraph/pregel/__init__.py", line 1600, in _apply_writes
    raise InvalidUpdateError(
langgraph.errors.InvalidUpdateError: Invalid update for channel medical with values [{'new_generation': '\n{\n  "症状体征": [\n    {\n      "名称": "咳嗽",\n      "持续时间": "两天",\n      "详细描述": "患儿持续咳嗽，有痰，白天晚上都会咳，但咳嗽不算厉害。",\n      "严重程度": "轻微",\n      "频率": "频繁",\n      "部位": "呼吸道"\n    }\n  ],\n  "检查检验": [],\n  "药品": [\n    {\n      "名称": "消炎药",\n      "剂量": "根据医生建议",\n      "详细描述": "用于治疗患儿的咳嗽症状",\n      "频次": "每日按照医嘱"\n    },\n    {\n      "名称": "化痰止咳药",\n      "剂量": "根据医生建议",\n      "详细描述": "帮助患儿缓解咳嗽和排痰",\n      "频次": "每日按照医嘱"\n    }\n  ],\n  "手术": [],\n  "诊断": [\n    {\n      "名称": "咳嗽待查",\n      "详细描述": "根据患儿的症状和体征，待进一步检查和诊断",\n      "持续时间": "待查"\n    }\n  ]\n}\n', 'dialogue': '医生:你好，家长宝宝的咳嗽情况怎么样？多大的孩子\n患者:3岁了，不怎么咳了，有痰\n医生:白天咳还是晚上咳啊\n医生:没有鼻塞流鼻涕打喷嚏呀\n患者:没有，白天晚上都会\n医生:雾化不是公用的呀，都有消毒的，这个不要担心。\n患者:咳的不算厉害\n患者:能消毒干净啊，镇上的卫生院\n医生:咳嗽有痰没有鼻塞流鼻涕打喷嚏的情况考虑是支气管炎，痰是炎症的分泌物。\n患者:现在一般不都是一次性管子了吗，怎么还有这种公用的\n医生:医院应该消毒就是干净的呀。\n医生:现在一般的都是自己买一个，如果没有的话，公用的，也会进行消毒。\n患者:卫生院也算是医院吧\n医生:嗯嗯\n患者:那里没有自己买的\n医生:那就没有关系啊，他们肯定会消毒的，只要是看病的地方消毒都会做到的。\n医生:不要担心。\n患者:所以我听到心里都难受的不行，就怕不干净，得什么传染病\n医生:理解\n医生:还没有听说雾化有过传染，那确实嘛，以前是集体消毒，现在都是每人买一个。\n患者:好的，谢谢医生\n医生:那现在咳嗽有痰，加强保暖，多喝水，勤拍背。\n医生:可以口服一点消炎药和化痰止咳药。', 'extrction': '{\n  "症状体征": [\n    {\n      "名称": "咳嗽",\n      "持续时间": "两天",\n      "详细描述": "患儿有咳嗽症状，有痰，白天晚上都会咳，但不算厉害。",\n      "严重程度": "不算厉害"\n    }\n  ],\n  "检查检验": [],\n  "药品": [\n    {\n      "名称": "消炎药",\n      "详细描述": "口服消炎药，用于治疗患儿的咳嗽症状。",\n      "剂量": "未知",\n      "频次": "未知"\n    },\n    {\n      "名称": "化痰止咳药",\n      "详细描述": "口服化痰止咳药，帮助患儿缓解咳嗽和排痰。",\n      "剂量": "未知",\n      "频次": "未知"\n    }\n  ],\n  "手术": [],\n  "诊断": [\n    {\n      "名称": "支气管炎",\n      "详细描述": "根据患儿的症状，医生初步诊断为支气管炎。",\n      "持续时间": "未知"\n    }\n  ]\n}', 'medical_record': '主诉:咳嗽两天。\n现病史:患儿咳嗽两天，做过雾化。\n辅助检查:暂缺。\n既往史:不详。\n诊断:咳嗽待查。\n建议:消炎药，化痰止咳药，保暖，多喝水，勤拍背。', 'medical_iterations': 2}, {'feedback': '完成过滤', 'correct': True, 'new_generation': '事件抽取：\n\n{\n  "症状体征": [\n    {\n      "名称": "咳嗽",\n      "持续时间": "两天",\n      "详细描述": "患儿持续咳嗽，有痰，白天晚上都会咳，但咳嗽不算厉害。",\n      "严重程度": "轻微",\n      "频率": "频繁",\n      "部位": "呼吸道"\n    }\n  ],\n  "检查检验": [],\n  "药品": [\n    {\n      "名称": "消炎药",\n      "剂量": "未知",\n      "详细描述": "用于治疗患儿的咳嗽症状",\n      "频次": "未知"\n    },\n    {\n      "名称": "化痰止咳药",\n      "剂量": "未知",\n      "详细描述": "帮助患儿缓解咳嗽和排痰",\n      "频次": "未知"\n    }\n  ],\n  "手术": [],\n  "诊断": [\n    {\n      "名称": "支气管炎",\n      "详细描述": "根据患儿的症状和体征，医生诊断为支气管炎",\n      "持续时间": "未知（急性或慢性待查）"\n    }\n  ]\n}\n\n\n是否修改，原因：\n- 修改了“症状体征”中的“颜色”字段，因为对话中没有提及痰的颜色，所以将其删除。\n- 其他字段保持不变，因为它们都是根据对话和电子病历中的信息准确提取的。', 'dialogue': '医生:你好，家长宝宝的咳嗽情况怎么样？多大的孩子\n患者:3岁了，不怎么咳了，有痰\n医生:白天咳还是晚上咳啊\n医生:没有鼻塞流鼻涕打喷嚏呀\n患者:没有，白天晚上都会\n医生:雾化不是公用的呀，都有消毒的，这个不要担心。\n患者:咳的不算厉害\n患者:能消毒干净啊，镇上的卫生院\n医生:咳嗽有痰没有鼻塞流鼻涕打喷嚏的情况考虑是支气管炎，痰是炎症的分泌物。\n患者:现在一般不都是一次性管子了吗，怎么还有这种公用的\n医生:医院应该消毒就是干净的呀。\n医生:现在一般的都是自己买一个，如果没有的话，公用的，也会进行消毒。\n患者:卫生院也算是医院吧\n医生:嗯嗯\n患者:那里没有自己买的\n医生:那就没有关系啊，他们肯定会消毒的，只要是看病的地方消毒都会做到的。\n医生:不要担心。\n患者:所以我听到心里都难受的不行，就怕不干净，得什么传染病\n医生:理解\n医生:还没有听说雾化有过传染，那确实嘛，以前是集体消毒，现在都是每人买一个。\n患者:好的，谢谢医生\n医生:那现在咳嗽有痰，加强保暖，多喝水，勤拍背。\n医生:可以口服一点消炎药和化痰止咳药。', 'extrction': '{\n  "症状体征": [\n    {\n      "名称": "咳嗽",\n      "持续时间": "两天",\n      "详细描述": "患儿有咳嗽症状，有痰，白天晚上都会咳，但不算厉害。",\n      "严重程度": "不算厉害"\n    }\n  ],\n  "检查检验": [],\n  "药品": [\n    {\n      "名称": "消炎药",\n      "详细描述": "口服消炎药，用于治疗患儿的咳嗽症状。",\n      "剂量": "未知",\n      "频次": "未知"\n    },\n    {\n      "名称": "化痰止咳药",\n      "详细描述": "口服化痰止咳药，帮助患儿缓解咳嗽和排痰。",\n      "剂量": "未知",\n      "频次": "未知"\n    }\n  ],\n  "手术": [],\n  "诊断": [\n    {\n      "名称": "支气管炎",\n      "详细描述": "根据患儿的症状，医生初步诊断为支气管炎。",\n      "持续时间": "未知"\n    }\n  ]\n}', 'medical_record': '主诉:咳嗽两天。\n现病史:患儿咳嗽两天，做过雾化。\n辅助检查:暂缺。\n既往史:不详。\n诊断:咳嗽待查。\n建议:消炎药，化痰止咳药，保暖，多喝水，勤拍背。', 'medical_iterations': 1}]

Description

I did some tests: I found that when chan='medical', after completing the mentor_pass task, len(vals)==2, which is the reason for the error.
If I force the switch

if chan== 'medical' and len(vals)>1:

vals= [vals[-1]]

It can run, but it will always loop in medical_mentor and medical_mentor_pass and cannot jump out of end. Why is this?

System Info

langgraph==0.0.66

[hinthornw]

Hello @xtu-xiaoc - could you share what GraphState looks like here?

[beschmitt]

Has there been an answer to this? I have the same problem with the following GraphSate:

class GraphState(TypedDict):
    """
    Represents the state of our graph.

    Attributes:
        question: question
        generation: LLM generation
        web_search: whether to add search
        documents: list of documents
    """

    language: str
    question: str
    question_multi_query: List[str]
    generation: str
    web_search: str
    question_relevant: str
    documents: Annotated[List[str], operator.add]
    is_question: str
    memory: ChatMessageHistory
    is_retrieval_necessary: str
    transformed_query: str
    is_question_relevant_or_appropriate: str

I have one conditional branch and the error message always displays that there was an invalid update for the first element of GraphState. So I'm assuming that a step is missing in the documentation where you join the two different states of GraphState back to one when the branch converges.

[Glinte]

try making a reducer for every single key

[beschmitt]

I can try but in the documentation it says that if you don't make a reducer for a key it assumes that every new entry should overwrite the old one.
Which operator would I use for overwriting?

[Glinte]

I am >80% sure it will work. There is something about fan-ins that breaks default override behavior that I cannot explain in one sentence.

[Glinte]

You can just use something as dumb as
lambda x, y: y

[beschmitt]

Okay, creating a reducer for every key with your lambda-function worked. Thank you very much!

[Glinte]

@hinthornw I wonder if it is intended that fan-ins causes InvalidUpdateError when no reducer is provided. Basically the fan-in node receives everything at once and the type changes from T to list[T]

[beschmitt]

I have a feeling that the fan-in step is generally a bit buggy. I have three nodes that converge now. One value in my state dict is a boolean and one of the nodes sets that boolean. After the nodes converge that value gets set to None because it isn't set in the other two nodes.
Would be really useful to have a way to mark values coming from a specific node as priority for overwriting.

[Glinte]

that is just stable sorting no?

[Glinte]

you can also just... not rely on overriding and write a simple reducer

[beschmitt]

I tried writing my own reducer but it never worked. I wrote a function that accepted three values and then returned the value that wasn't None.
Then I passed this function into the state dict like this: Annotated[str, custom_function]
I got it working another way now but it just feels a bit buggy or like the documentation should be improved.

[Glinte]

a function that accepts three values is just wrong. try lambda x, y: y if y is not None else x

[beschmitt]

I wrote that function in the exact same logic as if you used operator.add, which takes two values (a, b) and returns a + b