diff --git a/CHANGELOG.md b/CHANGELOG.md index 2b82a28b..4ba639a5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- Agent and workflow spans no longer repeat earlier conversation turns in `gen_ai.input.messages`. When an app + re-sends the whole conversation, either as the graph state (for example a LangGraph `{"messages": [...]}` state) or + as a bare message list (for example the LangChain agent middleware), the exported input now starts at the current + user message, so continuous conversations no longer grow each trace's input with previous inputs and outputs + (SAO-19572). Tool-call and tool-response messages of the current turn are kept. A workflow or agent input that is + deliberately a message list, such as few-shot examples, is trimmed the same way. LLM span inputs still carry the + full prompt sent to the model. + ## [0.7.0] ### Added diff --git a/src/splunk_ao/converter/attribute_mapping.py b/src/splunk_ao/converter/attribute_mapping.py index 753f42bd..47210090 100644 --- a/src/splunk_ao/converter/attribute_mapping.py +++ b/src/splunk_ao/converter/attribute_mapping.py @@ -469,10 +469,34 @@ def _matching_child_finish_reason(span: WorkflowSpan | AgentSpan, output_message return None +def _current_turn_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Return the messages from the last user message onward, or all of them when there is no user message.""" + for index in range(len(messages) - 1, -1, -1): + if messages[index].get("role") == "user": + return messages[index:] + return messages + + +def _is_bare_message_list(value: Any) -> bool: + """Return True when the value is a non-empty list of role-tagged messages, not a state or update container.""" + parsed = _parse_json_string(value) + return ( + isinstance(parsed, Sequence) + and not isinstance(parsed, str | bytes | bytearray) + and _is_message_sequence(parsed) + ) + + def _set_orchestration_content(attrs: MutableMapping[str, AttributeValue], span: WorkflowSpan | AgentSpan) -> None: - input_messages, _ = _orchestration_messages(span.input, "user") + input_messages, input_full_history = _orchestration_messages(span.input, "user") if input_messages is not None: - attrs["gen_ai.input.messages"] = _json_string(input_messages) + # An app that re-sends the whole conversation, either as graph state ({"messages": [...]}) or as a + # bare message list (for example the LangChain agent middleware), would otherwise repeat every + # earlier turn in each new span's input. Export only the current turn; the untrimmed input_messages + # stay available below for the output history dedup. + trim_input = input_full_history or _is_bare_message_list(span.input) + exported_input = _current_turn_messages(input_messages) if trim_input else input_messages + attrs["gen_ai.input.messages"] = _json_string(exported_input) if span.output is None: return diff --git a/tests/test_attribute_mapping.py b/tests/test_attribute_mapping.py index c2b1b668..0203bbd7 100644 --- a/tests/test_attribute_mapping.py +++ b/tests/test_attribute_mapping.py @@ -457,6 +457,189 @@ def test_orchestration_output_omits_repeated_input_history() -> None: ] +def _orchestration_span(span_type: type[WorkflowSpan] | type[AgentSpan], **kwargs: Any) -> WorkflowSpan | AgentSpan: + span_kwargs: dict[str, Any] = {"name": "chatbot", **kwargs} + if span_type is AgentSpan: + span_kwargs["agent_type"] = AgentType.default + return span_type(**span_kwargs) + + +@pytest.mark.parametrize("span_type", [WorkflowSpan, AgentSpan]) +def test_orchestration_input_keeps_only_current_turn_of_full_history( + span_type: type[WorkflowSpan] | type[AgentSpan], +) -> None: + # Given: an app that re-sends the whole conversation (3 user turns) as the graph state on every request + user1 = {"role": "user", "content": "Look up patient P001"} + ai1 = {"role": "assistant", "content": "Patient P001 is George Rivera."} + user2 = {"role": "user", "content": "What is the dosage of Lisinopril?"} + ai2 = {"role": "assistant", "content": "10-40 mg once daily."} + user3 = {"role": "user", "content": "Look up patient P001 again"} + span = _orchestration_span( + span_type, + input=json.dumps({"messages": [user1, ai1, user2, ai2, user3]}), + output=json.dumps({"messages": [{"role": "assistant", "content": "Patient P001 is George Rivera."}]}), + ) + + # When: the orchestration content is converted + attrs = build_span_attributes(span) + + # Then: only the current user turn is exported as input, not the earlier questions and replies + assert json.loads(attrs["gen_ai.input.messages"]) == [_text_message("user", "Look up patient P001 again")] + + +@pytest.mark.parametrize("span_type", [WorkflowSpan, AgentSpan]) +def test_orchestration_input_keeps_tool_loop_messages_of_current_turn( + span_type: type[WorkflowSpan] | type[AgentSpan], +) -> None: + # Given: a graph node revisited mid-turn, whose state holds an earlier turn plus the current tool loop + user1 = {"role": "user", "content": "Hello"} + ai1 = {"role": "assistant", "content": "Hi, how can I help?"} + user2 = {"role": "user", "content": "Look up patient P001"} + tool_call = { + "role": "assistant", + "content": "", + "tool_calls": [{"id": "call-1", "function": {"name": "get_patient_info", "arguments": '{"id":"P001"}'}}], + } + tool_response = {"role": "tool", "content": "George Rivera", "tool_call_id": "call-1"} + span = _orchestration_span( + span_type, + input=json.dumps({"messages": [user1, ai1, user2, tool_call, tool_response]}), + output=json.dumps({"messages": [{"role": "assistant", "content": "Patient P001 is George Rivera."}]}), + ) + + # When: the orchestration content is converted + attrs = build_span_attributes(span) + + # Then: the input starts at the current user turn and keeps its tool call and tool response + input_messages = json.loads(attrs["gen_ai.input.messages"]) + assert [message["role"] for message in input_messages] == ["user", "assistant", "tool"] + assert input_messages[0] == _text_message("user", "Look up patient P001") + + +@pytest.mark.parametrize("span_type", [WorkflowSpan, AgentSpan]) +def test_orchestration_input_without_user_message_is_unchanged(span_type: type[WorkflowSpan] | type[AgentSpan]) -> None: + # Given: a state container with no user message to anchor the current turn on + ai = {"role": "assistant", "content": "Earlier reply"} + tool_response = {"role": "tool", "content": "result", "tool_call_id": "call-1"} + span = _orchestration_span(span_type, input=json.dumps({"messages": [ai, tool_response]})) + + # When: the orchestration content is converted + attrs = build_span_attributes(span) + + # Then: nothing is dropped from the input + assert [message["role"] for message in json.loads(attrs["gen_ai.input.messages"])] == ["assistant", "tool"] + + +@pytest.mark.parametrize("span_type", [WorkflowSpan, AgentSpan]) +def test_orchestration_input_trim_keeps_output_history_dedup(span_type: type[WorkflowSpan] | type[AgentSpan]) -> None: + # Given: a multi-turn input and an output that echoes the full input history before the new reply + user1 = {"role": "user", "content": "Hello"} + ai1 = {"role": "assistant", "content": "Hi, how can I help?"} + user2 = {"role": "user", "content": "What is Lisinopril?"} + final = {"role": "assistant", "content": "A blood pressure medication."} + span = _orchestration_span( + span_type, + input=json.dumps({"messages": [user1, ai1, user2]}), + output=json.dumps({"messages": [user1, ai1, user2, final]}), + ) + + # When: the orchestration content is converted + attrs = build_span_attributes(span) + + # Then: the input is trimmed to the current turn and the output still removes the full echoed history + assert json.loads(attrs["gen_ai.input.messages"]) == [_text_message("user", "What is Lisinopril?")] + assert json.loads(attrs["gen_ai.output.messages"]) == [ + _text_message("assistant", "A blood pressure medication.", finish_reason="unknown") + ] + + +def test_orchestration_input_partial_update_container_is_unchanged() -> None: + # Given: a LangGraph partial update (not the full state), which only carries newly produced messages + user1 = {"role": "user", "content": "Hello"} + user2 = {"role": "user", "content": "Second question"} + span = AgentSpan( + name="Agent", agent_type=AgentType.default, input=json.dumps({"update": {"messages": [user1, user2]}}) + ) + + # When: the orchestration content is converted + attrs = build_span_attributes(span) + + # Then: the update is exported as is + assert json.loads(attrs["gen_ai.input.messages"]) == [ + _text_message("user", "Hello"), + _text_message("user", "Second question"), + ] + + +@pytest.mark.parametrize("span_type", [WorkflowSpan, AgentSpan]) +def test_orchestration_bare_message_list_input_keeps_only_current_turn( + span_type: type[WorkflowSpan] | type[AgentSpan], +) -> None: + # Given: an agent whose input is the conversation as a bare message list (the LangChain agent middleware shape) + user1 = {"role": "user", "content": "What is the weather in Paris?"} + ai1 = {"role": "assistant", "content": "It is 72 degrees."} + user2 = {"role": "user", "content": "And the AAPL stock price?"} + ai2 = {"role": "assistant", "content": "It is 178.50 USD."} + user3 = {"role": "user", "content": "Remind me of the Paris weather"} + span = _orchestration_span(span_type, input=json.dumps([user1, ai1, user2, ai2, user3])) + + # When: the orchestration content is converted + attrs = build_span_attributes(span) + + # Then: only the current user turn is exported as input + assert json.loads(attrs["gen_ai.input.messages"]) == [_text_message("user", "Remind me of the Paris weather")] + + +@pytest.mark.parametrize("span_type", [WorkflowSpan, AgentSpan]) +def test_orchestration_bare_message_list_input_keeps_tool_loop_of_current_turn( + span_type: type[WorkflowSpan] | type[AgentSpan], +) -> None: + # Given: a bare message list holding an earlier turn plus the current turn's tool call and tool response + user1 = {"role": "user", "content": "Hello"} + ai1 = {"role": "assistant", "content": "Hi, how can I help?"} + user2 = {"role": "user", "content": "Look up patient P001"} + tool_call = { + "role": "assistant", + "content": "", + "tool_calls": [{"id": "call-1", "function": {"name": "get_patient_info", "arguments": '{"id":"P001"}'}}], + } + tool_response = {"role": "tool", "content": "George Rivera", "tool_call_id": "call-1"} + span = _orchestration_span(span_type, input=json.dumps([user1, ai1, user2, tool_call, tool_response])) + + # When: the orchestration content is converted + attrs = build_span_attributes(span) + + # Then: the input starts at the current user turn and keeps its tool call and tool response + input_messages = json.loads(attrs["gen_ai.input.messages"]) + assert [message["role"] for message in input_messages] == ["user", "assistant", "tool"] + assert input_messages[0] == _text_message("user", "Look up patient P001") + + +@pytest.mark.parametrize( + "messages", + [ + pytest.param([{"role": "user", "content": "Only question"}], id="single_user_turn"), + pytest.param( + [{"role": "assistant", "content": "Earlier reply"}, {"role": "tool", "content": "r", "tool_call_id": "c"}], + id="no_user_message", + ), + ], +) +def test_orchestration_bare_message_list_input_without_earlier_turns_is_unchanged( + messages: list[dict[str, Any]], +) -> None: + # Given: a bare message list with a single user turn, or with no user message to anchor the current turn on + span = AgentSpan(name="Agent", agent_type=AgentType.default, input=json.dumps(messages)) + + # When: the orchestration content is converted + attrs = build_span_attributes(span) + + # Then: nothing is dropped from the input + assert [message["role"] for message in json.loads(attrs["gen_ai.input.messages"])] == [ + message["role"] for message in messages + ] + + @pytest.mark.parametrize("span_type", [WorkflowSpan, AgentSpan]) def test_orchestration_full_history_keeps_last_terminal_message( span_type: type[WorkflowSpan] | type[AgentSpan], diff --git a/tests/test_crewai_handler.py b/tests/test_crewai_handler.py index b76261a6..be5babc9 100644 --- a/tests/test_crewai_handler.py +++ b/tests/test_crewai_handler.py @@ -1,6 +1,7 @@ import asyncio import hashlib import inspect +import json import sys import threading import uuid @@ -2152,3 +2153,65 @@ def test_tool_lifecycle_previous_event_id_with_error(crewai_callback) -> None: start_run_id = mock_start.call_args[1]["run_id"] end_run_id = mock_end.call_args[1]["run_id"] assert start_run_id == end_run_id + + +def _exported_message_texts(span: ReadableSpan) -> list[tuple[str, str]]: + messages = json.loads(span.attributes["gen_ai.input.messages"]) + return [ + (m["role"], " ".join(p.get("content", "") for p in m["parts"] if p.get("type") == "text")) for m in messages + ] + + +def test_crewai_root_input_stays_current_turn_when_kickoff_inputs_resend_history(crewai_bus) -> None: + # Given: a chat-style crew kicked off 3 times, each time with the whole conversation as its inputs + _listener, _logger, sink, events, deliver = crewai_bus + crew = MockCrew() + agent = MockAgent(crew=crew) + task = MockTask(agent=agent) + questions = ["Name one AI benefit in healthcare", "Name a second benefit", "Summarize the two benefits"] + history: list[dict[str, str]] = [] + + async def run() -> None: + for turn, question in enumerate(questions): + history.append({"role": "user", "content": question}) + call_id = f"call-{turn}" + # When: the turn runs with the full conversation as the kickoff inputs and the model prompt + await deliver( + [ + (crew, events["CrewKickoffStartedEvent"](crew_name="Crew", inputs={"messages": list(history)})), + (task, events["TaskStartedEvent"](task=task)), + (agent, events["AgentExecutionStartedEvent"](agent=agent, task=task, task_prompt=question)), + ( + MockSource(), + events["LLMCallStartedEvent"]( + call_id=call_id, + model="local-model", + agent_id=str(agent.id), + task_id=str(task.id), + messages=list(history), + ), + ), + ( + MockSource(), + events["LLMCallCompletedEvent"]( + call_id=call_id, agent_id=str(agent.id), task_id=str(task.id), response=f"Reply {turn}" + ), + ), + (agent, events["AgentExecutionCompletedEvent"](agent=agent, task=task, output=f"Reply {turn}")), + (task, events["TaskCompletedEvent"](task=task, output=MockOutput(f"Reply {turn}"))), + (crew, events["CrewKickoffCompletedEvent"](output=MockOutput(f"Reply {turn}"))), + ] + ) + history.append({"role": "assistant", "content": f"Reply {turn}"}) + + asyncio.run(run()) + + # Then: each crew root exports only the current question, whatever the turn + for turn, question in enumerate(questions): + spans = sink.spans[turn * 4 : turn * 4 + 4] + chat = next(s for s in spans if s.attributes["gen_ai.operation.name"] == "chat") + root = [s for s in spans if s.attributes["gen_ai.operation.name"] == "invoke_workflow"][-1] + assert _exported_message_texts(root) == [("user", question)] + # And: the chat span keeps the full prompt the model received, including every earlier question + prompt = " ".join(text for _role, text in _exported_message_texts(chat)) + assert all(earlier in prompt for earlier in questions[: turn + 1]) diff --git a/tests/test_langchain.py b/tests/test_langchain.py index 2932c533..293da833 100644 --- a/tests/test_langchain.py +++ b/tests/test_langchain.py @@ -1,3 +1,4 @@ +import json import time import uuid from collections.abc import Generator @@ -1810,3 +1811,58 @@ def test_async_callback_with_ingestion_hook_no_credentials(self): # Then: the callback is created successfully and the hook is attached assert callback._handler._splunk_ao_logger._ingestion_hook is mock_hook + + +def _exported_message_texts(span: ReadableSpan, attribute: str = "gen_ai.input.messages") -> list[tuple[str, str]]: + messages = json.loads((span.attributes or {})[attribute]) + return [ + (m["role"], " ".join(p.get("content", "") for p in m["parts"] if p.get("type") == "text")) for m in messages + ] + + +def test_langgraph_root_input_stays_current_turn_when_state_resends_history() -> None: + # Given: a LangGraph-style root, a node, and a chat model call, over 3 turns whose state re-sends the whole history + sink = RecordingSink() + logger = SplunkAOLogger(project_id="project-id", agent_stream_id="stream-id", _sink=sink) + callback = SplunkAOCallback(splunk_ao_logger=logger, flush_on_chain_end=False) + questions = ["Look up patient P001", "What is the dosage of Lisinopril?", "Look up patient P001 again"] + history: list[Any] = [] + try: + for turn, question in enumerate(questions, start=1): + history.append(HumanMessage(content=question)) + root_id, node_id, llm_id = uuid.uuid4(), uuid.uuid4(), uuid.uuid4() + reply = AIMessage(content=f"Reply {turn}") + + # When: the turn runs with the full conversation as the graph state + callback.on_chain_start( + serialized={"name": "LangGraph"}, inputs={"messages": list(history)}, run_id=root_id + ) + callback.on_chain_start( + serialized={"name": "chatbot"}, + inputs={"messages": list(history)}, + run_id=node_id, + parent_run_id=root_id, + ) + callback.on_chat_model_start( + serialized={"name": "gpt"}, messages=[list(history)], run_id=llm_id, parent_run_id=node_id + ) + callback.on_llm_end( + LLMResult(generations=[[ChatGeneration(message=reply)]], llm_output=None), + run_id=llm_id, + parent_run_id=node_id, + ) + callback.on_chain_end(outputs={"messages": [reply]}, run_id=node_id, parent_run_id=root_id) + callback.on_chain_end(outputs={"messages": [*history, reply]}, run_id=root_id) + history.append(reply) + + # Then: the root agent and the node export only the current question, whatever the turn + operations = [(span.attributes or {})["gen_ai.operation.name"] for span in sink.spans] + assert operations == ["chat", "invoke_workflow", "invoke_agent"] * 3 + for turn, question in enumerate(questions): + chat, node, root = sink.spans[turn * 3 : turn * 3 + 3] + assert _exported_message_texts(root) == [("user", question)] + assert _exported_message_texts(node) == [("user", question)] + # And: the chat span keeps the full prompt the model received (1, 3 and 5 messages) + assert len(_exported_message_texts(chat)) == 2 * turn + 1 + finally: + logger.terminate() diff --git a/tests/test_langchain_middleware.py b/tests/test_langchain_middleware.py index 7687f5e7..78fa78e5 100644 --- a/tests/test_langchain_middleware.py +++ b/tests/test_langchain_middleware.py @@ -525,3 +525,55 @@ def test_ingestion_hook_called(self, middleware_builder) -> None: middleware.after_agent(cast(AgentState, MockAgentState(messages=[AIMessage(content="response")])), rt) mock_hook.assert_called_once() + + +class _RecordingSink: + def __init__(self) -> None: + self.spans: list[Any] = [] + + def emit(self, span: Any) -> None: + self.spans.append(span) + + def force_flush(self) -> bool: + return True + + def shutdown(self) -> None: + return None + + +def _exported_message_texts(span: Any) -> list[tuple[str, str]]: + messages = json.loads((span.attributes or {})["gen_ai.input.messages"]) + return [ + (m["role"], " ".join(p.get("content", "") for p in m["parts"] if p.get("type") == "text")) for m in messages + ] + + +def test_middleware_root_input_stays_current_turn_when_state_resends_history() -> None: + # Given: a create_agent-style run over 3 turns whose state re-sends the whole conversation each time + sink = _RecordingSink() + logger = SplunkAOLogger(project_id="project-id", agent_stream_id="stream-id", _sink=sink) + middleware = SplunkAOMiddleware(splunk_ao_logger=logger, flush_on_chain_end=False) + runtime = cast("Runtime", MockRuntime()) + questions = ["What is the weather in Paris?", "And the AAPL stock price?", "Remind me of the Paris weather"] + history: list[Any] = [] + try: + for turn, question in enumerate(questions, start=1): + history.append(HumanMessage(content=question)) + state = cast(AgentState, MockAgentState(messages=list(history))) + reply = AIMessage(content=f"Reply {turn}") + + # When: the turn runs (agent start, one model call, agent end) + middleware.before_agent(state, runtime) + request = MockRequest(state=state, messages=list(history), model=MockModel("gpt")) + middleware.wrap_model_call(cast(Any, request), lambda _request, reply=reply: MockResponse(result=[reply])) + history.append(reply) + middleware.after_agent(cast(AgentState, MockAgentState(messages=list(history))), runtime) + + # Then: each root agent span exports only the current question, whatever the turn + roots = [s for s in sink.spans if (s.attributes or {}).get("gen_ai.operation.name") == "invoke_agent"] + assert [_exported_message_texts(root) for root in roots] == [[("user", question)] for question in questions] + # And: the chat span keeps the full prompt the model received (1, 3 and 5 messages) + chats = [s for s in sink.spans if (s.attributes or {}).get("gen_ai.operation.name") == "chat"] + assert [len(_exported_message_texts(chat)) for chat in chats] == [1, 3, 5] + finally: + logger.terminate() diff --git a/tests/test_openai_agents.py b/tests/test_openai_agents.py index 27d84871..d2ae1de1 100644 --- a/tests/test_openai_agents.py +++ b/tests/test_openai_agents.py @@ -1145,3 +1145,41 @@ def test_workflow_input_falls_back_when_generation_input_is_set_after_start() -> finally: set_trace_processors([]) logger.terminate() + + +def test_openai_agents_workflow_input_stays_current_turn_when_history_is_resent() -> None: + # Given: 3 turns, each a trace whose model call re-sends the whole conversation (as Runner.run with to_input_list) + sink = RecordingSink() + logger = SplunkAOLogger(project_id="project-id", agent_stream_id="stream-id", _sink=sink) + processor = SplunkAOTracingProcessor(splunk_ao_logger=logger, flush_on_trace_end=False) + set_trace_processors([processor]) + questions = ["Find activities in Paris", "Now find activities in Tokyo", "Which activities did you list for Paris?"] + history: list[dict[str, str]] = [{"role": "system", "content": "You are a concise travel agent."}] + try: + for turn, question in enumerate(questions, start=1): + history.append({"role": "user", "content": question}) + reply = {"role": "assistant", "content": f"Reply {turn}"} + + # When: the turn is traced + with agents_tracing.trace("Agent trace", trace_id=f"trace_multi_turn_{turn}") as openai_trace: + with agent_span("TravelAgent", span_id=f"span_agent_{turn}", parent=openai_trace) as root_span: + with generation_span( + input=list(history), output=[reply], model="gpt-4", span_id=f"span_llm_{turn}", parent=root_span + ): + pass + history.append(reply) + + # Then: each workflow span exports only the current question, whatever the turn + def messages(span: ReadableSpan) -> list[dict[str, Any]]: + return json.loads((span.attributes or {})["gen_ai.input.messages"]) + + def by_operation(operation: str) -> list[ReadableSpan]: + return [s for s in sink.spans if (s.attributes or {}).get("gen_ai.operation.name") == operation] + + workflows = by_operation("invoke_workflow") + assert [[m["parts"][0]["content"] for m in messages(w)] for w in workflows] == [[q] for q in questions] + # And: the chat span keeps the full prompt the model received (system + 1, 3 and 5 turns) + assert [len(messages(chat)) for chat in by_operation("chat")] == [2, 4, 6] + finally: + set_trace_processors([]) + logger.terminate()