replace .cursorrules with the new cursor mdc files; create a script to generate/update the mdc files from the doc folder
This commit is contained in:
BO WEN
@@ -0,0 +1,146 @@
|
||||
---
|
||||
description: Guidelines for using PocketFlow, Design Pattern, Agent
|
||||
globs:
|
||||
alwaysApply: false
|
||||
---
|
||||
# Agent
|
||||
|
||||
Agent is a powerful design pattern in which nodes can take dynamic actions based on the context.
|
||||
|
||||
|
||||
|
||||
## Implement Agent with Graph
|
||||
|
||||
1. **Context and Action:** Implement nodes that supply context and perform actions.
|
||||
2. **Branching:** Use branching to connect each action node to an agent node. Use action to allow the agent to direct the [flow](mdc:../core_abstraction/flow.md) between nodes—and potentially loop back for multi-step.
|
||||
3. **Agent Node:** Provide a prompt to decide action—for example:
|
||||
|
||||
```python
|
||||
f"""
|
||||
### CONTEXT
|
||||
Task: {task_description}
|
||||
Previous Actions: {previous_actions}
|
||||
Current State: {current_state}
|
||||
|
||||
### ACTION SPACE
|
||||
[1] search
|
||||
Description: Use web search to get results
|
||||
Parameters:
|
||||
- query (str): What to search for
|
||||
|
||||
[2] answer
|
||||
Description: Conclude based on the results
|
||||
Parameters:
|
||||
- result (str): Final answer to provide
|
||||
|
||||
### NEXT ACTION
|
||||
Decide the next action based on the current context and available action space.
|
||||
Return your response in the following format:
|
||||
|
||||
```yaml
|
||||
thinking: |
|
||||
|
||||
action:
|
||||
parameters:
|
||||
:
|
||||
```"""
|
||||
```
|
||||
|
||||
The core of building **high-performance** and **reliable** agents boils down to:
|
||||
|
||||
1. **Context Management:** Provide *relevant, minimal context.* For example, rather than including an entire chat history, retrieve the most relevant via [RAG](mdc:rag.md). Even with larger context windows, LLMs still fall victim to ["lost in the middle"](mdc:https:/arxiv.org/abs/2307.03172), overlooking mid-prompt content.
|
||||
|
||||
2. **Action Space:** Provide *a well-structured and unambiguous* set of actions—avoiding overlap like separate `read_databases` or `read_csvs`. Instead, import CSVs into the database.
|
||||
|
||||
## Example Good Action Design
|
||||
|
||||
- **Incremental:** Feed content in manageable chunks (500 lines or 1 page) instead of all at once.
|
||||
|
||||
- **Overview-zoom-in:** First provide high-level structure (table of contents, summary), then allow drilling into details (raw texts).
|
||||
|
||||
- **Parameterized/Programmable:** Instead of fixed actions, enable parameterized (columns to select) or programmable (SQL queries) actions, for example, to read CSV files.
|
||||
|
||||
- **Backtracking:** Let the agent undo the last step instead of restarting entirely, preserving progress when encountering errors or dead ends.
|
||||
|
||||
## Example: Search Agent
|
||||
|
||||
This agent:
|
||||
1. Decides whether to search or answer
|
||||
2. If searches, loops back to decide if more search needed
|
||||
3. Answers when enough context gathered
|
||||
|
||||
```python
|
||||
class DecideAction(Node):
|
||||
def prep(self, shared):
|
||||
context = shared.get("context", "No previous search")
|
||||
query = shared["query"]
|
||||
return query, context
|
||||
|
||||
def exec(self, inputs):
|
||||
query, context = inputs
|
||||
prompt = f"""
|
||||
Given input: {query}
|
||||
Previous search results: {context}
|
||||
Should I: 1) Search web for more info 2) Answer with current knowledge
|
||||
Output in yaml:
|
||||
```yaml
|
||||
action: search/answer
|
||||
reason: why this action
|
||||
search_term: search phrase if action is search
|
||||
```"""
|
||||
resp = call_llm(prompt)
|
||||
yaml_str = resp.split("```yaml")[1].split("```")[0].strip()
|
||||
result = yaml.safe_load(yaml_str)
|
||||
|
||||
assert isinstance(result, dict)
|
||||
assert "action" in result
|
||||
assert "reason" in result
|
||||
assert result["action"] in ["search", "answer"]
|
||||
if result["action"] == "search":
|
||||
assert "search_term" in result
|
||||
|
||||
return result
|
||||
|
||||
def post(self, shared, prep_res, exec_res):
|
||||
if exec_res["action"] == "search":
|
||||
shared["search_term"] = exec_res["search_term"]
|
||||
return exec_res["action"]
|
||||
|
||||
class SearchWeb(Node):
|
||||
def prep(self, shared):
|
||||
return shared["search_term"]
|
||||
|
||||
def exec(self, search_term):
|
||||
return search_web(search_term)
|
||||
|
||||
def post(self, shared, prep_res, exec_res):
|
||||
prev_searches = shared.get("context", [])
|
||||
shared["context"] = prev_searches + [
|
||||
{"term": shared["search_term"], "result": exec_res}
|
||||
]
|
||||
return "decide"
|
||||
|
||||
class DirectAnswer(Node):
|
||||
def prep(self, shared):
|
||||
return shared["query"], shared.get("context", "")
|
||||
|
||||
def exec(self, inputs):
|
||||
query, context = inputs
|
||||
return call_llm(f"Context: {context}\nAnswer: {query}")
|
||||
|
||||
def post(self, shared, prep_res, exec_res):
|
||||
print(f"Answer: {exec_res}")
|
||||
shared["answer"] = exec_res
|
||||
|
||||
# Connect nodes
|
||||
decide = DecideAction()
|
||||
search = SearchWeb()
|
||||
answer = DirectAnswer()
|
||||
|
||||
decide - "search" >> search
|
||||
decide - "answer" >> answer
|
||||
search - "decide" >> decide # Loop back
|
||||
|
||||
flow = Flow(start=decide)
|
||||
flow.run({"query": "Who won the Nobel Prize in Physics 2024?"})
|
||||
```
|
||||
@@ -0,0 +1,69 @@
|
||||
---
|
||||
description: Guidelines for using PocketFlow, Design Pattern, Map Reduce
|
||||
globs:
|
||||
alwaysApply: false
|
||||
---
|
||||
# Map Reduce
|
||||
|
||||
MapReduce is a design pattern suitable when you have either:
|
||||
- Large input data (e.g., multiple files to process), or
|
||||
- Large output data (e.g., multiple forms to fill)
|
||||
|
||||
and there is a logical way to break the task into smaller, ideally independent parts.
|
||||
|
||||
|
||||
|
||||
You first break down the task using [BatchNode](mdc:../core_abstraction/batch.md) in the map phase, followed by aggregation in the reduce phase.
|
||||
|
||||
### Example: Document Summarization
|
||||
|
||||
```python
|
||||
class SummarizeAllFiles(BatchNode):
|
||||
def prep(self, shared):
|
||||
files_dict = shared["files"] # e.g. 10 files
|
||||
return list(files_dict.items()) # [("file1.txt", "aaa..."), ("file2.txt", "bbb..."), ...]
|
||||
|
||||
def exec(self, one_file):
|
||||
filename, file_content = one_file
|
||||
summary_text = call_llm(f"Summarize the following file:\n{file_content}")
|
||||
return (filename, summary_text)
|
||||
|
||||
def post(self, shared, prep_res, exec_res_list):
|
||||
shared["file_summaries"] = dict(exec_res_list)
|
||||
|
||||
class CombineSummaries(Node):
|
||||
def prep(self, shared):
|
||||
return shared["file_summaries"]
|
||||
|
||||
def exec(self, file_summaries):
|
||||
# format as: "File1: summary\nFile2: summary...\n"
|
||||
text_list = []
|
||||
for fname, summ in file_summaries.items():
|
||||
text_list.append(f"{fname} summary:\n{summ}\n")
|
||||
big_text = "\n---\n".join(text_list)
|
||||
|
||||
return call_llm(f"Combine these file summaries into one final summary:\n{big_text}")
|
||||
|
||||
def post(self, shared, prep_res, final_summary):
|
||||
shared["all_files_summary"] = final_summary
|
||||
|
||||
batch_node = SummarizeAllFiles()
|
||||
combine_node = CombineSummaries()
|
||||
batch_node >> combine_node
|
||||
|
||||
flow = Flow(start=batch_node)
|
||||
|
||||
shared = {
|
||||
"files": {
|
||||
"file1.txt": "Alice was beginning to get very tired of sitting by her sister...",
|
||||
"file2.txt": "Some other interesting text ...",
|
||||
# ...
|
||||
}
|
||||
}
|
||||
flow.run(shared)
|
||||
print("Individual Summaries:", shared["file_summaries"])
|
||||
print("\nFinal Summary:\n", shared["all_files_summary"])
|
||||
```
|
||||
|
||||
> **Performance Tip**: The example above works sequentially. You can speed up the map phase by running it in parallel. See [(Advanced) Parallel](mdc:../core_abstraction/parallel.md) for more details.
|
||||
{: .note }
|
||||
@@ -0,0 +1,184 @@
|
||||
---
|
||||
description: Guidelines for using PocketFlow, Design Pattern, (Advanced) Multi-Agents
|
||||
globs:
|
||||
alwaysApply: false
|
||||
---
|
||||
# (Advanced) Multi-Agents
|
||||
|
||||
Multiple [Agents](mdc:flow.md) can work together by handling subtasks and communicating the progress.
|
||||
Communication between agents is typically implemented using message queues in shared storage.
|
||||
|
||||
> Most of time, you don't need Multi-Agents. Start with a simple solution first.
|
||||
{: .best-practice }
|
||||
|
||||
### Example Agent Communication: Message Queue
|
||||
|
||||
Here's a simple example showing how to implement agent communication using `asyncio.Queue`.
|
||||
The agent listens for messages, processes them, and continues listening:
|
||||
|
||||
```python
|
||||
class AgentNode(AsyncNode):
|
||||
async def prep_async(self, _):
|
||||
message_queue = self.params["messages"]
|
||||
message = await message_queue.get()
|
||||
print(f"Agent received: {message}")
|
||||
return message
|
||||
|
||||
# Create node and flow
|
||||
agent = AgentNode()
|
||||
agent >> agent # connect to self
|
||||
flow = AsyncFlow(start=agent)
|
||||
|
||||
# Create heartbeat sender
|
||||
async def send_system_messages(message_queue):
|
||||
counter = 0
|
||||
messages = [
|
||||
"System status: all systems operational",
|
||||
"Memory usage: normal",
|
||||
"Network connectivity: stable",
|
||||
"Processing load: optimal"
|
||||
]
|
||||
|
||||
while True:
|
||||
message = f"{messages[counter % len(messages)]} | timestamp_{counter}"
|
||||
await message_queue.put(message)
|
||||
counter += 1
|
||||
await asyncio.sleep(1)
|
||||
|
||||
async def main():
|
||||
message_queue = asyncio.Queue()
|
||||
shared = {}
|
||||
flow.set_params({"messages": message_queue})
|
||||
|
||||
# Run both coroutines
|
||||
await asyncio.gather(
|
||||
flow.run_async(shared),
|
||||
send_system_messages(message_queue)
|
||||
)
|
||||
|
||||
asyncio.run(main())
|
||||
```
|
||||
|
||||
The output:
|
||||
|
||||
```
|
||||
Agent received: System status: all systems operational | timestamp_0
|
||||
Agent received: Memory usage: normal | timestamp_1
|
||||
Agent received: Network connectivity: stable | timestamp_2
|
||||
Agent received: Processing load: optimal | timestamp_3
|
||||
```
|
||||
|
||||
### Interactive Multi-Agent Example: Taboo Game
|
||||
|
||||
Here's a more complex example where two agents play the word-guessing game Taboo.
|
||||
One agent provides hints while avoiding forbidden words, and another agent tries to guess the target word:
|
||||
|
||||
```python
|
||||
class AsyncHinter(AsyncNode):
|
||||
async def prep_async(self, shared):
|
||||
guess = await shared["hinter_queue"].get()
|
||||
if guess == "GAME_OVER":
|
||||
return None
|
||||
return shared["target_word"], shared["forbidden_words"], shared.get("past_guesses", [])
|
||||
|
||||
async def exec_async(self, inputs):
|
||||
if inputs is None:
|
||||
return None
|
||||
target, forbidden, past_guesses = inputs
|
||||
prompt = f"Generate hint for '{target}'\nForbidden words: {forbidden}"
|
||||
if past_guesses:
|
||||
prompt += f"\nPrevious wrong guesses: {past_guesses}\nMake hint more specific."
|
||||
prompt += "\nUse at most 5 words."
|
||||
|
||||
hint = call_llm(prompt)
|
||||
print(f"\nHinter: Here's your hint - {hint}")
|
||||
return hint
|
||||
|
||||
async def post_async(self, shared, prep_res, exec_res):
|
||||
if exec_res is None:
|
||||
return "end"
|
||||
await shared["guesser_queue"].put(exec_res)
|
||||
return "continue"
|
||||
|
||||
class AsyncGuesser(AsyncNode):
|
||||
async def prep_async(self, shared):
|
||||
hint = await shared["guesser_queue"].get()
|
||||
return hint, shared.get("past_guesses", [])
|
||||
|
||||
async def exec_async(self, inputs):
|
||||
hint, past_guesses = inputs
|
||||
prompt = f"Given hint: {hint}, past wrong guesses: {past_guesses}, make a new guess. Directly reply a single word:"
|
||||
guess = call_llm(prompt)
|
||||
print(f"Guesser: I guess it's - {guess}")
|
||||
return guess
|
||||
|
||||
async def post_async(self, shared, prep_res, exec_res):
|
||||
if exec_res.lower() == shared["target_word"].lower():
|
||||
print("Game Over - Correct guess!")
|
||||
await shared["hinter_queue"].put("GAME_OVER")
|
||||
return "end"
|
||||
|
||||
if "past_guesses" not in shared:
|
||||
shared["past_guesses"] = []
|
||||
shared["past_guesses"].append(exec_res)
|
||||
|
||||
await shared["hinter_queue"].put(exec_res)
|
||||
return "continue"
|
||||
|
||||
async def main():
|
||||
# Set up game
|
||||
shared = {
|
||||
"target_word": "nostalgia",
|
||||
"forbidden_words": ["memory", "past", "remember", "feeling", "longing"],
|
||||
"hinter_queue": asyncio.Queue(),
|
||||
"guesser_queue": asyncio.Queue()
|
||||
}
|
||||
|
||||
print("Game starting!")
|
||||
print(f"Target word: {shared['target_word']}")
|
||||
print(f"Forbidden words: {shared['forbidden_words']}")
|
||||
|
||||
# Initialize by sending empty guess to hinter
|
||||
await shared["hinter_queue"].put("")
|
||||
|
||||
# Create nodes and flows
|
||||
hinter = AsyncHinter()
|
||||
guesser = AsyncGuesser()
|
||||
|
||||
# Set up flows
|
||||
hinter_flow = AsyncFlow(start=hinter)
|
||||
guesser_flow = AsyncFlow(start=guesser)
|
||||
|
||||
# Connect nodes to themselves
|
||||
hinter - "continue" >> hinter
|
||||
guesser - "continue" >> guesser
|
||||
|
||||
# Run both agents concurrently
|
||||
await asyncio.gather(
|
||||
hinter_flow.run_async(shared),
|
||||
guesser_flow.run_async(shared)
|
||||
)
|
||||
|
||||
asyncio.run(main())
|
||||
```
|
||||
|
||||
The Output:
|
||||
|
||||
```
|
||||
Game starting!
|
||||
Target word: nostalgia
|
||||
Forbidden words: ['memory', 'past', 'remember', 'feeling', 'longing']
|
||||
|
||||
Hinter: Here's your hint - Thinking of childhood summer days
|
||||
Guesser: I guess it's - popsicle
|
||||
|
||||
Hinter: Here's your hint - When childhood cartoons make you emotional
|
||||
Guesser: I guess it's - nostalgic
|
||||
|
||||
Hinter: Here's your hint - When old songs move you
|
||||
Guesser: I guess it's - memories
|
||||
|
||||
Hinter: Here's your hint - That warm emotion about childhood
|
||||
Guesser: I guess it's - nostalgia
|
||||
Game Over - Correct guess!
|
||||
```
|
||||
@@ -0,0 +1,157 @@
|
||||
---
|
||||
description: Guidelines for using PocketFlow, Design Pattern, RAG
|
||||
globs:
|
||||
alwaysApply: false
|
||||
---
|
||||
# RAG (Retrieval Augmented Generation)
|
||||
|
||||
For certain LLM tasks like answering questions, providing relevant context is essential. One common architecture is a **two-stage** RAG pipeline:
|
||||
|
||||
|
||||
|
||||
1. **Offline stage**: Preprocess and index documents ("building the index").
|
||||
2. **Online stage**: Given a question, generate answers by retrieving the most relevant context.
|
||||
|
||||
---
|
||||
## Stage 1: Offline Indexing
|
||||
|
||||
We create three Nodes:
|
||||
1. `ChunkDocs` – [chunks](mdc:../utility_function/chunking.md) raw text.
|
||||
2. `EmbedDocs` – [embeds](mdc:../utility_function/embedding.md) each chunk.
|
||||
3. `StoreIndex` – stores embeddings into a [vector database](mdc:../utility_function/vector.md).
|
||||
|
||||
```python
|
||||
class ChunkDocs(BatchNode):
|
||||
def prep(self, shared):
|
||||
# A list of file paths in shared["files"]. We process each file.
|
||||
return shared["files"]
|
||||
|
||||
def exec(self, filepath):
|
||||
# read file content. In real usage, do error handling.
|
||||
with open(filepath, "r", encoding="utf-8") as f:
|
||||
text = f.read()
|
||||
# chunk by 100 chars each
|
||||
chunks = []
|
||||
size = 100
|
||||
for i in range(0, len(text), size):
|
||||
chunks.append(text[i : i + size])
|
||||
return chunks
|
||||
|
||||
def post(self, shared, prep_res, exec_res_list):
|
||||
# exec_res_list is a list of chunk-lists, one per file.
|
||||
# flatten them all into a single list of chunks.
|
||||
all_chunks = []
|
||||
for chunk_list in exec_res_list:
|
||||
all_chunks.extend(chunk_list)
|
||||
shared["all_chunks"] = all_chunks
|
||||
|
||||
class EmbedDocs(BatchNode):
|
||||
def prep(self, shared):
|
||||
return shared["all_chunks"]
|
||||
|
||||
def exec(self, chunk):
|
||||
return get_embedding(chunk)
|
||||
|
||||
def post(self, shared, prep_res, exec_res_list):
|
||||
# Store the list of embeddings.
|
||||
shared["all_embeds"] = exec_res_list
|
||||
print(f"Total embeddings: {len(exec_res_list)}")
|
||||
|
||||
class StoreIndex(Node):
|
||||
def prep(self, shared):
|
||||
# We'll read all embeds from shared.
|
||||
return shared["all_embeds"]
|
||||
|
||||
def exec(self, all_embeds):
|
||||
# Create a vector index (faiss or other DB in real usage).
|
||||
index = create_index(all_embeds)
|
||||
return index
|
||||
|
||||
def post(self, shared, prep_res, index):
|
||||
shared["index"] = index
|
||||
|
||||
# Wire them in sequence
|
||||
chunk_node = ChunkDocs()
|
||||
embed_node = EmbedDocs()
|
||||
store_node = StoreIndex()
|
||||
|
||||
chunk_node >> embed_node >> store_node
|
||||
|
||||
OfflineFlow = Flow(start=chunk_node)
|
||||
```
|
||||
|
||||
Usage example:
|
||||
|
||||
```python
|
||||
shared = {
|
||||
"files": ["doc1.txt", "doc2.txt"], # any text files
|
||||
}
|
||||
OfflineFlow.run(shared)
|
||||
```
|
||||
|
||||
---
|
||||
## Stage 2: Online Query & Answer
|
||||
|
||||
We have 3 nodes:
|
||||
1. `EmbedQuery` – embeds the user’s question.
|
||||
2. `RetrieveDocs` – retrieves top chunk from the index.
|
||||
3. `GenerateAnswer` – calls the LLM with the question + chunk to produce the final answer.
|
||||
|
||||
```python
|
||||
class EmbedQuery(Node):
|
||||
def prep(self, shared):
|
||||
return shared["question"]
|
||||
|
||||
def exec(self, question):
|
||||
return get_embedding(question)
|
||||
|
||||
def post(self, shared, prep_res, q_emb):
|
||||
shared["q_emb"] = q_emb
|
||||
|
||||
class RetrieveDocs(Node):
|
||||
def prep(self, shared):
|
||||
# We'll need the query embedding, plus the offline index/chunks
|
||||
return shared["q_emb"], shared["index"], shared["all_chunks"]
|
||||
|
||||
def exec(self, inputs):
|
||||
q_emb, index, chunks = inputs
|
||||
I, D = search_index(index, q_emb, top_k=1)
|
||||
best_id = I[0][0]
|
||||
relevant_chunk = chunks[best_id]
|
||||
return relevant_chunk
|
||||
|
||||
def post(self, shared, prep_res, relevant_chunk):
|
||||
shared["retrieved_chunk"] = relevant_chunk
|
||||
print("Retrieved chunk:", relevant_chunk[:60], "...")
|
||||
|
||||
class GenerateAnswer(Node):
|
||||
def prep(self, shared):
|
||||
return shared["question"], shared["retrieved_chunk"]
|
||||
|
||||
def exec(self, inputs):
|
||||
question, chunk = inputs
|
||||
prompt = f"Question: {question}\nContext: {chunk}\nAnswer:"
|
||||
return call_llm(prompt)
|
||||
|
||||
def post(self, shared, prep_res, answer):
|
||||
shared["answer"] = answer
|
||||
print("Answer:", answer)
|
||||
|
||||
embed_qnode = EmbedQuery()
|
||||
retrieve_node = RetrieveDocs()
|
||||
generate_node = GenerateAnswer()
|
||||
|
||||
embed_qnode >> retrieve_node >> generate_node
|
||||
OnlineFlow = Flow(start=embed_qnode)
|
||||
```
|
||||
|
||||
Usage example:
|
||||
|
||||
```python
|
||||
# Suppose we already ran OfflineFlow and have:
|
||||
# shared["all_chunks"], shared["index"], etc.
|
||||
shared["question"] = "Why do people like cats?"
|
||||
|
||||
OnlineFlow.run(shared)
|
||||
# final answer in shared["answer"]
|
||||
```
|
||||
@@ -0,0 +1,112 @@
|
||||
---
|
||||
description: Guidelines for using PocketFlow, Design Pattern, Structured Output
|
||||
globs:
|
||||
alwaysApply: false
|
||||
---
|
||||
# Structured Output
|
||||
|
||||
In many use cases, you may want the LLM to output a specific structure, such as a list or a dictionary with predefined keys.
|
||||
|
||||
There are several approaches to achieve a structured output:
|
||||
- **Prompting** the LLM to strictly return a defined structure.
|
||||
- Using LLMs that natively support **schema enforcement**.
|
||||
- **Post-processing** the LLM's response to extract structured content.
|
||||
|
||||
In practice, **Prompting** is simple and reliable for modern LLMs.
|
||||
|
||||
### Example Use Cases
|
||||
|
||||
- Extracting Key Information
|
||||
|
||||
```yaml
|
||||
product:
|
||||
name: Widget Pro
|
||||
price: 199.99
|
||||
description: |
|
||||
A high-quality widget designed for professionals.
|
||||
Recommended for advanced users.
|
||||
```
|
||||
|
||||
- Summarizing Documents into Bullet Points
|
||||
|
||||
```yaml
|
||||
summary:
|
||||
- This product is easy to use.
|
||||
- It is cost-effective.
|
||||
- Suitable for all skill levels.
|
||||
```
|
||||
|
||||
- Generating Configuration Files
|
||||
|
||||
```yaml
|
||||
server:
|
||||
host: 127.0.0.1
|
||||
port: 8080
|
||||
ssl: true
|
||||
```
|
||||
|
||||
## Prompt Engineering
|
||||
|
||||
When prompting the LLM to produce **structured** output:
|
||||
1. **Wrap** the structure in code fences (e.g., `yaml`).
|
||||
2. **Validate** that all required fields exist (and let `Node` handles retry).
|
||||
|
||||
### Example Text Summarization
|
||||
|
||||
```python
|
||||
class SummarizeNode(Node):
|
||||
def exec(self, prep_res):
|
||||
# Suppose `prep_res` is the text to summarize.
|
||||
prompt = f"""
|
||||
Please summarize the following text as YAML, with exactly 3 bullet points
|
||||
|
||||
{prep_res}
|
||||
|
||||
Now, output:
|
||||
```yaml
|
||||
summary:
|
||||
- bullet 1
|
||||
- bullet 2
|
||||
- bullet 3
|
||||
```"""
|
||||
response = call_llm(prompt)
|
||||
yaml_str = response.split("```yaml")[1].split("```")[0].strip()
|
||||
|
||||
import yaml
|
||||
structured_result = yaml.safe_load(yaml_str)
|
||||
|
||||
assert "summary" in structured_result
|
||||
assert isinstance(structured_result["summary"], list)
|
||||
|
||||
return structured_result
|
||||
```
|
||||
|
||||
> Besides using `assert` statements, another popular way to validate schemas is [Pydantic](mdc:https:/github.com/pydantic/pydantic)
|
||||
{: .note }
|
||||
|
||||
### Why YAML instead of JSON?
|
||||
|
||||
Current LLMs struggle with escaping. YAML is easier with strings since they don't always need quotes.
|
||||
|
||||
**In JSON**
|
||||
|
||||
```json
|
||||
{
|
||||
"dialogue": "Alice said: \"Hello Bob.\\nHow are you?\\nI am good.\""
|
||||
}
|
||||
```
|
||||
|
||||
- Every double quote inside the string must be escaped with `\"`.
|
||||
- Each newline in the dialogue must be represented as `\n`.
|
||||
|
||||
**In YAML**
|
||||
|
||||
```yaml
|
||||
dialogue: |
|
||||
Alice said: "Hello Bob.
|
||||
How are you?
|
||||
I am good."
|
||||
```
|
||||
|
||||
- No need to escape interior quotes—just place the entire text under a block literal (`|`).
|
||||
- Newlines are naturally preserved without needing `\n`.
|
||||
@@ -0,0 +1,49 @@
|
||||
---
|
||||
description: Guidelines for using PocketFlow, Design Pattern, Workflow
|
||||
globs:
|
||||
alwaysApply: false
|
||||
---
|
||||
# Workflow
|
||||
|
||||
Many real-world tasks are too complex for one LLM call. The solution is to **Task Decomposition**: decompose them into a [chain](mdc:../core_abstraction/flow.md) of multiple Nodes.
|
||||
|
||||
|
||||
|
||||
> - You don't want to make each task **too coarse**, because it may be *too complex for one LLM call*.
|
||||
> - You don't want to make each task **too granular**, because then *the LLM call doesn't have enough context* and results are *not consistent across nodes*.
|
||||
>
|
||||
> You usually need multiple *iterations* to find the *sweet spot*. If the task has too many *edge cases*, consider using [Agents](mdc:agent.md).
|
||||
{: .best-practice }
|
||||
|
||||
### Example: Article Writing
|
||||
|
||||
```python
|
||||
class GenerateOutline(Node):
|
||||
def prep(self, shared): return shared["topic"]
|
||||
def exec(self, topic): return call_llm(f"Create a detailed outline for an article about {topic}")
|
||||
def post(self, shared, prep_res, exec_res): shared["outline"] = exec_res
|
||||
|
||||
class WriteSection(Node):
|
||||
def prep(self, shared): return shared["outline"]
|
||||
def exec(self, outline): return call_llm(f"Write content based on this outline: {outline}")
|
||||
def post(self, shared, prep_res, exec_res): shared["draft"] = exec_res
|
||||
|
||||
class ReviewAndRefine(Node):
|
||||
def prep(self, shared): return shared["draft"]
|
||||
def exec(self, draft): return call_llm(f"Review and improve this draft: {draft}")
|
||||
def post(self, shared, prep_res, exec_res): shared["final_article"] = exec_res
|
||||
|
||||
# Connect nodes
|
||||
outline = GenerateOutline()
|
||||
write = WriteSection()
|
||||
review = ReviewAndRefine()
|
||||
|
||||
outline >> write >> review
|
||||
|
||||
# Create and run flow
|
||||
writing_flow = Flow(start=outline)
|
||||
shared = {"topic": "AI Safety"}
|
||||
writing_flow.run(shared)
|
||||
```
|
||||
|
||||
For *dynamic cases*, consider using [Agents](mdc:agent.md).
|
||||
Reference in New Issue
Block a user