update docs

This commit is contained in:
zachary62 2025-01-12 00:17:29 +00:00
parent 1d6f7a3e78
commit 7db8c049e8
5 changed files with 64 additions and 100 deletions

View File

@ -13,30 +13,18 @@ Process large inputs by splitting them into chunks using [BatchNode](./batch.md)
```python ```python
class MapSummaries(BatchNode): class MapSummaries(BatchNode):
def prep(self, shared): def prep(self, shared): return [shared["text"][i:i+10000] for i in range(0, len(shared["text"]), 10000)]
text = shared["text"] def exec(self, chunk): return call_llm(f"Summarize this chunk: {chunk}")
return [text[i:i+10000] for i in range(0, len(text), 10000)] def post(self, shared, prep_res, exec_res_list): shared["summaries"] = exec_res_list
def exec(self, chunk):
return call_llm(f"Summarize this chunk: {chunk}")
def post(self, shared, prep_res, exec_res_list):
shared["summaries"] = exec_res_list
class ReduceSummaries(Node): class ReduceSummaries(Node):
def prep(self, shared): def prep(self, shared): return shared["summaries"]
return shared["summaries"] def exec(self, summaries): return call_llm(f"Combine these summaries: {summaries}")
def post(self, shared, prep_res, exec_res): shared["final_summary"] = exec_res
def exec(self, summaries):
return call_llm(f"Combine these summaries: {summaries}")
def post(self, shared, prep_res, exec_res):
shared["final_summary"] = exec_res
# Connect nodes # Connect nodes
map_node = MapSummaries() map_node = MapSummaries()
reduce_node = ReduceSummaries() reduce_node = ReduceSummaries()
map_node >> reduce_node map_node >> reduce_node
# Create flow # Create flow

View File

@ -43,85 +43,61 @@ flow = Flow(start=chat)
### 2. Improved Memory Management ### 2. Improved Memory Management
We can: We can:
1. Recursively summarize conversations for overview. 1. Limit the chat history to the most recent 4.
2. Use [vector search](./tool.md) to retrieve relevant past exchanges for details 2. Use [vector search](./tool.md) to retrieve relevant exchanges beyond the last 4.
```python ```python
class HandleInput(Node): class ChatWithMemory(Node):
def prep(self, shared): def prep(self, s):
if "history" not in shared: # Initialize shared dict
shared["history"] = [] s.setdefault("history", [])
shared["summary"] = "" s.setdefault("memory_index", None)
shared["memory_index"] = None
shared["memories"] = []
user_input = input("You: ") user_input = input("You: ")
query_embedding = get_embedding(user_input)
relevant_memories = [] # Retrieve relevant past if we have enough history and an index
if shared["memory_index"] is not None: relevant = []
indices, _ = search_index(shared["memory_index"], query_embedding, top_k=2) if len(s["history"]) > 8 and s["memory_index"]:
relevant_memories = [shared["memories"][i[0]] for i in indices] idx, _ = search_index(s["memory_index"], get_embedding(user_input), top_k=2)
relevant = [s["history"][i[0]] for i in idx]
shared["current_input"] = { return {"user_input": user_input, "recent": s["history"][-8:], "relevant": relevant}
"summary": shared["summary"],
"relevant": relevant_memories,
"input": user_input
}
class GenerateResponse(Node): def exec(self, c):
def prep(self, shared): messages = [{"role": "system", "content": "You are a helpful assistant."}]
return shared["current_input"] # Include relevant history if any
if c["relevant"]:
messages.append({"role": "system", "content": f"Relevant: {c['relevant']}"})
# Add recent history and the current user input
messages += c["recent"] + [{"role": "user", "content": c["user_input"]}]
return call_llm(messages)
def exec(self, context): def post(self, s, pre, ans):
prompt = f"""Context: # Update chat history
Summary: {context['summary']} s["history"] += [
Relevant past: {context['relevant']} {"role": "user", "content": pre["user_input"]},
User: {context['input']} {"role": "assistant", "content": ans}
]
Response:""" # When first reaching 8 messages, create index
return call_llm(prompt) if len(s["history"]) == 8:
embeddings = []
for i in range(0, 8, 2):
e = s["history"][i]["content"] + " " + s["history"][i+1]["content"]
embeddings.append(get_embedding(e))
s["memory_index"] = create_index(embeddings)
def post(self, shared, prep_res, exec_res): # Embed older exchanges once we exceed 8 messages
shared["history"].append({"role": "user", "content": prep_res["input"]}) elif len(s["history"]) > 8:
shared["history"].append({"role": "assistant", "content": exec_res}) pair = s["history"][-10:-8]
embedding = get_embedding(pair[0]["content"] + " " + pair[1]["content"])
s["memory_index"].add(np.array([embedding]).astype('float32'))
class UpdateMemory(Node): print(f"Assistant: {ans}")
def prep(self, shared): return "continue"
return shared["current_input"]["input"]
def exec(self, user_input): chat = ChatWithMemory()
return get_embedding(user_input) chat - "continue" >> chat
flow = Flow(start=chat)
def post(self, shared, prep_res, exec_res): flow.run({})
shared["memories"].append(prep_res)
if shared["memory_index"] is None:
shared["memory_index"] = create_index([exec_res])
else:
shared["memory_index"].add(np.array([exec_res]))
class UpdateSummary(Node):
def prep(self, shared):
if shared["history"]:
return shared["history"][-10:]
return None
def exec(self, recent_history):
if recent_history:
return call_llm(f"Summarize this conversation:\n{recent_history}")
return ""
def post(self, shared, prep_res, exec_res):
if exec_res:
shared["summary"] = exec_res
# Connect nodes
input_node = HandleInput()
response_node = GenerateResponse()
memory_node = UpdateMemory()
summary_node = UpdateSummary()
input_node >> response_node >> memory_node >> summary_node >> input_node
chat_flow = Flow(start=input_node)
``` ```

View File

@ -18,7 +18,7 @@ A **Node** is the smallest building block. Each Node has 3 steps:
- The **main execution** step, with optional retries and error handling (below). - The **main execution** step, with optional retries and error handling (below).
- Examples: *primarily for LLMs, but can also for remote APIs*. - Examples: *primarily for LLMs, but can also for remote APIs*.
- ⚠️ If retries enabled, ensure idempotent implementation. - ⚠️ If retries enabled, ensure idempotent implementation.
- ⚠️ This step must not write to the `shared` store. If reading from `shared` is necessary, retrieve and pass it along in `prep()`. - ⚠️ This must **NOT** write to `shared`. If reads are necessary, extract them in `prep()` and pass them in `prep_res`.
- Returns `exec_res`, which is passed to `post()`. - Returns `exec_res`, which is passed to `post()`.
3. `post(shared, prep_res, exec_res)` 3. `post(shared, prep_res, exec_res)`
@ -30,7 +30,7 @@ A **Node** is the smallest building block. Each Node has 3 steps:
{: .note } {: .note }
## Fault Tolerance & Retries ### Fault Tolerance & Retries
Nodes can **retry** execution if `exec()` raises an exception. You control this via two parameters when you create the Node: Nodes can **retry** execution if `exec()` raises an exception. You control this via two parameters when you create the Node:
@ -55,10 +55,9 @@ def exec_fallback(self, shared, prep_res, exc):
raise exc raise exc
``` ```
By default, it just re-raises `exc`. But you can return a fallback result instead. By default, it just re-raises `exc`. But you can return a fallback result instead, which becomes the `exec_res` passed to `post()`.
That fallback result becomes the `exec_res` passed to `post()`.
## Example ### Example: Summarize file
```python ```python
class SummarizeFile(Node): class SummarizeFile(Node):
@ -83,10 +82,10 @@ class SummarizeFile(Node):
# Return "default" by not returning anything # Return "default" by not returning anything
summarize_node = SummarizeFile(max_retries=3) summarize_node = SummarizeFile(max_retries=3)
# Run the node standalone for testing (calls prep->exec->post).
# If exec() fails, it retries up to 3 times before calling exec_fallback().
summarize_node.set_params({"filename": "test_file.txt"}) summarize_node.set_params({"filename": "test_file.txt"})
# node.run() calls prep->exec->post
# If exec() fails, it retries up to 3 times before calling exec_fallback()
action_result = summarize_node.run(shared) action_result = summarize_node.run(shared)
print("Action returned:", action_result) # Usually "default" print("Action returned:", action_result) # Usually "default"

View File

@ -38,7 +38,6 @@ class AnswerQuestion(Node):
# Connect nodes # Connect nodes
prep = PrepareEmbeddings() prep = PrepareEmbeddings()
qa = AnswerQuestion() qa = AnswerQuestion()
prep >> qa prep >> qa
# Create flow # Create flow

View File

@ -11,6 +11,8 @@ Visualizing the nested graph can help understanding. While we **dont** includ
### Example: Visualization of Node with Mermaid ### Example: Visualization of Node with Mermaid
This code recursively traverses the nested graph, assigns unique IDs to each node, and treats Flow nodes as subgraphs to generate Mermaid syntax for a hierarchical visualization.
{% raw %} {% raw %}
```python ```python
def build_mermaid(start): def build_mermaid(start):