# Agentic Workflows
An agent plans, calls tools and works across documents and steps. ContextGem reads one document in depth and returns typed values, each with the text that it came from, or with the probability of each answer to a closed question. The two meet at three points of a workflow.
| Where | What ContextGem does | What the agent gets |
|--------------------------------------------------------------------|------------------------------------------------------------------|--------------------------------------------------------------------------------------|
| [At ingestion](#agentic-ingestion-label) | Extracts the metadata of each new document once, with a pipeline | Records to filter and route on, without reading the documents again |
| [In a tool](#agentic-tool-label) | Extracts what the agent asks for, when it asks | Values with the sentences they came from, without the document in its context |
| [At a decision point](#agentic-decision-label) | Answers closed questions with a decision model | A branch to take, or the tools for its turn, and how sure the model is of the answer |
The examples below work on the same service agreement, and show the answers that the models gave.
## đź“Ą At Ingestion: Metadata for Every Document
When a document enters your system, e.g. a knowledge base that agents search, extract its metadata once and store it with the document. An agent then finds documents by their type, parties or dates, and does not have to read a document to learn what it is.
```python
# ContextGem: Extracting the Metadata of a Document at Ingestion
import os
from contextgem import (
BooleanConcept,
DateConcept,
Document,
DocumentDecisionModel,
DocumentLLM,
DocumentModelGroup,
ExtractionPipeline,
LabelConcept,
StringConcept,
)
# The metadata fields, defined once for all the documents
metadata_pipeline = ExtractionPipeline(
concepts=[
# Closed questions: answered by the decision model, with probabilities
LabelConcept(
name="Document type",
description="Type of the document",
labels=["NDA", "Service agreement", "Lease", "Other"],
classification_type="multi_class",
model_role="decide",
),
BooleanConcept(
name="Automatic renewal",
description="The agreement renews automatically",
model_role="decide",
),
# Open values: extracted by the LLM
StringConcept(
name="Parties",
description="Names of the parties to the document",
),
DateConcept(
name="Effective date",
description="Date from which the document is in effect",
),
]
)
model_group = DocumentModelGroup(
models=[
DocumentLLM(
model="openai/gpt-6-luna",
api_key=os.getenv("CONTEXTGEM_OPENAI_API_KEY"),
),
DocumentDecisionModel(
model="typesafe/jev-latest",
api_key=os.getenv("CONTEXTGEM_TYPESAFE_API_KEY"),
role="decide",
),
]
)
# A new document comes in
document = Document(
raw_text=(
"Service Agreement\n"
"This agreement is made on 1 March 2026 between Acme Ltd (Customer) "
"and Northwind AS (Supplier).\n"
"The Supplier shall provide cloud hosting services as described "
"in Schedule 1.\n"
"This agreement renews automatically for successive one-year terms.\n"
"Either party may terminate this agreement for any reason by giving "
"30 days' written notice to the other party.\n"
"This agreement is governed by the laws of Norway.\n"
),
)
document.assign_pipeline(metadata_pipeline) # each document gets a copy
document = model_group.extract_all(document)
doc_type = document.get_concept_by_name("Document type").extracted_items[0]
label = doc_type.value[0]
renewal = document.get_concept_by_name("Automatic renewal").extracted_items[0]
parties = document.get_concept_by_name("Parties").extracted_items
dates = document.get_concept_by_name("Effective date").extracted_items
# The record of the document, for your search index or database
record = {
"id": "acme-service-agreement",
"document_type": label, # "Service agreement"
"document_type_probability": doc_type.probabilities[label], # 1.0
"automatic_renewal": renewal.value, # True, with a probability of 0.99
"parties": [p.value for p in parties], # ["Acme Ltd", "Northwind AS"]
"effective_date": [d.value.isoformat() for d in dates], # ["2026-03-01"]
# The document with its extracted items, which `Document.from_json()`
# loads later without extracting again
"contextgem_document": document.to_json(),
}
```
**Key points:**
* **Define the fields once**, as an [`ExtractionPipeline`](https://contextgem.shcherbak.ai/api/pipelines/index.html.md#contextgem.public.pipelines.ExtractionPipeline). `assign_pipeline()` gives each document a copy of its aspects and concepts, so one pipeline serves all the documents. The pipeline is serialized with `to_json()`, e.g. to keep the definition of the metadata next to the records made with it (see [Extraction Pipelines](https://contextgem.shcherbak.ai/pipelines/extraction_pipelines/index.html.md))
* **Send each field to the model that suits it.** In a [`DocumentModelGroup`](https://contextgem.shcherbak.ai/api/models/index.html.md#contextgem.public.models.DocumentModelGroup), a decision model answers the closed questions (a type, a yes/no flag) fast and at a low cost per document, and an LLM extracts the open values. Store the probabilities with the values, so that a search can leave out the uncertain ones, or a person can check them (see [Decision Models](https://contextgem.shcherbak.ai/decision_models/decision_models/index.html.md))
* **Keep the processed document.** `document.to_json()` holds every extracted item with its references and justifications, and `Document.from_json()` restores it later, e.g. in a tool that returns the sources of a value, without extracting again (see [Serializing objects and results](https://contextgem.shcherbak.ai/serialization/index.html.md))
* **Index aspects as labelled passages.** An aspect extracts the paragraphs on a topic, e.g. the termination clauses. Indexed as passages labelled with the name of the aspect, they let a search look in the termination clauses only (see [Aspect Extraction](https://contextgem.shcherbak.ai/aspects/aspects/index.html.md))
* **Extract many documents concurrently** with `extract_all_async()`, one call per document, run together with `asyncio.gather()`. With `use_concurrency=True`, the requests of all the calls that run in an event loop share the rate set by the `async_limiter` of each model (see [Configuring LLM(s)](https://contextgem.shcherbak.ai/llms/llm_config/index.html.md))
## 🛠️ In a Tool: Extraction on Demand
When an agent needs facts from a document, it calls a tool that runs an extraction. The extraction reads the whole document in a request of its own, and the agent gets back the values with the sentences that they came from: the document stays out of the context of the agent, and the agent can quote its sources.
```python
# ContextGem: Extraction as a Tool of an Agent
import json
import os
from contextgem import Document, DocumentLLM, StringConcept, register_tool
# The documents that the agent can read, e.g. loaded from your store
documents = {
"acme-service-agreement": Document(
raw_text=(
"Service Agreement\n"
"This agreement is made on 1 March 2026 between Acme Ltd "
"(Customer) and Northwind AS (Supplier).\n"
"The Supplier shall provide cloud hosting services as described "
"in Schedule 1.\n"
"This agreement renews automatically for successive one-year "
"terms.\n"
"Either party may terminate this agreement for any reason by "
"giving 30 days' written notice to the other party.\n"
"This agreement is governed by the laws of Norway.\n"
),
),
}
# The LLM that extracts, which can be another model than the agent's
extractor = DocumentLLM(
model="openai/gpt-6-luna",
api_key=os.getenv("CONTEXTGEM_OPENAI_API_KEY"),
)
@register_tool
async def extract_from_document(
document_id: str,
name: str,
description: str,
) -> str:
"""
Extracts information from a document, with the sentences it was found in.
:param document_id: The ID of the document
:param name: A short name of the information, e.g. "Payment term"
:param description: What to extract, e.g. "Days to pay an invoice"
"""
# A copy: the stored document stays as it is
document = documents[document_id].clone()
document.add_concepts(
[
StringConcept(
name=name,
description=description,
add_references=True,
reference_depth="sentences",
)
]
)
await extractor.extract_concepts_from_document_async(document)
items = document.get_concept_by_name(name).extracted_items
# The result: the values and their sources, not the whole document
return json.dumps(
[
{
"value": item.value,
"sources": [s.raw_text for s in item.reference_sentences],
}
for item in items
]
)
# The agent: here, ContextGem's own chat with tools
agent = DocumentLLM(
model="openai/gpt-6-luna",
api_key=os.getenv("CONTEXTGEM_OPENAI_API_KEY"),
system_message=(
"You answer questions about the user's documents, which you read with "
"the tool. Quote the sentences that your answer relies on. Documents: "
"acme-service-agreement (service agreement with Acme Ltd)."
),
tools=[extract_from_document],
)
question = "How much notice do we need to end the agreement with Acme?"
answer = agent.chat(question)
# The agent called the tool with the name "Termination notice period", and got
# the value "Either party must give 30 days' written notice to terminate the
# agreement for any reason.", with its source: the sentence of the document
# on termination.
# answer: "You need to give Acme **30 days' written notice** to end the
# agreement.", followed by that sentence as a quote
```
**Key points:**
* **The tool is a Python function** with type hints and a docstring. Here, ContextGem’s own chat runs it (see [Chat with Tools](https://contextgem.shcherbak.ai/llms/chat_with_tools/index.html.md)). Agent frameworks that take a Python function as a tool take this one as it is, e.g. the OpenAI Agents SDK with `function_tool(extract_from_document)`, or PydanticAI with `Agent(..., tools=[extract_from_document])`
* **The agent describes what it needs** as the name and the description of a concept, as you would, and ContextGem builds the prompt from them. For the questions that you know in advance, define the concepts yourself, or extract them at ingestion and have the tool return the stored values, without a request
* **Return the values and their sources only.** The result of a tool is added to the context of the agent, and is sent with each of its next requests
* **Use the async methods in async code.** Agent frameworks run tools in an event loop: `extract_concepts_from_document_async()` and the other `_async` methods run in that loop, and segment sentences, where they need them, in a worker thread, while a sync method called in a running event loop runs in a thread of its own and blocks the loop until it is done
* **The extractor can be another model** than the agent’s: a smaller LLM, a decision model for closed questions, or a model group. Its usage and cost are tracked apart from the agent’s, with `extractor.get_usage()` and `extractor.get_cost()`
### Serving the Tool over MCP
To give the tool to any agent that connects to Model Context Protocol (MCP) servers, e.g. a coding agent or a desktop assistant, serve the same function with an MCP server. With the [MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk) (`uv add "mcp>=2"`):
```python
from mcp.server.mcpserver import MCPServer
# `extract_from_document` and what it uses: as in the example above
server = MCPServer("documents")
server.add_tool(extract_from_document)
if __name__ == "__main__":
server.run() # over stdio, or server.run("streamable-http") over HTTP
```
## 🔀 At a Decision Point: Choosing the Next Step
Where a workflow branches, e.g. before an agent takes a step that is hard to undo, ask the question as a concept of a decision model. A decision model answers a closed question with the probability of each possible answer and generates no text, so the workflow branches on a value that you defined, and the probability tells how clear the answer is.
```python
# ContextGem: Deciding the Next Step of a Workflow with a Decision Model
import os
from contextgem import BooleanConcept, Document, DocumentDecisionModel
decision_model = DocumentDecisionModel(
model="typesafe/jev-latest",
api_key=os.getenv("CONTEXTGEM_TYPESAFE_API_KEY"),
)
# The document that the next step depends on
document = Document(
raw_text=(
"Service Agreement\n"
"This agreement is made on 1 March 2026 between Acme Ltd (Customer) "
"and Northwind AS (Supplier).\n"
"The Supplier shall provide cloud hosting services as described "
"in Schedule 1.\n"
"This agreement renews automatically for successive one-year terms.\n"
"Either party may terminate this agreement for any reason by giving "
"30 days' written notice to the other party.\n"
"This agreement is governed by the laws of Norway.\n"
),
)
question = BooleanConcept(
name="Termination for convenience",
description="Either party may terminate the agreement without cause",
)
document.add_concepts([question])
document = decision_model.extract_all(document)
concept = document.get_concept_by_name("Termination for convenience")
p_yes = concept.extracted_items[0].probabilities[True] # 0.98
# Act on a clear answer, and leave an unclear one to a person
if p_yes >= 0.9:
next_step = "draft_termination_notice" # the step taken here
elif p_yes <= 0.1:
next_step = "reply_not_terminable"
else:
next_step = "human_review"
```
**Key points:**
* **Act on clear answers, and escalate the rest.** Two bounds of your choice set apart the answers that the workflow acts on, and an answer in between goes where you send it, e.g. to a person, as here, to an LLM that reasons the question through, or to the branch that is safe to take. A probability is the estimate of the model: check the bounds on a sample of your documents before you rely on them (see [⚠️ How Far to Trust the Probabilities](https://contextgem.shcherbak.ai/decision_models/decision_models/index.html.md#decision-models-trust-label))
* **Route by a label.** A [`LabelConcept`](https://contextgem.shcherbak.ai/api/concepts/index.html.md#contextgem.public.concepts.LabelConcept) with `classification_type="multi_class"` picks one of your labels, e.g. the workflow or the agent that takes a document, with the probability of each label
* **Check the output of an agent.** The same questions check a document that an agent wrote, e.g. that a draft states the notice period, before it is sent. A failed check goes back to the agent as feedback
* **Only a decision model gives probabilities.** An item extracted by an LLM, including an LLM that is a fallback of the decision model, has `probabilities=None`: send it to a person, or decide on its value
* **Use an LLM where the answer takes reasoning**, dates or arithmetic (see [🧠When to Use a Decision Model](https://contextgem.shcherbak.ai/decision_models/decision_models/index.html.md#decision-models-when-to-use-label))
### Choosing the Tools of an Agent
The descriptions of all the tools of an agent are sent with each of its requests, and among many tools the agent can pick a wrong one. Before the turn of the agent, a decision model can decide which groups of tools the request needs: a [`LabelConcept`](https://contextgem.shcherbak.ai/api/concepts/index.html.md#contextgem.public.concepts.LabelConcept) with `classification_type="multi_label"` and the groups as its labels asks one yes/no question for each label, and the agent gets the tools of the groups at or above the decision threshold.
```python
# ContextGem: Choosing the Tools of an Agent with a Decision Model
import os
from contextgem import (
Document,
DocumentDecisionModel,
DocumentLLM,
LabelConcept,
register_tool,
)
# The tools of the agent (here, with made-up results)
@register_tool
def search_contracts(query: str) -> str:
"""Searches the contracts of the company."""
return "acme-service-agreement"
@register_tool
def look_up_invoice(invoice_number: str) -> str:
"""Looks up an invoice by its number."""
return '{"amount": "EUR 4,200", "status": "overdue"}'
@register_tool
def send_email(to: str, subject: str, body: str) -> str:
"""Sends an email."""
return "Sent."
@register_tool
def schedule_meeting(attendees: list[str], start: str) -> str:
"""Schedules a meeting."""
return "Scheduled."
# The tools in groups, each with a label that says what its tools do
tool_groups = {
"Search contracts": [search_contracts],
"Look up invoices": [look_up_invoice],
"Send emails": [send_email],
"Schedule meetings": [schedule_meeting],
}
decision_model = DocumentDecisionModel(
model="typesafe/jev-latest",
api_key=os.getenv("CONTEXTGEM_TYPESAFE_API_KEY"),
)
agent = DocumentLLM(
model="openai/gpt-6-luna",
api_key=os.getenv("CONTEXTGEM_OPENAI_API_KEY"),
system_message="You are an assistant of a finance team.",
)
request = "Remind billing@acme.example of invoice 1043, with the amount due."
# Before the turn of the agent: which groups of tools does the request need?
document = Document(raw_text=request)
document.add_concepts(
[
LabelConcept(
name="Tools needed",
description="Kinds of actions needed to carry out the request",
labels=list(tool_groups),
classification_type="multi_label",
# A tool too many costs less than a tool missing
decision_threshold=0.3,
)
]
)
document = decision_model.extract_all(document)
# The item has the labels at or above the threshold, each with a probability
# of its own. No item means that no label reached it: keep all the tools then
items = document.get_concept_by_name("Tools needed").extracted_items
groups = items[0].value if items else list(tool_groups)
# groups: ["Look up invoices", "Send emails"]. items[0].probabilities:
# {"Search contracts": 0.14, "Look up invoices": 0.94, "Send emails": 0.96,
# "Schedule meetings": 0.02}
agent.tools = [tool for group in groups for tool in tool_groups[group]]
answer = agent.chat(request)
# The agent looked up the invoice, then sent the email.
# answer: "Sent a reminder to billing@acme.example for overdue invoice 1043.
# The amount due is EUR 4,200."
```
**Key points:**
* **Name each group by what its tools do.** A label is sent as it is, without a description of its own, so “Look up invoices” says more than `look_up_invoice`. Put related tools under one label, so that each label has a clear meaning
* **Set the threshold low.** A tool offered in vain costs the tokens of its description, while a tool left out can leave the agent unable to do the task. When no label reaches the threshold, no item is extracted: keep all the tools then, or a default set
* **Decide on what the choice depends on**, e.g. the request together with the last turns of the conversation, as the text of the document. The questions of up to 128 labels go in one request, and more in further requests on the same text (see [đź“Ź Limits](https://contextgem.shcherbak.ai/decision_models/decision_models/index.html.md#decision-models-limits-label))
* **The tools can change from one turn of a chat to the next.** A [`ChatSession`](https://contextgem.shcherbak.ai/api/llms/index.html.md#contextgem.public.llms.ChatSession) keeps the tool calls of the earlier turns, with their results, when a turn offers other tools or none, so the agent keeps what it found with them (checked with `gpt-6-luna` and `claude-haiku-5-5`). The thinking blocks of the earlier turns are sent back only while the tools stay the same (see [Chat with Tools](https://contextgem.shcherbak.ai/llms/chat_with_tools/index.html.md))
* **With fallback LLMs, set the tools of each fallback LLM first**, and then those of the agent: the tools of a fallback LLM must be those of the LLM it falls back from, which is checked when the tools of the agent change (see [Tools and Fallback LLMs](https://contextgem.shcherbak.ai/llms/chat_with_tools/index.html.md#llm-tools-fallback-label))
## ⚙️ At Any of These Points
* **Hand results to the next step as JSON.** Documents with their extracted items, pipelines and model configurations are serialized with `to_json()` and restored with `from_json()`, also in another process or service. The credentials of the models (API key and API base) are left out (see [Serializing objects and results](https://contextgem.shcherbak.ai/serialization/index.html.md))
* **Track the cost of each step.** Each model counts its own usage and cost, so give each step a model of its own and read its `get_usage()` and `get_cost()` (see [Configuring LLM(s)](https://contextgem.shcherbak.ai/llms/llm_config/index.html.md))
* **Plan for failures.** Fallback models take over when a model fails. When none is left, [`ModelAPIError`](https://contextgem.shcherbak.ai/api/exceptions/index.html.md#contextgem.public.exceptions.ModelAPIError) or [`ModelExtractionError`](https://contextgem.shcherbak.ai/api/exceptions/index.html.md#contextgem.public.exceptions.ModelExtractionError) is raised, both caught by `except ContextGemError`. With `raise_exception_on_extraction_error` set to `False` in the extraction method, a warning is logged instead, and the aspects and concepts of the failed call have no items
* **Treat documents as untrusted input.** The text of a document is sent to the models, and can hold instructions aimed at them. Pass extracted values to an agent as data, e.g. as the result of a tool, and take the steps that matter on closed answers, whose values are the ones that you defined
* **Leave the agent’s work to the agent framework:** planning, memory, search across documents and the orchestration of steps (see [How it works](https://contextgem.shcherbak.ai/how_it_works/index.html.md))