잘 돌고 있는데 멈춘 것처럼 보인다

목표

같은 그래프를 다섯 가지 방식으로 흘려보내며 무엇이 언제 나오는지 직접 센다. 갱신만 모아 마지막 상태를 다시 세워 보고, 흘려보낸 마지막 상태가 invoke 의 답과 같은지 대조한다.

왜 중요한가

에이전트에서 흘려보낼 것은 토큰만이 아니다. 모델을 마지막에 한 번 부르는 그래프라면 토큰 스트리밍을 붙여도 앞의 침묵은 그대로다. 사용자에게 필요한 것은 "지금 어느 단계인지" 와 "무엇이 새로 정해졌는지" 이고, 그것은 모델이 아니라 그래프가 알고 있다.
모드마다 단위가 다르다. updates 는 노드 단위라 나란히 도는 노드 둘이 같은 수퍼스텝에 있어도 이벤트가 따로 나오고, values 는 수퍼스텝 단위라 그 둘의 결과가 합쳐진 뒤 한 번 나온다. debug 는 수퍼스텝 번호까지 알려 준다. 이 차이를 모르면 "왜 두 번 나오지" 를 버그로 오해한다.
갱신만 모아 상태를 다시 세우려면 리듀서를 알아야 한다. 그래프 안에서는 리듀서가 해 주던 일을 바깥에서는 직접 해야 하고, 이어 붙이는 열쇠를 덮어쓰면 발자국이 한 칸만 남는다.
그리고 상태에 남기고 싶지 않은 진행 표시는 custom 으로 내보낸다. 상태에 넣으면 체크포인트에 실리고 재개할 때 옛 표시가 되살아난다.
채점기는 여러분이 적어 둔 설명을 믿지 않는다. 여러분의 모듈을 실제로 불러 매번 다른 회의록으로 돌려 보고, 이벤트의 개수·순서·모양을 채점기가 따로 센 값과 대조한다. 시간은 재지 않는다.

단계

1. /root/work/agstream/stream.py 에 State·노드 넷(split·keypoints·actions·compose)·build_graph()·start(text)·run_values(text) 를 만드세요. split 다음에 keypoints 와 actions 가 나란히 돌고, 둘 다 compose 로 모입니다.
2. run_updates(text) 를 더해 stream_mode="updates" 의 이벤트를 그대로 목록으로 돌려주게 하세요.
3. REDUCERS·rebuild(text)·rebuild_report(text) 를 더해 갱신만 모아 마지막 상태를 다시 세우고, 그것이 values 의 마지막과 같은지 대조하게 하세요.
4. run_debug(text) 를 더해 {수퍼스텝번호: [노드이름...정렬됨]} 을 돌려주게 하세요.
5. run_multi(text) 와 mode_sequence(text) 를 더해 stream_mode=["updates", "values"] 의 결과를 다루게 하세요.
6. compose 가 writer: StreamWriter 를 받아 조각 두 개를 내보내게 하고, run_custom(text) 를 더하세요.
7. compare_invoke(text) 를 더해 흘려보낸 마지막 상태와 invoke 의 답이 같은지, 가장 먼저 도착하는 노드가 무엇인지, 이벤트가 몇 개인지를 한 번에 돌려주게 하세요.
8. /root/work/agstream/minutes.txt 에 잴 회의록을 저장하고, 그것으로 재어 /root/work/agstream/stream_report.json 과 /root/work/agstream/stream_report.md 에 기록하세요.

참고

• 실행 계약: 채점기는 /root/work/agstream/stream.py 를 파이썬 모듈로 불러 위에 적은 이름들을 직접 씁니다. 스크립트로 실행하지 않습니다.
• 상태 열쇠: text·lines·points·todos·summary·trace. trace 만 이어 붙이는 리듀서를 쓰고 나머지는 덮어씁니다. 노드 이름과 상태 열쇠는 겹치면 안 됩니다 — 겹치면 컴파일할 때 ValueError: 'x' is already being used as a state key 가 납니다.
• 노드가 하는 일: split 은 text 를 줄로 나눠 공백 줄을 버리고 lines 에 담습니다. keypoints 는 "결정" 이 든 줄을 points 에, actions 는 "하기로" 가 든 줄을 todos 에 담습니다. compose 는 summary 에 "결정 N건 · 할 일 M건" 을 담습니다. 네 노드 모두 trace 에 자기 이름을 씁니다.
• start(text) 는 {"text": text, "trace": []} 를 돌려줍니다.
• run_values·run_updates·run_multi·run_custom 은 stream(...) 이 준 것을 가공하지 말고 그대로 목록으로 돌려줍니다.
• rebuild 는 start(text) 에서 시작해 updates 이벤트를 차례로 적용한 딕셔너리를 돌려줍니다. REDUCERS 는 {"trace": "append"} 처럼 이어 붙이는 열쇠를 적어 둔 표입니다.
• rebuild_report(text) 의 답: {"same": 참거짓, "stream_order": [...], "merged_order": [...]}. stream_order 는 재구성한 상태의 trace, merged_order 는 values 마지막 이벤트의 trace 입니다. 둘이 같지 않습니다 — 왜 다른지는 직접 보고 적으세요.
• run_debug 는 event["type"] == "task" 인 것만 보고 event["step"] 과 event["payload"]["name"] 을 모읍니다. 값은 정렬합니다.
• mode_sequence(text) 는 run_multi 의 튜플에서 모드 이름만 순서대로 뽑습니다.
• compare_invoke(text) 의 답: {"same": 참거짓, "first_node": 문자열, "values_events": 정수, "updates_events": 정수}.
• 8단계의 /root/work/agstream/minutes.txt 는 보고서를 만든 회의록입니다. 채점기가 그 파일을 읽어 같은 입력으로 다시 재므로, 보고서를 만든 뒤에 내용을 바꾸면 숫자가 어긋납니다.
• 이 파드에는 인터넷이 없습니다. langgraph 0.2.60 이 이미 들어 있습니다.
• 공식 문서: [Streaming](https://docs.langchain.com/oss/python/langgraph/streaming) · [Graph API overview](https://docs.langchain.com/oss/python/langgraph/graph-api) · [Types 레퍼런스](https://reference.langchain.com/python/langgraph/types/)
• 흔한 실수: updates 가 수퍼스텝 단위라고 믿기, 재구성할 때 이어 붙이는 열쇠를 덮어쓰기, 진행 표시를 상태에 넣기, stream() 의 결과를 가공해 돌려주기(채점기는 원래 모양을 봅니다).

단계 8개

1. 상태 전체가 단계마다 나온다
2. 갱신은 노드마다 하나씩
3. 갱신만으로 상태를 다시 세운다
4. 몇 번째 단계에서 무엇이 함께 도는가
5. 여러 모드를 함께 듣는다
6. 상태에 남기지 않고 보여 준다
7. 흘려보낸 마지막과 한 번에 받은 답
8. 무엇을 언제 보여 줄지 정리한다