+
---
-## Define an agent workflow
+## Add SPARQL query validation
-Define the node to call the LLM
+Initialize the prefixes map and VoID classes schema that will be used by validation
```python
-def call_model(state: AgentState):
- """Call the model with the retrieved documents as context."""
- prompt_with_context = prompt_template.invoke({
- "messages": state["messages"],
- "relevant_docs": state['relevant_docs'],
- })
- response = llm.invoke(prompt_with_context)
- return {"messages": [response]}
+import logging
+from sparql_llm.utils import get_prefixes_and_schema_for_endpoints
+from index import endpoints
+
+logging.getLogger("httpx").setLevel(logging.WARNING)
+logging.info("Initializing endpoints metadata...")
+prefixes_map, endpoints_void_dict = get_prefixes_and_schema_for_endpoints(endpoints)
```
---
-## Define an agent workflow
+## Add SPARQL query validation
-Define the workflow "graph"
+Create the validation function
```python
-from langgraph.graph import StateGraph
-
-builder = StateGraph(AgentState)
-
-builder.add_node(retrieve_docs)
-builder.add_node(call_model)
-
-builder.add_edge("__start__", "retrieve_docs")
-builder.add_edge("retrieve_docs", "call_model")
-builder.add_edge("call_model", "__end__")
+from sparql_llm import validate_sparql_in_msg
+from langchain_core.messages import AIMessage
-graph = builder.compile()
+async def validate_output(last_msg: str) -> str | None:
+ """Validate the output of a LLM call, e.g. SPARQL queries generated."""
+ validation_outputs = validate_sparql_in_msg(last_msg, prefixes_map, endpoints_void_dict)
+ for validation_output in validation_outputs:
+ # Add step when missing prefixes have been fixed
+ if validation_output["fixed_query"]:
+ async with cl.Step(name="missing prefixes correction β
") as step:
+ step.output = f"Missing prefixes added to the generated query:\n```sparql\n{validation_output['fixed_query']}\n```"
+ # Create a new message to ask the model to fix the errors
+ if validation_output["errors"]:
+ recall_msg = f"""Fix the SPARQL query helping yourself with the error message and context from previous messages in a way that it is a fully valid query.\n
+### Error messages:\n- {'\n- '.join(validation_output['errors'])}\n
+### Erroneous SPARQL query\n```sparql\n{validation_output.get('fixed_query', validation_output['original_query'])}\n```"""
+ async with cl.Step(name=f"SPARQL query validation, got {len(validation_output['errors'])} errors to fix π") as step:
+ step.output = recall_msg
+ return recall_msg
```
---
-## Define an agent workflow
+## Add SPARQL query validation
-Update the UI
+Update the main `on_message` function running the chat to add a loop that makes sure the validation passes, if not we recall the LLM asking to fix the wrong query
```python
+max_try_count = 3
+
@cl.on_message
async def on_message(msg: cl.Message):
- answer = cl.Message(content="")
- async for msg, metadata in graph.astream(
- {"messages": cl.chat_context.to_openai()},
- stream_mode="messages",
- ):
- if not msg.response_metadata:
- await answer.stream_token(msg.content)
+ # [...]
+ for _i in range(max_try_count):
+ answer = cl.Message(content="")
+ for resp in llm.stream(messages):
+ await answer.stream_token(resp.content)
+ await answer.send()
+ validation_msg = await validate_output(answer.content)
+ if validation_msg is None:
+ break
else:
- await answer.send()
- answer = cl.Message(content="")
+ messages.append(("human", validation_msg))
```
> Try running your agent again now
---
-## Add SPARQL query validation
+## Use an agent framework
+
+Optionally you can move to fully use an "agent framework" like [LangGraph](https://langchain-ai.github.io/langgraph/#):
+
+β
Give access to some nice features
+
+- switch between streaming and complete response
+- parallel execution of nodes
+- generate a visual diagram for your workflow
+
+β
Provide structure to build your workflow
-Add fields to the state related to validation
+β οΈ Can be slower at runtime than doing things yourself
+
+β οΈ Relies on more dependencies increasing the overall complexity of the system, some people might find it more confusing than just using good old loops
+
+---
+
+## Use an agent framework
+
+Add the `langgraph` dependency to your `pyproject.toml`
+
+Define the state and update the retrieve function
```python
+from langgraph.graph.message import MessagesState
+
class AgentState(MessagesState):
- # [...]
+ """State of the agent available inside each node."""
+ relevant_docs: str
passed_validation: bool
try_count: int
+
+
+async def retrieve_docs(state: AgentState) -> dict[str, str]:
+ question = state["messages"][-1].content
+ # [...]
+ # This will update relevant_docs in the state:
+ return {"relevant_docs": relevant_docs}
```
---
-## Add SPARQL query validation
+## Use an agent framework
-Initialize the prefixes map and VoID classes schema that will be used by validation
+Define the node to call the LLM
```python
-import logging
-from sparql_llm.utils import get_prefixes_and_schema_for_endpoints
-from index import endpoints
-
-logging.getLogger("httpx").setLevel(logging.WARNING)
-logging.info("Initializing endpoints metadata...")
-prefixes_map, endpoints_void_dict = get_prefixes_and_schema_for_endpoints(endpoints)
+def call_model(state: AgentState):
+ """Call the model with the retrieved documents as context."""
+ response = llm.invoke([
+ ("system", SYSTEM_PROMPT.format(relevant_docs=state["relevant_docs"])),
+ *state["messages"],
+ ])
+ return {"messages": [response]}
```
---
-## Add SPARQL query validation
+## Use an agent framework
-Create the validation node
+Update the function that does validation
```python
-from sparql_llm import validate_sparql_in_msg
-
-async def validate_output(state: AgentState) -> dict[str, bool | list[tuple[str, str]] | int]:
- """Node to validate the output of a LLM call, e.g. SPARQL queries generated."""
- recall_messages = []
- validation_outputs = validate_sparql_in_msg(state["messages"][-1].content, prefixes_map, endpoints_void_dict)
- for validation_output in validation_outputs:
- # Handle when missing prefixes have been fixed
- if validation_output["fixed_query"]:
- async with cl.Step(name="missing prefixes correction β
") as step:
- step.output = f"Missing prefixes added to the generated query:\n```sparql\n{validation_output['fixed_query']}\n```"
- # Add a new message to ask the model to fix the errors
- if validation_output["errors"]:
- recall_msg = f"""Fix the SPARQL query helping yourself with the error message and context from previous messages in a way that it is a fully valid query.\n
-### Error messages:\n- {'\n- '.join(validation_output['errors'])}\n
-### Erroneous SPARQL query\n```sparql\n{validation_output['original_query']}\n```"""
- async with cl.Step(name=f"SPARQL query validation, got {len(validation_output['errors'])} errors to fix π") as step:
- step.output = recall_msg
+async def validate_output(state) -> dict[str, bool | list[tuple[str, str]] | int]:
+ recall_messages = []
+ last_msg = next(msg.content for msg in reversed(state["messages"]) if msg.content)
+ # [...]
+ # Add a new message to ask the model to fix the error
recall_messages.append(("human", recall_msg))
return {
"messages": recall_messages,
@@ -576,19 +701,21 @@ async def validate_output(state: AgentState) -> dict[str, bool | list[tuple[str,
}
```
+
+
---
-## Add SPARQL query validation
+## Use an agent framework
Create a conditional edge to route the workflow based on validation results
```python
from typing import Literal
-max_try_fix_sparql = 3
+max_try_count = 3
def route_model_output(state: AgentState) -> Literal["call_model", "__end__"]:
"""Determine the next node based on the model's output."""
- if state["try_count"] > max_try_fix_sparql:
+ if state["try_count"] > max_try_count:
return "__end__"
if not state["passed_validation"]:
return "call_model"
@@ -597,15 +724,67 @@ def route_model_output(state: AgentState) -> Literal["call_model", "__end__"]:
---
-## Add SPARQL query validation
+## Use an agent framework
-Add this new edge to the workflow graph
+Define the workflow "graph"
```python
+from langgraph.graph import StateGraph
+
+builder = StateGraph(AgentState)
+
+builder.add_node(retrieve_docs)
+builder.add_node(call_model)
builder.add_node(validate_output)
+builder.add_edge("__start__", "retrieve_docs")
+builder.add_edge("retrieve_docs", "call_model")
builder.add_edge("call_model", "validate_output")
builder.add_conditional_edges("validate_output", route_model_output)
+
+graph = builder.compile()
+```
+
+---
+
+## Use an agent framework
+
+Update the UI
+
+```python
+@cl.on_message
+async def on_message(msg: cl.Message):
+ answer = cl.Message(content="")
+ async for msg, metadata in graph.astream(
+ {"messages": cl.chat_context.to_openai()},
+ stream_mode="messages",
+ ):
+ if not msg.response_metadata:
+ await answer.stream_token(msg.content)
+ else:
+ print(msg.usage_metadata)
+ await answer.send()
+ answer = cl.Message(content="")
```
> Try running your agent again now
+
+---
+
+## Thank you
+
+[Complete script on GitHub](https://github.com/sib-swiss/sparql-llm/blob/main/tutorial/app.py)
+
+
+
+Live deployment for SIB endpoints (UniProt, Bgee, OMA, Rheaβ¦)
+
+[**chat.expasy.org**](https://chat.expasy.org)
+
+
+
+Code: [**github.com/sib-swiss/sparql-llm**](https://github.com/sib-swiss/sparql-llm)
+
+Short paper: [arxiv.org/abs/2410.06062](https://arxiv.org/abs/2410.06062)
+
+Standalone components available as a pip package: [pypi.org/project/sparql-llm](https://pypi.org/project/sparql-llm)
diff --git a/tutorial/slides/public/sparql_workflow.png b/tutorial/slides/public/sparql_workflow.png
new file mode 100644
index 0000000..51b3955
Binary files /dev/null and b/tutorial/slides/public/sparql_workflow.png differ
diff --git a/tutorial/uv.lock b/tutorial/uv.lock
index 6901233..9707790 100644
--- a/tutorial/uv.lock
+++ b/tutorial/uv.lock
@@ -1590,7 +1590,8 @@ wheels = [
[[package]]
name = "sparql-llm"
-source = { directory = "../packages/sparql-llm" }
+version = "0.0.8"
+source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "beautifulsoup4" },
{ name = "curies-rs" },
@@ -1598,22 +1599,9 @@ dependencies = [
{ name = "langchain-core" },
{ name = "rdflib" },
]
-
-[package.metadata]
-requires-dist = [
- { name = "beautifulsoup4", specifier = ">=4.13.0" },
- { name = "curies-rs", specifier = ">=0.1.3" },
- { name = "httpx", specifier = ">=0.27.2" },
- { name = "langchain-core", specifier = ">=0.3.34" },
- { name = "rdflib", specifier = ">=7.0.0" },
-]
-
-[package.metadata.requires-dev]
-dev = [
- { name = "mypy", specifier = ">=1.15.0" },
- { name = "pytest", specifier = ">=8.3.4" },
- { name = "pytest-cov", specifier = ">=6.0.0" },
- { name = "ruff", specifier = ">=0.9.5" },
+sdist = { url = "https://files.pythonhosted.org/packages/96/d6/a7e418dbc29d1571cbbbc8bb40f000fc35b4735e845f1ce91b8c6d22945b/sparql_llm-0.0.8.tar.gz", hash = "sha256:8b567f021d9074f072cbda852f7570ffd6bd26e03868019f8eef0acdb3cea670", size = 193778 }
+wheels = [
+ { url = "https://files.pythonhosted.org/packages/71/ce/a68cb0274aa211e97a4c4efb014abce4eb8e4c4938ba3de12b797f5449a4/sparql_llm-0.0.8-py3-none-any.whl", hash = "sha256:465bf8ffa9fc88497541a17d35b44b153312d1b786e580c084596864eb1de009", size = 85356 },
]
[[package]]
@@ -1780,7 +1768,7 @@ requires-dist = [
{ name = "langchain-qdrant", specifier = ">=0.2.0" },
{ name = "langgraph", specifier = ">=0.2.73" },
{ name = "qdrant-client", specifier = ">=1.13.0" },
- { name = "sparql-llm", directory = "../packages/sparql-llm" },
+ { name = "sparql-llm", specifier = ">=0.0.8" },
]
[[package]]
-```python
-from langgraph.graph.message import MessagesState
+Why do we add validation of the query generated:
-class AgentState(MessagesState):
- """State of the agent available inside each node."""
- relevant_docs: str
+π§ fix missing prefixes
+π detect use of a wrong predicate with a class
-async def retrieve_docs(state: AgentState) -> dict[str, str]:
- question = state["messages"][-1].content
- # [...]
- async with cl.Step(name=f"{len(retrieved_docs)} relevant documents ποΈ") as step:
- step.output = relevant_docs
- # This will update relevant_docs in the state:
- return {"relevant_docs": relevant_docs}
-```
+
+
+ 
+
+
+